You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/08/10 13:53:47 UTC
(cassandra-accord) branch trunk-tmp2 updated (5ea12703 -> 16e1551d)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch trunk-tmp2
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
discard 5ea12703 Minor tracing and progress log improvements: Fix - FetchData tracing parameter was not being propagated - Progress log waiting state could infinite loop without making progress Improve: - Progress log waiting state tracing - Simplify progress log run counters
discard 81d1d8c2 Startup sequence improvements
new 16e1551d Minor tracing and progress log improvements: Fix - FetchData tracing parameter was not being propagated - Progress log waiting state could infinite loop without making progress Improve: - Progress log waiting state tracing - Simplify progress log run counters
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5ea12703)
\
N -- N -- N refs/heads/trunk-tmp2 (16e1551d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
accord-core/src/main/java/accord/api/ConfigurationService.java | 4 +---
.../src/main/java/accord/impl/AbstractConfigurationService.java | 1 -
accord-core/src/main/java/accord/topology/TopologyManager.java | 7 ++++---
3 files changed, 5 insertions(+), 7 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra-accord) 01/01: Minor tracing and progress log improvements: Fix - FetchData tracing parameter was not being propagated - Progress log waiting state could infinite loop without making progress Improve: - Progress log waiting state tracing - Simplify progress log run counters
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk-tmp2
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 16e1551dcfee6b51be2cedb44f5648847695742c
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Sat Aug 9 14:22:03 2025 +0100
Minor tracing and progress log improvements:
Fix
- FetchData tracing parameter was not being propagated
- Progress log waiting state could infinite loop without making progress
Improve:
- Progress log waiting state tracing
- Simplify progress log run counters
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20825
---
accord-core/src/main/java/accord/api/Agent.java | 6 +-
accord-core/src/main/java/accord/api/Tracing.java | 12 +++
.../src/main/java/accord/coordinate/FetchData.java | 27 +++--
.../main/java/accord/coordinate/MaybeRecover.java | 2 -
.../src/main/java/accord/coordinate/Propose.java | 1 -
.../impl/progresslog/DefaultProgressLog.java | 18 +---
.../java/accord/impl/progresslog/HomeState.java | 31 +++---
.../java/accord/impl/progresslog/TxnState.java | 7 +-
.../java/accord/impl/progresslog/WaitingState.java | 112 ++++++++++++++-------
.../src/main/java/accord/local/Bootstrap.java | 1 -
.../src/main/java/accord/local/DepsCalculator.java | 2 -
.../src/main/java/accord/local/DurableBefore.java | 1 -
.../main/java/accord/local/SafeCommandStore.java | 1 -
.../src/main/java/accord/messages/Apply.java | 1 -
.../main/java/accord/messages/InformDecided.java | 1 -
.../src/main/java/accord/messages/Propagate.java | 85 +++++++++-------
.../java/accord/primitives/AbstractRanges.java | 2 -
.../test/java/accord/impl/RemoteListenersTest.java | 2 -
.../src/test/java/accord/impl/TestAgent.java | 6 +-
.../java/accord/impl/basic/TestProgressLogs.java | 1 -
.../src/test/java/accord/impl/list/ListAgent.java | 10 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 8 +-
.../main/java/accord/maelstrom/MaelstromAgent.java | 6 +-
23 files changed, 195 insertions(+), 148 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/Agent.java b/accord-core/src/main/java/accord/api/Agent.java
index 4dae236c..a1b2216e 100644
--- a/accord-core/src/main/java/accord/api/Agent.java
+++ b/accord-core/src/main/java/accord/api/Agent.java
@@ -130,7 +130,7 @@ public interface Agent extends UncaughtExceptionListener
*
* This should aim to prevent two home replicas from attempting to initiate coordination at the same time.
*/
- long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount);
+ long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt);
/**
* This method permits implementations to configure a delay for waiting to attempt to progress the local
@@ -139,7 +139,7 @@ public interface Agent extends UncaughtExceptionListener
* This method should only attempt to minimise wasted work that would anyway be achieved by the transaction's
* coordinator, while ensuring prompt when the coordinator considers the transaction to be durable.
*/
- long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units);
+ long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil blockedUntil, TimeUnit units);
/**
* When a peer is queries for a local state, asynchronous callbacks may be registered.
@@ -147,7 +147,7 @@ public interface Agent extends UncaughtExceptionListener
* This method configures a retry timeout on the node querying its peer to renew any callback registrations
* and re-query the local state.
*/
- long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units);
+ long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil retrying, TimeUnit units);
long retrySyncPointDelay(Node node, int attempt, TimeUnit units);
long retryDurabilityDelay(Node node, int attempt, TimeUnit units);
long expireEpochWait(TimeUnit units);
diff --git a/accord-core/src/main/java/accord/api/Tracing.java b/accord-core/src/main/java/accord/api/Tracing.java
index 718bec59..e6753300 100644
--- a/accord-core/src/main/java/accord/api/Tracing.java
+++ b/accord-core/src/main/java/accord/api/Tracing.java
@@ -18,6 +18,9 @@
package accord.api;
import java.util.Arrays;
+import java.util.function.BiConsumer;
+
+import javax.annotation.Nullable;
import accord.local.CommandStore;
@@ -39,4 +42,13 @@ public interface Tracing
return failure.getClass().getSimpleName() + ":" + failure.getLocalizedMessage()
+ (ste.length > 0 ? " (@" + ste[0].getClassName() + "." + ste[0].getMethodName() + ":" + ste[0].getLineNumber() + ")" : "");
}
+
+ static <V> BiConsumer<V, Throwable> wrap(BiConsumer<V, Throwable> wrap, String context, @Nullable Tracing tracing)
+ {
+ if (tracing == null) return wrap;
+ return (success, fail) -> {
+ if (fail != null) tracing.trace(null, "Failure when %s: %s", context, format(fail));
+ wrap.accept(success, fail);
+ };
+ }
}
diff --git a/accord-core/src/main/java/accord/coordinate/FetchData.java b/accord-core/src/main/java/accord/coordinate/FetchData.java
index 9b3374a5..38cdd83c 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchData.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchData.java
@@ -63,7 +63,7 @@ public class FetchData extends CheckShards<Route<?>>
}
// TODO (expected): separate keys we fetch deps and txns for
- public static class FetchRequest
+ public static class FetchRequest implements BiConsumer<FetchResult, Throwable>
{
final SequentialAsyncExecutor executor;
final Known fetch;
@@ -90,6 +90,13 @@ public class FetchData extends CheckShards<Route<?>>
this.reportTo = reportTo;
this.tracing = tracing;
}
+
+ @Override
+ public void accept(FetchResult success, Throwable fail)
+ {
+ if (fail != null && tracing != null) tracing.trace(null, "Failed fetch: %s", Tracing.format(fail));
+ callback.accept(success, fail);
+ }
}
/**
@@ -112,7 +119,11 @@ public class FetchData extends CheckShards<Route<?>>
{
long srcEpoch = request.srcEpoch;
if (!node.topology().hasAtLeastEpoch(srcEpoch))
- return node.withEpochAtLeast(srcEpoch, request.executor, request.callback, () -> fetchSpecific(node, query, maxRoute, request));
+ {
+ if (request.tracing != null)
+ request.tracing.trace(null, "Epoch %d not ready; waiting", srcEpoch);
+ return node.withEpochAtLeast(srcEpoch, request.executor, request, () -> fetchSpecific(node, query, maxRoute, request));
+ }
return fetchData(node, query, maxRoute, request);
}
@@ -128,15 +139,15 @@ public class FetchData extends CheckShards<Route<?>>
// (i.e. if preaccept/accept contact a later epoch than execution is decided for)
final LatentStoreSelector reportTo;
- private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
+ private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
{
- this(node, target, txnId, invalidIf, route, route.withHomeKey(), maxRoute, sourceEpoch, reportTo, callback);
+ this(node, target, txnId, invalidIf, route, route.withHomeKey(), maxRoute, sourceEpoch, reportTo, callback, tracing);
}
- private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> routeWithHomeKey, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
+ private FetchData(Node node, Known target, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> routeWithHomeKey, Route<?> maxRoute, long sourceEpoch, LatentStoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
{
// TODO (desired, efficiency): restore behaviour of only collecting info if e.g. Committed or Executed
- super(node, node.someSequentialExecutor(), txnId, routeWithHomeKey, sourceEpoch, CheckStatus.IncludeInfo.All, null, invalidIf);
+ super(node, node.someSequentialExecutor(), txnId, routeWithHomeKey, sourceEpoch, CheckStatus.IncludeInfo.All, null, invalidIf, tracing);
this.reportTo = reportTo;
this.maxRoute = maxRoute;
Invariants.requireArgument(routeWithHomeKey.contains(route.homeKey()), "route %s does not contain %s", routeWithHomeKey, route.homeKey());
@@ -147,14 +158,14 @@ public class FetchData extends CheckShards<Route<?>>
private static FetchData fetchData(Node node, Route<?> route, Route<?> maxRoute, FetchRequest req)
{
Invariants.require(!req.contactable.isEmpty());
- FetchData fetch = new FetchData(node, req.fetch, req.txnId, req.invalidIf, route, maxRoute, req.srcEpoch, req.reportTo, req.callback);
+ FetchData fetch = new FetchData(node, req.fetch, req.txnId, req.invalidIf, route, maxRoute, req.srcEpoch, req.reportTo, req.callback, req.tracing);
fetch.start();
return fetch;
}
private static FetchData fetchData(Node node, Known fetch, TxnId txnId, InvalidIf invalidIf, Route<?> route, Route<?> maxRoute, long sourceEpoch, StoreSelector reportTo, BiConsumer<? super FetchResult, Throwable> callback)
{
- FetchData fetchData = new FetchData(node, fetch, txnId, invalidIf, route, maxRoute, sourceEpoch, reportTo, callback);
+ FetchData fetchData = new FetchData(node, fetch, txnId, invalidIf, route, maxRoute, sourceEpoch, reportTo, callback, node.agent().trace(txnId, FETCH));
fetchData.start();
return fetchData;
}
diff --git a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
index cfe1d182..10aa851a 100644
--- a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
+++ b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
@@ -21,9 +21,7 @@ package accord.coordinate;
import java.util.function.BiConsumer;
import accord.api.Tracing;
-import accord.local.CommandStores;
import accord.local.CommandStores.LatentStoreSelector;
-import accord.local.CommandStores.StoreSelector;
import accord.local.SequentialAsyncExecutor;
import accord.messages.InformDurable;
import accord.primitives.*;
diff --git a/accord-core/src/main/java/accord/coordinate/Propose.java b/accord-core/src/main/java/accord/coordinate/Propose.java
index 7cef6754..9391f349 100644
--- a/accord-core/src/main/java/accord/coordinate/Propose.java
+++ b/accord-core/src/main/java/accord/coordinate/Propose.java
@@ -21,7 +21,6 @@ package accord.coordinate;
import java.util.Map;
import java.util.function.BiConsumer;
-import accord.api.CoordinatorEventListener;
import accord.api.ProtocolModifiers.Faults;
import accord.api.RoutingKey;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
index 38af0f2b..4a870064 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -781,18 +781,8 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
Invariants.require(run.pendingTimer() != runKind, "We are actively executing %s, but we also have a pending scheduled task to run this same TxnState later. This should not happen.", runKind);
validatePreRunState(run, runKind);
- if (runKind == Home)
- {
- boolean isRetry = run.homeProgress() != Queued;
- if (isRetry) run.incrementHomeRetryCounter();
- run.runHome(DefaultProgressLog.this, safeStore, safeCommand);
- }
- else
- {
- boolean isRetry = run.waitingProgress() != Queued;
- if (isRetry) run.incrementWaitingRetryCounter();
- run.runWaiting(DefaultProgressLog.this, safeStore, safeCommand);
- }
+ if (runKind == Home) run.runHome(DefaultProgressLog.this, safeStore, safeCommand);
+ else run.runWaiting(DefaultProgressLog.this, safeStore, safeCommand);
}
long nextCallbackId()
@@ -1015,7 +1005,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
@Nonnull
public int waitingRetryCounter()
{
- return current.waitingRetryCounter();
+ return current.waitingRunCounter();
}
@Nonnull
@@ -1032,7 +1022,7 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
public int homeRetryCounter()
{
- return current.homeRetryCounter();
+ return current.homeRunCounter();
}
}
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
index 9179dbaa..0bdaac7f 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -56,11 +56,11 @@ abstract class HomeState extends WaitingState
private static final long PROGRESS_MASK = 0x3;
private static final int STATUS_SHIFT = PROGRESS_SHIFT + 2;
private static final long STATUS_MASK = 0x7;
- private static final int RETRY_COUNTER_SHIFT = STATUS_SHIFT + 3;
- private static final long RETRY_COUNTER_MASK = 0x7;
+ private static final int RUN_COUNTER_SHIFT = STATUS_SHIFT + 3;
+ private static final long RUN_COUNTER_MASK = 0x7;
private static final long SET_MASK = ~((PROGRESS_MASK << PROGRESS_SHIFT)
| (STATUS_MASK << STATUS_SHIFT));
- static final int HOME_STATE_END_SHIFT = RETRY_COUNTER_SHIFT + 3;
+ static final int HOME_STATE_END_SHIFT = RUN_COUNTER_SHIFT + 3;
static
{
@@ -103,23 +103,23 @@ abstract class HomeState extends WaitingState
return Progress.forOrdinal((int) ((encodedState >>> PROGRESS_SHIFT) & PROGRESS_MASK));
}
- final int homeRetryCounter()
+ final int homeRunCounter()
{
- return (int) ((encodedState >>> RETRY_COUNTER_SHIFT) & RETRY_COUNTER_MASK);
+ return (int) ((encodedState >>> RUN_COUNTER_SHIFT) & RUN_COUNTER_MASK);
}
- final void incrementHomeRetryCounter()
+ final void incrementHomeRunCounter()
{
- long shiftedMask = RETRY_COUNTER_MASK << RETRY_COUNTER_SHIFT;
+ long shiftedMask = RUN_COUNTER_MASK << RUN_COUNTER_SHIFT;
long current = encodedState & shiftedMask;
- long updated = Math.min(shiftedMask, current + (1L << RETRY_COUNTER_SHIFT));
+ long updated = Math.min(shiftedMask, current + (1L << RUN_COUNTER_SHIFT));
encodedState &= ~shiftedMask;
encodedState |= updated;
}
- final void clearHomeRetryCounter()
+ final void clearHomeRunCounter()
{
- long shiftedMask = RETRY_COUNTER_MASK << RETRY_COUNTER_SHIFT;
+ long shiftedMask = RUN_COUNTER_MASK << RUN_COUNTER_SHIFT;
encodedState &= ~shiftedMask;
}
@@ -131,13 +131,14 @@ abstract class HomeState extends WaitingState
if (newPhase.compareTo(phase()) > 0)
{
instance.clearPendingAndActive(Home, txnId);
- clearHomeRetryCounter();
+ clearHomeRunCounter();
set(safeStore, instance, newPhase, newProgress);
}
}
final void runHome(DefaultProgressLog instance, SafeCommandStore safeStore, SafeCommand safeCommand)
{
+ incrementHomeRunCounter();
Tracing tracing = instance.node.agent().trace(txnId, HOME_PROGRESS);
Invariants.require(!isHomeDoneOrUninitialised());
Command command = safeCommand.current();
@@ -190,7 +191,7 @@ abstract class HomeState extends WaitingState
if (tracing != null)
{
tracing.trace(safeStore.commandStore(), "Failed to recover: " + Tracing.format(fail));
- tracing.trace(safeStore.commandStore(), "Waiting to retry (%d) with progress token %s", state.homeRetryCounter(), prevProgressToken);
+ tracing.trace(safeStore.commandStore(), "Waiting to retry (%d) with progress token %s", state.homeRunCounter(), prevProgressToken);
}
safeStore.agent().onCaughtException(fail, "Failed recovering " + state);
@@ -198,7 +199,6 @@ abstract class HomeState extends WaitingState
// re-save prior progress token
if (prevProgressToken != null && prevProgressToken.compareTo(command) > 0)
instance.saveProgressToken(command.txnId(), prevProgressToken);
- state.incrementHomeRetryCounter();
state.set(safeStore, instance, status, Queued);
}
else
@@ -216,10 +216,9 @@ abstract class HomeState extends WaitingState
else
{
if (tracing != null)
- tracing.trace(safeStore.commandStore(), "Callback: progress token %s reports not durable; saving token and scheduling retry (%d).", token, state.homeRetryCounter());
+ tracing.trace(safeStore.commandStore(), "Callback: progress token %s reports not durable; saving token and scheduling retry (%d).", token, state.homeRunCounter());
if (prevProgressToken != null && token.compareTo(command) > 0)
instance.saveProgressToken(command.txnId(), token);
- state.incrementHomeRetryCounter();
state.set(safeStore, instance, status, Queued);
}
}
@@ -236,7 +235,7 @@ abstract class HomeState extends WaitingState
void setHomeDone(DefaultProgressLog instance)
{
set(null, instance, Done, NoneExpected);
- clearHomeRetryCounter();
+ clearHomeRunCounter();
instance.clearPendingAndActive(Home, txnId);
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
index 97c0dc02..6a6eb0c9 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
@@ -18,7 +18,6 @@
package accord.impl.progresslog;
-import java.time.temporal.TemporalUnit;
import javax.annotation.Nullable;
import com.google.common.primitives.Ints;
@@ -59,15 +58,15 @@ public final class TxnState extends HomeState implements PreLoadContext
{
default: throw new UnhandledEnum(updated);
case Waiting:
- newDelay = instance.commandStore.agent().slowReplicaDelay(instance.node, safeStore, txnId, 1 + waitingRetryCounter(), blockedUntil, MICROSECONDS);
+ newDelay = instance.commandStore.agent().slowReplicaDelay(instance.node, safeStore, txnId, 1 + waitingRunCounter(), blockedUntil, MICROSECONDS);
break;
case Home:
- newDelay = instance.commandStore.agent().slowCoordinatorDelay(instance.node, safeStore, txnId, MICROSECONDS, 1 + homeRetryCounter());
+ newDelay = instance.commandStore.agent().slowCoordinatorDelay(instance.node, safeStore, txnId, MICROSECONDS, 1 + homeRunCounter());
}
Invariants.require(newDelay > 0);
break;
case Awaiting:
- int retries = updated == TxnStateKind.Home ? homeRetryCounter() : waitingRetryCounter();
+ int retries = updated == TxnStateKind.Home ? homeRunCounter() : waitingRunCounter();
newDelay = instance.commandStore.agent().slowAwaitDelay(instance.node, safeStore, txnId, 1 + retries, blockedUntil, MICROSECONDS);
Invariants.require(newDelay > 0);
break;
diff --git a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
index da23e910..d31846a3 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
@@ -89,8 +89,10 @@ abstract class WaitingState extends BaseTxnState
private static final long BLOCKED_UNTIL_MASK = 0x7;
private static final int HOME_SATISFIES_SHIFT = 5;
private static final long HOME_SATISFIES_MASK = 0x7;
- private static final int AWAIT_SHIFT = 8;
- private static final int AWAIT_BITS = 28;
+ private static final int AWAIT_STARTED_SHIFT = 8;
+ private static final int AWAIT_STARTED_BIT = 1 << AWAIT_STARTED_SHIFT;
+ private static final int AWAIT_SHIFT = AWAIT_STARTED_SHIFT + 1;
+ private static final int AWAIT_BITS = 27;
private static final long AWAIT_MASK = (1L << AWAIT_BITS) - 1;
private static final int AWAIT_EPOCH_SHIFT = AWAIT_SHIFT + AWAIT_BITS;
private static final int AWAIT_EPOCH_BITS = 4;
@@ -173,6 +175,7 @@ abstract class WaitingState extends BaseTxnState
private void clearAwaitState()
{
+ encodedState &= ~AWAIT_STARTED_BIT;
encodedState = clearRoundState(encodedState, AWAIT_SHIFT, AWAIT_MASK);
}
@@ -191,6 +194,16 @@ abstract class WaitingState extends BaseTxnState
encodedState = setBitSet(encodedState, bitSet, roundSize, AWAIT_SHIFT);
}
+ private boolean hasAwaitStarted()
+ {
+ return 0 != (encodedState & AWAIT_STARTED_BIT);
+ }
+
+ private void setAwaitStarted()
+ {
+ encodedState |= AWAIT_STARTED_BIT;
+ }
+
private int awaitRoundIndex(int roundSize)
{
return roundIndex(encodedState, roundSize, AWAIT_SHIFT, AWAIT_MASK);
@@ -208,12 +221,12 @@ abstract class WaitingState extends BaseTxnState
encodedState = setMaxRoundIndexAndClearBitSet(encodedState, roundSize, AWAIT_SHIFT, AWAIT_BITS, AWAIT_MASK);
}
- final int waitingRetryCounter()
+ final int waitingRunCounter()
{
return (int) ((encodedState >>> RETRY_COUNTER_SHIFT) & RETRY_COUNTER_MASK);
}
- final void incrementWaitingRetryCounter()
+ final void incrementWaitingRunCounter()
{
long shiftedMask = RETRY_COUNTER_MASK << RETRY_COUNTER_SHIFT;
long current = encodedState & shiftedMask;
@@ -222,7 +235,7 @@ abstract class WaitingState extends BaseTxnState
encodedState |= updated;
}
- final void clearWaitingRetryCounter()
+ final void clearWaitingRunCounter()
{
long shiftedMask = RETRY_COUNTER_MASK << RETRY_COUNTER_SHIFT;
encodedState &= ~shiftedMask;
@@ -349,7 +362,7 @@ abstract class WaitingState extends BaseTxnState
{
set(null, owner, CanApply, NoneExpected);
owner.clearPendingAndActive(Waiting, txnId);
- clearWaitingRetryCounter();
+ clearWaitingRunCounter();
}
void setBlockedUntil(SafeCommandStore safeStore, DefaultProgressLog owner, BlockedUntil blockedUntil)
@@ -358,7 +371,7 @@ abstract class WaitingState extends BaseTxnState
if (blockedUntil.compareTo(currentlyBlockedUntil) > 0 || isUninitialised())
{
clearAwaitState();
- clearWaitingRetryCounter();
+ clearWaitingRunCounter();
owner.clearPendingAndActive(Waiting, txnId);
set(safeStore, owner, blockedUntil, Queued);
}
@@ -384,6 +397,7 @@ abstract class WaitingState extends BaseTxnState
private void runInternal(SafeCommandStore safeStore, SafeCommand safeCommand, DefaultProgressLog owner, @Nullable Tracing tracing)
{
+ incrementWaitingRunCounter();
BlockedUntil blockedUntil = blockedUntil();
Command command = safeCommand.current();
Invariants.require(!owner.hasPending(Waiting, txnId));
@@ -434,7 +448,6 @@ abstract class WaitingState extends BaseTxnState
// the fetchRoute may also be only the home shard, if that is sufficient to answer our query (e.g. for executeAt)
Route<?> awaitRoute = awaitRoute(slicedRoute, blockedUntil);
Route<?> fetchRoute = fetchRoute(slicedRoute, awaitRoute, blockedUntil, safeStore, lowEpoch, txnId, highEpoch, route);
-
if (awaitRoute.isHomeKeyOnlyRoute())
{
// at this point we can switch to polling as we know someone has the relevant state
@@ -445,8 +458,11 @@ abstract class WaitingState extends BaseTxnState
}
int roundSize = awaitRoundSize(awaitRoute);
- if (hasNewLowEpoch(safeStore, txnId, prevLowEpoch, lowEpoch) || hasNewHighEpoch(safeStore, txnId, prevHighEpoch, highEpoch))
+ if (hasAwaitStarted() && (hasNewLowEpoch(safeStore, txnId, prevLowEpoch, lowEpoch) || hasNewHighEpoch(safeStore, txnId, prevHighEpoch, highEpoch)))
{
+ if (tracing != null)
+ tracing.trace(owner.commandStore, "Epoch bounds changed between invocations from [%d...%d] to [%d...%d]", prevLowEpoch, prevHighEpoch, lowEpoch, highEpoch);
+
// update round counters because we have changed the epochs involved
Route<?> prevSlicedRoute = slicedRoute(safeStore, txnId, route, prevLowEpoch, prevHighEpoch);
Route<?> prevAwaitRoute = awaitRoute(prevSlicedRoute, blockedUntil);
@@ -480,6 +496,7 @@ abstract class WaitingState extends BaseTxnState
// to either periodically fetch the whole remaining route or gradually increase the slice length
if (tracing != null)
tracing.trace(owner.commandStore, "Blocked until %s. Waiting for %s to satisfy; round %d of %d.", blockedUntil, awaitRoute, roundIndex, (awaitRoute.size() + (roundSize - 1))/roundSize);
+ setAwaitStarted();
awaitSlice(owner, blockedUntil, txnId, executeAt, awaitRoute, (roundIndex << 1) | 1);
}
@@ -538,8 +555,8 @@ abstract class WaitingState extends BaseTxnState
tracing.trace(owner.commandStore, "Callback success. Home key ready.");
// the home shard was found to already have the necessary state, with no distributed await;
// we can immediately progress the state machine
- Invariants.require(0 == state.awaitRoundIndex(roundSize));
- Invariants.require(0 == state.awaitBitSet(roundSize));
+ Invariants.expect(0 == state.awaitRoundIndex(roundSize));
+ Invariants.expect(0 == state.awaitBitSet(roundSize));
state.runInternal(safeStore, safeCommand, owner, tracing);
}
else
@@ -552,7 +569,8 @@ abstract class WaitingState extends BaseTxnState
break;
case AwaitSlice:
- Invariants.require(awaitRoute == slicedRoute);
+ Invariants.expect(state.hasAwaitStarted());
+ Invariants.require(awaitRoute.equals(slicedRoute));
// In a production system it is safe for the roundIndex to get corrupted as we will just start polling a bit early,
// but for testing we would like to know it has happened.
if (Invariants.expect(roundStart < slicedRoute.size()))
@@ -562,9 +580,9 @@ abstract class WaitingState extends BaseTxnState
if (notReady.isEmpty())
{
- Invariants.expect((int) awaitRoute.findNextSameKindIntersection(roundStart, (Unseekables) ready, 0) / roundSize == roundIndex);
- // TODO (desired): in this case perhaps upgrade to fetch for next round?
- state.updateAwaitRound(roundIndex + 1, roundSize);
+ Invariants.expect((int) slicedRoute.findNextSameKindIntersection(roundStart, (Unseekables) ready, 0) / roundSize == roundIndex);
+ if (roundStart + roundSize >= slicedRoute.size()) state.setAwaitDone(roundSize);
+ else state.updateAwaitRound(roundIndex + 1, roundSize);
state.runInternal(safeStore, safeCommand, owner, tracing);
}
else
@@ -581,6 +599,8 @@ abstract class WaitingState extends BaseTxnState
case FetchRoute:
if (state.homeSatisfies().compareTo(blockedUntil) < 0)
{
+ if (tracing != null)
+ tracing.trace(owner.commandStore, "Successfully fetched route; invoking runInternal");
state.runInternal(safeStore, safeCommand, owner, tracing);
return;
}
@@ -588,35 +608,57 @@ abstract class WaitingState extends BaseTxnState
case Fetch:
{
Participants<?> notReady = slicedRoute.without(ready);
- Invariants.expect(!notReady.isEmpty(), "Fetch was successful for all keys, but the WaitingState has not been cleared");
- int nextIndex;
- if (roundStart >= awaitRoute.size()) nextIndex = -1;
- else if (slicedRoute == awaitRoute) nextIndex = (int) awaitRoute.findNextSameKindIntersection(roundStart, (Unseekables) notReady, 0);
- else
+ if (!awaitRoute.equals(slicedRoute))
{
- Invariants.require(roundIndex == 0);
- nextIndex = 0;
- }
-
- if (nextIndex < 0)
- {
- // we don't think we have anything to wait for, but we have encountered some notReady responses; queue up a retry
- state.setAwaitDone(roundSize);
- state.retry(safeStore, safeCommand, owner, blockedUntil, tracing);
+ Invariants.expect(awaitRoute.isHomeKeyOnlyRoute());
+ Invariants.expect(state.homeSatisfies().compareTo(blockedUntil) >= 0);
+ // nothing to do, fall through and retry
}
else
{
- Invariants.require(nextIndex >= roundStart);
- roundIndex = nextIndex / roundSize;
- state.updateAwaitRound(roundIndex, roundSize);
- state.initialiseAwaitBitSet(awaitRoute, notReady, roundIndex, roundSize);
- state.runInternal(safeStore, safeCommand, owner, tracing);
+ Invariants.expect(!notReady.isEmpty(), "Fetch was successful for all keys, but the WaitingState has not been cleared");
+ if (roundStart < slicedRoute.size())
+ {
+ int nextIndex;
+ if (slicedRoute.equals(awaitRoute)) nextIndex = (int) slicedRoute.findNextSameKindIntersection(roundStart, (Unseekables) notReady, 0);
+ else
+ {
+ if (Invariants.expect(!state.hasAwaitStarted())) nextIndex = 0;
+ else nextIndex = -1;
+ }
+ state.setAwaitStarted();
+
+ if (nextIndex >= 0)
+ {
+ if (tracing != null)
+ tracing.trace(owner.commandStore, "Found %s notReady %s", notReady);
+
+ Invariants.require(nextIndex >= roundStart);
+ Invariants.require(roundStart < slicedRoute.size());
+ roundIndex = nextIndex / roundSize;
+ Invariants.require(roundIndex * roundSize < slicedRoute.size());
+ state.updateAwaitRound(roundIndex, roundSize);
+ state.initialiseAwaitBitSet(slicedRoute, notReady, roundIndex, roundSize);
+ state.runInternal(safeStore, safeCommand, owner, tracing);
+ return;
+ }
+ }
}
+
+ if (tracing != null)
+ tracing.trace(owner.commandStore, "Found %s notReady, but no more intersections; marking await done and continuing.", notReady);
+
+ // we don't think we have anything to wait for, but we have encountered some notReady responses; queue up a retry
+ state.setAwaitDone(roundSize);
+ state.incrementWaitingRunCounter();
+ state.set(safeStore, owner, blockedUntil, Queued);
}
}
}
else
{
+ if (tracing != null)
+ tracing.trace(owner.commandStore, "Fai");
safeStore.agent().onCaughtException(fail, "Failed fetching data for " + state);
state.retry(safeStore, safeCommand, owner, blockedUntil, tracing);
}
@@ -764,7 +806,7 @@ abstract class WaitingState extends BaseTxnState
{
if (tracing != null)
tracing.trace(owner.commandStore, "Retry queued for later.");
- // queue a retry
+ incrementWaitingRunCounter();
set(safeStore, owner, blockedUntil, Queued);
}
}
diff --git a/accord-core/src/main/java/accord/local/Bootstrap.java b/accord-core/src/main/java/accord/local/Bootstrap.java
index 4adcf210..7ff85b7d 100644
--- a/accord-core/src/main/java/accord/local/Bootstrap.java
+++ b/accord-core/src/main/java/accord/local/Bootstrap.java
@@ -42,7 +42,6 @@ import accord.utils.async.AsyncResults;
import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
import static accord.local.durability.DurabilityService.SyncRemote.MinorityQuorum;
-import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.utils.Invariants.illegalState;
diff --git a/accord-core/src/main/java/accord/local/DepsCalculator.java b/accord-core/src/main/java/accord/local/DepsCalculator.java
index e4edaae8..c7cb4db0 100644
--- a/accord-core/src/main/java/accord/local/DepsCalculator.java
+++ b/accord-core/src/main/java/accord/local/DepsCalculator.java
@@ -26,7 +26,6 @@ import accord.primitives.Deps;
import accord.primitives.EpochSupplier;
import accord.primitives.Participants;
import accord.primitives.RangeDeps;
-import accord.primitives.Status;
import accord.primitives.Status.Durability;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@@ -37,7 +36,6 @@ import accord.utils.Invariants;
import static accord.coordinate.ExecuteFlag.HAS_UNIQUE_HLC;
import static accord.coordinate.ExecuteFlag.READY_TO_EXECUTE;
import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
-import static accord.local.CommandSummaries.SummaryStatus.COMMITTED;
import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
diff --git a/accord-core/src/main/java/accord/local/DurableBefore.java b/accord-core/src/main/java/accord/local/DurableBefore.java
index 4d14c45a..61382732 100644
--- a/accord-core/src/main/java/accord/local/DurableBefore.java
+++ b/accord-core/src/main/java/accord/local/DurableBefore.java
@@ -25,7 +25,6 @@ import javax.annotation.Nullable;
import accord.api.RoutingKey;
import accord.primitives.AbstractRanges;
import accord.primitives.Participants;
-import accord.primitives.Ranges;
import accord.primitives.Status.Durability;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 4900bc13..41f917c7 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -29,7 +29,6 @@ import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.api.RoutingKey;
-import accord.impl.InMemoryCommandStore;
import accord.local.CommandStores.RangesForEpochSupplier;
import accord.local.RedundantBefore.RedundantBeforeSupplier;
import accord.local.cfk.CommandsForKey;
diff --git a/accord-core/src/main/java/accord/messages/Apply.java b/accord-core/src/main/java/accord/messages/Apply.java
index bfc8499a..94447f9d 100644
--- a/accord-core/src/main/java/accord/messages/Apply.java
+++ b/accord-core/src/main/java/accord/messages/Apply.java
@@ -21,7 +21,6 @@ package accord.messages;
import java.util.concurrent.CancellationException;
import javax.annotation.Nullable;
-import accord.api.DataStore;
import accord.api.Result;
import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.CommandStore;
diff --git a/accord-core/src/main/java/accord/messages/InformDecided.java b/accord-core/src/main/java/accord/messages/InformDecided.java
index 6bc6d136..647b5d25 100644
--- a/accord-core/src/main/java/accord/messages/InformDecided.java
+++ b/accord-core/src/main/java/accord/messages/InformDecided.java
@@ -19,7 +19,6 @@ package accord.messages;
import accord.api.RoutingKey;
import accord.local.Node;
-import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.primitives.Route;
import accord.primitives.TxnId;
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index f030f7bb..d0894057 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -101,7 +101,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
@Nullable final Writes writes;
@Nullable final Result result;
final BiConsumer<? super FetchResult, Throwable> callback;
- final @Nullable Tracing trace;
+ final @Nullable Tracing tracing;
private transient volatile FetchResult fetchResult;
private static final AtomicReferenceFieldUpdater<Propagate, FetchResult> fetchResultUpdater = AtomicReferenceFieldUpdater.newUpdater(Propagate.class, FetchResult.class, "fetchResult");
@@ -121,7 +121,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
@Nullable Timestamp committedExecuteAt,
@Nullable Writes writes,
@Nullable Result result,
- BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing trace)
+ BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
{
this.node = node;
this.txnId = txnId;
@@ -142,7 +142,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
this.writes = writes;
this.result = result;
this.callback = callback;
- this.trace = trace;
+ this.tracing = tracing;
}
public static void propagate(Node node, TxnId txnId, InvalidIf previouslyKnownToBeInvalidIf, long sourceEpoch, WithQuorum withQuorum, Route<?> queried, Participants<?> contactable, LatentStoreSelector reportTo, @Nullable Known target, CheckStatusOkFull full, BiConsumer<? super FetchResult, Throwable> callback, @Nullable Tracing tracing)
@@ -199,8 +199,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// ideally integrated with safeStore.get()
if (participants.owns().isEmpty() && safeStore.ifInitialised(txnId) == null)
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Uninitialised and not owned; skipping");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Uninitialised and not owned; skipping");
return null;
}
@@ -218,11 +218,13 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
{
// Already know the outcome, waiting on durability so maybe update with new durability information which can also trigger cleanup
case Persist:
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Already persisted; skipping");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Already persisted; skipping");
return updateDurability(safeStore, safeCommand, participants);
case Cleanup:
case Invalidate:
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Already invalidated/erased; skipping");
return null;
}
@@ -285,8 +287,15 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
SaveStatus propagate = found.atLeast(currentlyKnown).propagatesSaveStatus();
if (propagate.known.isSatisfiedBy(currentlyKnown))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Already know at least as much as peer responses.");
+ if (invalidIf == IfUncommitted)
+ {
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Marking invalidIfUncommitted");
+ safeStore.progressLog().invalidIfUncommitted(txnId);
+ }
+
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Already know at least as much as peer responses.");
updateFetchResult(found, participants.owns());
return updateDurability(safeStore, safeCommand, participants);
}
@@ -303,31 +312,31 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
throw illegalState("Invalid states to propagate: " + propagate);
case Invalidated:
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Invalidating");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Invalidating");
Commands.commitInvalidate(safeStore, safeCommand, route);
break;
case Applied:
case PreApplied:
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Applying");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Applying");
Invariants.require(committedExecuteAt != null);
// we must use the remote executeAt, as it might have a uniqueHlc we aren't aware of at commit
confirm(Commands.apply(safeStore, safeCommand, participants, Ballot.ZERO, txnId, route, committedExecuteAt, stableDeps, partialTxn, writes, result));
break;
case Stable:
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Committing as stable");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Committing as stable");
confirm(Commands.commit(safeStore, safeCommand, participants, Stable, acceptedOrCommitted, txnId, route, partialTxn, executeAtIfKnown, stableDeps, null));
break;
case Committed:
// TODO (expected): we can propagate Committed as Stable if we have any other Stable result AND a quorum of committedDeps
case PreCommitted:
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Pre-committing");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Pre-committing");
confirm(Commands.precommit(safeStore, safeCommand, participants, txnId, executeAtIfKnown, promised));
// TODO (desired): would it be clearer to yield a SaveStatus so we can have PreCommittedWithDefinition
if (!found.definition().isKnown())
@@ -337,16 +346,16 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// only preaccept if we coordinate the transaction
if (safeStore.ranges().coordinates(txnId).intersects(route) && Route.isFullRoute(route))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Pre-accepting");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Pre-accepting");
Commands.preaccept(safeStore, safeCommand, participants, txnId, partialTxn, null, false);
}
case NotDefined:
if (invalidIf == IfUncommitted)
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Marking invalidIfUncommitted");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Marking invalidIfUncommitted");
safeStore.progressLog().invalidIfUncommitted(txnId);
}
break;
@@ -393,8 +402,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
Participants<?> stillTouches = participants.stillTouches();
if (stillTouches.isEmpty())
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "No longer participating (stillTouches is empty); using knownForAny: %s", known.knownForAny());
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "No longer participating (stillTouches is empty); using knownForAny: %s", known.knownForAny());
return known.knownForAny();
}
@@ -402,8 +411,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// try to see if we can safely purge the full command
if (tryPurge(safeStore, safeCommand, status))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Redundant with status %s; purged", status);
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Redundant with status %s; purged", status);
return null;
}
@@ -425,15 +434,15 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
{
Invariants.require(notStaleTouches.containsAll(stillTouches));
Invariants.require(notStaleOwnsOrMayExecute.containsAll(stillOwnsOrMayExecute));
- if (trace != null)
- trace.trace(safeStore.commandStore(), "No longer touches any keys that were found truncated");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "No longer touches any keys that were found truncated");
return required;
}
if (stillOwnsOrMayExecute.isEmpty() && (!found.is(Outcome.Apply) || known.hasFullyTruncated(staleTouches)))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "No longer owns or executes any keys that were found truncated; marking vestigial");
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "No longer owns or executes any keys that were found truncated; marking vestigial");
Commands.setTruncatedOrVestigial(safeStore, safeCommand, participants);
return null;
}
@@ -446,8 +455,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// should solve another way
if (!known.hasFullyTruncated(stale))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Has participants %s that could not be fetched. Some responses were not truncated, so we may have raced with completion. Aborting.", stale);
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Has participants %s that could not be fetched. Some responses were not truncated, so we may have raced with completion. Aborting.", stale);
return null;
}
@@ -455,8 +464,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// wait until we know the shard is ahead and we are behind
if (!safeStore.redundantBefore().isShardOnlyApplied(txnId, stale))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Has participants %s that could not be fetched, but the shard(s) have not been marked universally durable so we will not mark ourselves stale. Aborting.", stale);
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Has participants %s that could not be fetched, but the shard(s) have not been marked universally durable so we will not mark ourselves stale. Aborting.", stale);
return null;
}
@@ -465,8 +474,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
// TODO (expected): if the above last ditch doesn't work, see if only the stale ranges can't apply and do some shenanigans to apply partially and move on
if (ProtocolModifiers.Toggles.markStaleIfCannotExecute(txnId))
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Has participants %s that could not be fetched and the shard(s) have been marked universally durable. We have marked ourselves stale, and will apply the remaining ranges.", stale);
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Has participants %s that could not be fetched and the shard(s) have been marked universally durable. We have marked ourselves stale, and will apply the remaining ranges.", stale);
safeStore.commandStore().markShardStale(safeStore, executeAtIfKnown == null ? txnId : executeAtIfKnown, stale.toRanges(), true);
if (!stale.containsAll(stillTouches) || !stale.containsAll(stillOwnsOrMayExecute))
@@ -474,8 +483,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
}
else
{
- if (trace != null)
- trace.trace(safeStore.commandStore(), "Has participants %s that could not be fetched and the shard(s) have been marked universally durable. This transaction type is configured not to induce staleness, so erasing.", stale);
+ if (tracing != null)
+ tracing.trace(safeStore.commandStore(), "Has participants %s that could not be fetched and the shard(s) have been marked universally durable. This transaction type is configured not to induce staleness, so erasing.", stale);
}
// TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
@@ -516,6 +525,8 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
{
if (null != callback)
callback.accept(failure != null ? null : finaliseFetchResult(), failure);
+ if (failure != null && tracing != null)
+ tracing.trace(null, "Failed propagate: %s", Tracing.format(failure));
}
private static void confirm(Commands.CommitOutcome outcome)
diff --git a/accord-core/src/main/java/accord/primitives/AbstractRanges.java b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
index db97ffb1..7205a5f3 100644
--- a/accord-core/src/main/java/accord/primitives/AbstractRanges.java
+++ b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
@@ -28,8 +28,6 @@ import com.google.common.collect.Iterators;
import accord.api.Key;
import accord.api.RoutingKey;
-import accord.utils.ArrayBuffers;
-import accord.utils.ArrayBuffers.ObjectBufferCache;
import accord.utils.ArrayBuffers.ObjectBuffers;
import accord.utils.IndexedFoldToLong;
import accord.utils.IndexedTriFold;
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index f00ce10b..8472ef79 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -54,7 +54,6 @@ import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
-import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
import accord.primitives.SaveStatus;
import accord.primitives.Status.Durability;
@@ -68,7 +67,6 @@ import accord.utils.AccordGens;
import accord.utils.RandomSource;
import accord.utils.RandomTestRunner;
import accord.utils.async.AsyncChain;
-import accord.utils.async.AsyncChains;
import org.agrona.collections.IntHashSet;
import org.agrona.collections.ObjectHashSet;
diff --git a/accord-core/src/test/java/accord/impl/TestAgent.java b/accord-core/src/test/java/accord/impl/TestAgent.java
index adb1a5a2..c30002e5 100644
--- a/accord-core/src/test/java/accord/impl/TestAgent.java
+++ b/accord-core/src/test/java/accord/impl/TestAgent.java
@@ -153,19 +153,19 @@ public class TestAgent implements Agent
}
@Override
- public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt)
{
return units.convert(1L, SECONDS);
}
@Override
- public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
+ public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
@Override
- public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units)
+ public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, ProgressLog.BlockedUntil retrying, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
diff --git a/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java b/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java
index f03bebc3..9db4a6e7 100644
--- a/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java
+++ b/accord-core/src/test/java/accord/impl/basic/TestProgressLogs.java
@@ -18,7 +18,6 @@
package accord.impl.basic;
-import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import accord.api.ProgressLog;
diff --git a/accord-core/src/test/java/accord/impl/list/ListAgent.java b/accord-core/src/test/java/accord/impl/list/ListAgent.java
index 515d8104..0038a481 100644
--- a/accord-core/src/test/java/accord/impl/list/ListAgent.java
+++ b/accord-core/src/test/java/accord/impl/list/ListAgent.java
@@ -211,22 +211,22 @@ public class ListAgent implements InMemoryAgent, CoordinatorEventListener
}
@Override
- public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt)
{
// TODO (required): meta randomise
- return units.convert(rnd.nextInt(100, 1000) * retryCount, MILLISECONDS);
+ return units.convert(rnd.nextInt(100, 1000) * attempt, MILLISECONDS);
}
@Override
- public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
+ public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil blockedUntil, TimeUnit units)
{
return units.convert(rnd.nextInt(100, 1000), MILLISECONDS);
}
@Override
- public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
+ public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil retrying, TimeUnit units)
{
- int retryDelay = Math.min(16, 1 << retryCount);
+ int retryDelay = Math.min(16, 1 << attempt);
return units.convert(retryDelay, SECONDS);
}
diff --git a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
index 4e7261f2..8f71ce49 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -62,7 +62,6 @@ import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.TimeService;
-import accord.primitives.RangeDeps;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.local.StoreParticipants;
@@ -89,7 +88,6 @@ import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
-import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResults;
import static accord.local.Command.Executed.executed;
@@ -1091,19 +1089,19 @@ public class CommandsForKeyTest
}
@Override
- public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt)
{
return 0;
}
@Override
- public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
+ public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil blockedUntil, TimeUnit units)
{
return 0;
}
@Override
- public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
+ public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil retrying, TimeUnit units)
{
return 0;
}
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
index a59731cd..a536708e 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/MaelstromAgent.java
@@ -152,19 +152,19 @@ public class MaelstromAgent implements Agent, CoordinatorEventListener
}
@Override
- public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt)
{
return units.convert(1L, SECONDS);
}
@Override
- public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
+ public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, ProgressLog.BlockedUntil blockedUntil, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
@Override
- public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units)
+ public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, ProgressLog.BlockedUntil retrying, TimeUnit units)
{
return units.convert(1L, SECONDS);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org