You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by if...@apache.org on 2024/11/19 17:45:17 UTC
(cassandra-accord) 01/05: Investigate mixup failures
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen pushed a commit to branch semi-integrated-burn-test-rebased
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit b0ec1c4ab073600b2c749cbc0c1ad38369547fd5
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Tue Nov 12 13:04:37 2024 +0000
Investigate mixup failures
---
.../src/main/java/accord/api/ProgressLog.java | 1 +
.../accord/coordinate/CoordinateEphemeralRead.java | 2 +-
.../accord/coordinate/CoordinateSyncPoint.java | 2 +-
.../accord/coordinate/CoordinationAdapter.java | 18 +-
.../java/accord/coordinate/FetchMaxConflict.java | 2 +-
.../src/main/java/accord/coordinate/Infer.java | 4 +-
.../src/main/java/accord/coordinate/Persist.java | 15 +-
.../java/accord/coordinate/PersistSyncPoint.java | 4 +-
.../main/java/accord/coordinate/PersistTxn.java | 6 +-
.../java/accord/coordinate/RecoverWithRoute.java | 2 +-
.../coordinate/tracking/FastPathTracker.java | 113 ++++----
.../java/accord/impl/AbstractSafeCommandStore.java | 4 +-
.../java/accord/impl/DefaultLocalListeners.java | 2 +-
.../java/accord/impl/DurabilityScheduling.java | 2 +-
.../java/accord/impl/InMemoryCommandStore.java | 20 +-
.../java/accord/impl/progresslog/BaseTxnState.java | 15 +-
.../impl/progresslog/DefaultProgressLog.java | 4 +
.../impl/progresslog/DefaultProgressLogs.java | 2 +-
.../java/accord/impl/progresslog/HomeState.java | 8 +-
.../java/accord/impl/progresslog/TxnState.java | 5 +
.../java/accord/impl/progresslog/WaitingState.java | 4 +-
.../src/main/java/accord/local/Bootstrap.java | 4 +-
.../src/main/java/accord/local/Cleanup.java | 26 +-
.../src/main/java/accord/local/CommandStore.java | 5 +-
.../src/main/java/accord/local/Commands.java | 13 +-
.../src/main/java/accord/local/PreLoadContext.java | 7 +
.../main/java/accord/local/RedundantBefore.java | 37 ++-
.../main/java/accord/local/RedundantStatus.java | 37 +--
.../main/java/accord/local/SafeCommandStore.java | 24 +-
.../main/java/accord/local/StoreParticipants.java | 3 +-
.../main/java/accord/local/cfk/CommandsForKey.java | 49 ++--
.../src/main/java/accord/local/cfk/NotifySink.java | 2 +-
.../main/java/accord/local/cfk/PostProcess.java | 18 +-
.../src/main/java/accord/local/cfk/Updating.java | 65 ++---
.../src/main/java/accord/messages/Apply.java | 24 +-
.../src/main/java/accord/messages/CheckStatus.java | 8 +-
.../main/java/accord/messages/InformDurable.java | 2 +
.../src/main/java/accord/messages/PreAccept.java | 3 +
.../src/main/java/accord/messages/Propagate.java | 2 +-
.../src/main/java/accord/messages/ReadData.java | 8 +-
.../java/accord/primitives/AbstractRanges.java | 25 +-
.../src/main/java/accord/primitives/KeyRoute.java | 31 +++
.../src/main/java/accord/primitives/Keys.java | 11 +-
.../src/main/java/accord/primitives/Known.java | 6 +-
.../main/java/accord/primitives/LatestDeps.java | 1 -
.../src/main/java/accord/primitives/RangeDeps.java | 2 +-
.../main/java/accord/primitives/RangeRoute.java | 19 ++
.../src/main/java/accord/primitives/Ranges.java | 15 ++
.../src/main/java/accord/primitives/Route.java | 2 +
.../main/java/accord/primitives/RoutingKeys.java | 4 +
.../src/main/java/accord/primitives/Timestamp.java | 2 +-
.../src/main/java/accord/primitives/Txn.java | 12 +-
.../main/java/accord/topology/TopologyManager.java | 2 +-
.../src/main/java/accord/utils/SortedArrays.java | 10 +-
.../java/accord/utils/ThreadPoolScheduler.java | 2 +-
.../src/main/java/accord/utils/Timestamped.java | 5 +
accord-core/src/test/java/accord/KeysTest.java | 58 +++++
.../src/test/java/accord/burn/BurnTest.java | 2 +-
.../tracking/FastPathTrackerReconciler.java | 5 +-
.../coordinate/tracking/ReadTrackerReconciler.java | 1 +
.../coordinate/tracking/ReadTrackerTest.java | 1 +
.../test/java/accord/impl/RemoteListenersTest.java | 2 +-
.../src/test/java/accord/impl/basic/Cluster.java | 283 +++++++++++++++------
.../accord/impl/basic/DelayedCommandStores.java | 43 +---
.../src/test/java/accord/impl/basic/Journal.java | 167 ++++++++----
.../src/test/java/accord/impl/basic/NodeSink.java | 16 +-
.../src/test/java/accord/impl/basic/Packet.java | 7 +
.../src/test/java/accord/impl/basic/Pending.java | 22 ++
.../java/accord/impl/basic/PendingRunnable.java | 18 ++
.../java/accord/impl/basic/RandomDelayQueue.java | 8 +-
.../impl/basic/RecurringPendingRunnable.java | 16 +-
.../accord/impl/basic/TaskExecutorService.java | 6 +
.../src/test/java/accord/impl/list/ListAgent.java | 4 +-
.../accord/impl/list/ListFetchCoordinator.java | 2 +-
.../test/java/accord/impl/list/ListRequest.java | 10 +-
.../src/test/java/accord/impl/list/ListResult.java | 2 +-
.../src/test/java/accord/impl/list/ListStore.java | 90 +++++--
.../test/java/accord/impl/mock/MockCluster.java | 6 +-
.../src/test/java/accord/local/CommandsTest.java | 5 +-
.../java/accord/local/ImmutableCommandTest.java | 26 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 4 +-
.../test/java/accord/messages/PreAcceptTest.java | 2 +-
.../test/java/accord/messages/ReadDataTest.java | 5 +-
.../java/accord/primitives/AbstractRangesTest.java | 6 +
.../test/java/accord/primitives/RangeDepsTest.java | 4 +-
.../java/accord/topology/TopologyRandomizer.java | 3 +-
.../src/test/java/accord/utils/AccordGens.java | 1 -
.../test/java/accord/utils/LogGroupTimersTest.java | 3 +-
.../verify/StrictSerializabilityVerifier.java | 2 +-
89 files changed, 1034 insertions(+), 517 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/ProgressLog.java b/accord-core/src/main/java/accord/api/ProgressLog.java
index 3f1263cb..cb4cf154 100644
--- a/accord-core/src/main/java/accord/api/ProgressLog.java
+++ b/accord-core/src/main/java/accord/api/ProgressLog.java
@@ -26,6 +26,7 @@ import accord.local.Node;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.SaveStatus;
+import accord.primitives.Status;
import accord.primitives.Status.Durability;
import accord.primitives.Participants;
import accord.primitives.Route;
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java b/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
index ab571c70..565b3905 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java
@@ -136,6 +136,6 @@ public class CoordinateEphemeralRead extends AbstractCoordinatePreAccept<Result,
{
Deps deps = Deps.merge(oks, oks.domainSize(), SortedListMap::getValue, ok -> ok.deps);
topologies = node.topology().reselect(topologies, QuorumEpochIntersections.preaccept.include, route, executeAtEpoch, executeAtEpoch, Owned);
- new ExecuteEphemeralRead(node, topologies, route, txnId, txn, executeAtEpoch, deps, this).start();
+ new ExecuteEphemeralRead(node, topologies, route, txnId.withEpoch(executeAtEpoch), txn, executeAtEpoch, deps, this).start();
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java b/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
index 6f271cec..69b4d943 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateSyncPoint.java
@@ -177,6 +177,6 @@ public class CoordinateSyncPoint<U extends Unseekable> extends CoordinatePreAcce
Timestamp executeAt = txnId;
Txn txn = node.agent().emptySystemTxn(txnId.kind(), txnId.domain());
Deps deps = syncPoint.waitFor;
- Apply.sendMaximal(node, to, txnId, syncPoint.route(), txn, executeAt, deps, txn.execute(txnId, executeAt, null), txn.result(txnId, executeAt, null));
+ Apply.sendMaximal(node, to, txnId, syncPoint.route(), txn, executeAt, deps, txn.execute(txnId, executeAt, null), txn.result(txnId, executeAt, null), syncPoint.route());
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java b/accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java
index 3c52d0a2..66d61cba 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinationAdapter.java
@@ -34,6 +34,7 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Participants;
+import accord.primitives.Route;
import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -59,7 +60,7 @@ public interface CoordinationAdapter<R>
void propose(Node node, @Nullable Topologies preaccept, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super R, Throwable> callback);
void stabilise(Node node, @Nullable Topologies any, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super R, Throwable> callback);
void execute(Node node, @Nullable Topologies any, FullRoute<?> route, ExecutePath path, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super R, Throwable> callback);
- void persist(Node node, @Nullable Topologies any, FullRoute<?> route, Participants<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super R, Throwable> callback);
+ void persist(Node node, @Nullable Topologies any, FullRoute<?> route, Route<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super R, Throwable> callback);
default void persist(Node node, @Nullable Topologies any, FullRoute<?> route, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super R, Throwable> callback)
{
persist(node, any, route, route, txnId, txn, executeAt, deps, writes, result, callback);
@@ -159,13 +160,12 @@ public interface CoordinationAdapter<R>
public static final StandardTxnAdapter INSTANCE = new StandardTxnAdapter();
@Override
- public void persist(Node node, Topologies any, FullRoute<?> route, Participants<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
+ public void persist(Node node, Topologies any, FullRoute<?> route, Route<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
{
- // TODO (required): we aren't using sendTo
Topologies all = execution(node, any, sendTo, txnId, executeAt);
if (callback != null) callback.accept(result, null);
- new PersistTxn(node, all, txnId, route, txn, executeAt, deps, writes, result)
+ new PersistTxn(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, route)
.start(Apply.FACTORY, Minimal, all, writes, result);
}
}
@@ -174,12 +174,12 @@ public interface CoordinationAdapter<R>
{
public static final RecoveryTxnAdapter INSTANCE = new RecoveryTxnAdapter();
@Override
- public void persist(Node node, Topologies any, FullRoute<?> route, Participants<?> participants, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
+ public void persist(Node node, Topologies any, FullRoute<?> route, Route<?> sendTo, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super Result, Throwable> callback)
{
- Topologies all = execution(node, any, participants, txnId, executeAt);
+ Topologies all = execution(node, any, sendTo, txnId, executeAt);
if (callback != null) callback.accept(result, null);
- new PersistTxn(node, all, txnId, route, txn, executeAt, deps, writes, result)
+ new PersistTxn(node, all, txnId, sendTo, txn, executeAt, deps, writes, result, route)
.start(Apply.FACTORY, Maximal, all, writes, result);
}
}
@@ -224,7 +224,7 @@ public interface CoordinationAdapter<R>
}
@Override
- public void persist(Node node, Topologies any, FullRoute<?> route, Participants<?> participants, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super SyncPoint<U>, Throwable> callback)
+ public void persist(Node node, Topologies any, FullRoute<?> route, Route<?> participants, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super SyncPoint<U>, Throwable> callback)
{
Topologies all = forExecution(node, route, txnId, executeAt, deps);
@@ -335,7 +335,7 @@ public interface CoordinationAdapter<R>
}
@Override
- public void persist(Node node, Topologies any, FullRoute<?> route, Participants<?> participants, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super SyncPoint<U>, Throwable> callback)
+ public void persist(Node node, Topologies any, FullRoute<?> route, Route<?> participants, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, BiConsumer<? super SyncPoint<U>, Throwable> callback)
{
throw new UnsupportedOperationException();
}
diff --git a/accord-core/src/main/java/accord/coordinate/FetchMaxConflict.java b/accord-core/src/main/java/accord/coordinate/FetchMaxConflict.java
index 9fe651e1..b2038f2a 100644
--- a/accord-core/src/main/java/accord/coordinate/FetchMaxConflict.java
+++ b/accord-core/src/main/java/accord/coordinate/FetchMaxConflict.java
@@ -39,7 +39,7 @@ import static accord.coordinate.tracking.RequestStatus.Success;
* Perform initial rounds of PreAccept and Accept until we have reached agreement about when we should execute.
* If we are preempted by a recovery coordinator, we abort and let them complete (and notify us about the execution result)
*
- * TODO (desired, testing): dedicated burn test to validate outcomes
+ * TODO (expected): we don't need this anymore, we can just use the local MaxConflicts collection when creating a snapshot for streaming
*/
public class FetchMaxConflict extends AbstractCoordinatePreAccept<Timestamp, GetMaxConflictOk>
{
diff --git a/accord-core/src/main/java/accord/coordinate/Infer.java b/accord-core/src/main/java/accord/coordinate/Infer.java
index 787e45a7..81c935f5 100644
--- a/accord-core/src/main/java/accord/coordinate/Infer.java
+++ b/accord-core/src/main/java/accord/coordinate/Infer.java
@@ -41,7 +41,7 @@ import static accord.primitives.Status.PreCommitted;
import static accord.primitives.Route.castToRoute;
import static accord.primitives.Route.isRoute;
-// TODO (required): dedicated randomised testing of all inferences
+// TODO (testing): dedicated randomised testing of all inferences
public class Infer
{
public enum InvalidIf
@@ -73,7 +73,7 @@ public class Infer
public boolean inferInvalidWithQuorum(Known known)
{
- return this == IfUncommitted && !known.isDecided();
+ return this == IfUncommitted && (!known.isDecided() || !known.definition.isOrWasKnown());
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/Persist.java b/accord-core/src/main/java/accord/coordinate/Persist.java
index 042cb228..1082f631 100644
--- a/accord-core/src/main/java/accord/coordinate/Persist.java
+++ b/accord-core/src/main/java/accord/coordinate/Persist.java
@@ -31,6 +31,7 @@ import accord.messages.Callback;
import accord.messages.InformDurable;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
+import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
@@ -46,27 +47,29 @@ public abstract class Persist implements Callback<ApplyReply>
{
protected final Node node;
protected final TxnId txnId;
- protected final FullRoute<?> route;
+ protected final Route<?> sendTo;
protected final Txn txn;
protected final Timestamp executeAt;
protected final Deps stableDeps;
protected final Writes writes;
protected final Result result;
+ protected final FullRoute<?> route;
protected final Topologies topologies;
protected final QuorumTracker tracker;
protected final Set<Id> persistedOn;
boolean isDone;
- protected Persist(Node node, Topologies all, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps stableDeps, Writes writes, Result result)
+ protected Persist(Node node, Topologies all, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps stableDeps, Writes writes, Result result, FullRoute<?> route)
{
this.node = node;
this.txnId = txnId;
- this.route = route;
+ this.sendTo = sendTo;
this.txn = txn;
this.executeAt = executeAt;
this.stableDeps = stableDeps;
this.writes = writes;
this.result = result;
+ this.route = route;
this.topologies = all;
this.tracker = new QuorumTracker(all);
this.persistedOn = new HashSet<>();
@@ -81,7 +84,7 @@ public abstract class Persist implements Callback<ApplyReply>
case Redundant:
case Applied:
persistedOn.add(from);
- if (tracker.recordSuccess(from) == Success)
+ if (sendTo == route && tracker.recordSuccess(from) == Success)
{
if (!isDone)
{
@@ -96,7 +99,7 @@ public abstract class Persist implements Callback<ApplyReply>
}
break;
case Insufficient:
- Apply.sendMaximal(node, from, txnId, route, txn, executeAt, stableDeps, writes, result);
+ Apply.sendMaximal(node, from, txnId, route, txn, executeAt, stableDeps, writes, result, route);
}
}
@@ -124,7 +127,7 @@ public abstract class Persist implements Callback<ApplyReply>
}
else
{
- node.send(contact, to -> factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result), this);
+ node.send(contact, to -> factory.create(kind, to, all, txnId, sendTo, txn, executeAt, stableDeps, writes, result, route), this);
}
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/PersistSyncPoint.java b/accord-core/src/main/java/accord/coordinate/PersistSyncPoint.java
index f0be6f96..2eec31b6 100644
--- a/accord-core/src/main/java/accord/coordinate/PersistSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/PersistSyncPoint.java
@@ -34,7 +34,7 @@ public class PersistSyncPoint extends Persist
{
public PersistSyncPoint(Node node, Topologies topologies, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
{
- super(node, topologies, txnId, route, txn, executeAt, deps, writes, result);
+ super(node, topologies, txnId, route, txn, executeAt, deps, writes, result, route);
}
@Override
@@ -50,7 +50,7 @@ public class PersistSyncPoint extends Persist
{
for (Node.Id to : contact)
{
- Apply apply = factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result);
+ Apply apply = factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result, route);
if (apply == null)
tracker.recordSuccess(to);
else
diff --git a/accord-core/src/main/java/accord/coordinate/PersistTxn.java b/accord-core/src/main/java/accord/coordinate/PersistTxn.java
index b2db464c..c9139f01 100644
--- a/accord-core/src/main/java/accord/coordinate/PersistTxn.java
+++ b/accord-core/src/main/java/accord/coordinate/PersistTxn.java
@@ -22,6 +22,8 @@ import accord.api.Result;
import accord.local.Node;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
+import accord.primitives.Participants;
+import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
@@ -30,8 +32,8 @@ import accord.topology.Topologies;
public class PersistTxn extends Persist
{
- public PersistTxn(Node node, Topologies topologies, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
+ public PersistTxn(Node node, Topologies topologies, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> route)
{
- super(node, topologies, txnId, route, txn, executeAt, deps, writes, result);
+ super(node, topologies, txnId, sendTo, txn, executeAt, deps, writes, result, route);
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/RecoverWithRoute.java b/accord-core/src/main/java/accord/coordinate/RecoverWithRoute.java
index c8fae845..5647a41c 100644
--- a/accord-core/src/main/java/accord/coordinate/RecoverWithRoute.java
+++ b/accord-core/src/main/java/accord/coordinate/RecoverWithRoute.java
@@ -177,7 +177,7 @@ public class RecoverWithRoute extends CheckShards<FullRoute<?>>
{
// we might have only part of the full transaction, and a shard may have truncated;
// in this case we want to skip straight to apply, but only for the shards that haven't truncated
- Participants<?> sendTo = route.participants().without(full.truncatedResponse());
+ Route<?> sendTo = route.without(full.truncatedResponse());
if (!sendTo.isEmpty())
{
known = full.knownFor(txnId, sendTo, sendTo);
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
index d42e198a..d02d8bb6 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
@@ -22,6 +22,8 @@ import accord.coordinate.tracking.QuorumTracker.QuorumShardTracker;
import accord.local.Node;
import accord.topology.Shard;
import accord.topology.Topologies;
+import accord.utils.Invariants;
+
import com.google.common.annotations.VisibleForTesting;
import javax.annotation.Nonnull;
@@ -43,6 +45,7 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
protected int fastPathAccepts;
protected int fastPathFailures;
protected int fastPathDelayed;
+ protected boolean complete;
public FastPathShardTracker(Shard shard)
{
@@ -51,19 +54,43 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
// return NewQuorumSuccess ONLY once fast path is rejected
public abstract ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node);
- public abstract ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node);
+ final ShardOutcome<? super FastPathTracker> complete(ShardOutcome<? super FastPathTracker> result)
+ {
+ Invariants.checkState(!complete);
+ complete = true;
+ return result;
+ }
+
+ public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
+ {
+ if (complete)
+ return NoChange;
+
+ ++successes;
+ if (shard.fastPathElectorate.contains(node))
+ {
+ ++fastPathAccepts;
+ if (hasMetFastPathCriteria())
+ return complete(NewFastPathSuccess);
+ }
+
+ return quorumIfHasRejectedFastPath();
+ }
public final ShardOutcome<? super FastPathTracker> onFailure(@Nonnull Node.Id from)
{
+ if (complete)
+ return NoChange;
+
if (++failures > shard.maxFailures)
- return Fail;
+ return complete(Fail);
if (shard.fastPathElectorate.contains(from))
{
++fastPathFailures;
- if (isNewFastPathReject() && hasReachedQuorum())
- return Success;
+ if (hasRejectedFastPath() && hasReachedQuorum())
+ return complete(Success);
}
return NoChange;
@@ -71,40 +98,28 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
public final ShardOutcome<? super FastPathTracker> onDelayed(@Nonnull Node.Id from)
{
+ if (complete)
+ return NoChange;
+
if (shard.fastPathElectorate.contains(from))
{
++fastPathDelayed;
- if (isNewFastPathDelayed() && hasReachedQuorum())
- return Success;
+ if (isFastPathDelayed() && hasReachedQuorum())
+ return complete(Success);
}
return NoChange;
}
- final ShardOutcome<? super FastPathTracker> quorumIfNewAndHasRejectedFastPath()
- {
- return isNewSlowPathSuccess() && hasRejectedFastPath() ? Success : NoChange;
- }
-
- final boolean isNewSlowPathSuccess()
- {
- return successes == shard.slowPathQuorumSize;
- }
-
- final boolean isNewFastPathReject()
+ final ShardOutcome<? super FastPathTracker> quorumIfHasRejectedFastPath()
{
- return fastPathFailures == shard.fastPathRejectSize && fastPathDelayed < shard.fastPathRejectSize;
+ return hasReachedQuorum() && hasRejectedFastPath() ? complete(Success) : NoChange;
}
- final boolean isNewFastPathDelayed()
+ final boolean isFastPathDelayed()
{
- return fastPathDelayed == shard.fastPathRejectSize && fastPathFailures < shard.fastPathRejectSize;
- }
-
- final boolean isNewFastPathSuccess()
- {
- return fastPathAccepts == shard.fastPathQuorumSize && (fastPathDelayed < shard.fastPathRejectSize || isNewSlowPathSuccess());
+ return fastPathDelayed >= shard.fastPathRejectSize;
}
@VisibleForTesting
@@ -116,7 +131,7 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
@VisibleForTesting
public final boolean hasRejectedFastPath()
{
- return Math.max(fastPathFailures, fastPathDelayed) >= shard.fastPathRejectSize;
+ return fastPathFailures >= shard.fastPathRejectSize;
}
}
@@ -129,24 +144,7 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
public ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node)
{
- ++successes;
- if (!shard.fastPathElectorate.contains(node))
- return quorumIfNewAndHasRejectedFastPath();
-
- ++fastPathAccepts;
- if (isNewFastPathSuccess())
- return NewFastPathSuccess;
-
- if (isNewSlowPathSuccess() && hasRejectedFastPath())
- return Success;
-
- return NoChange;
- }
-
- @Override
- public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
- {
- return onQuorumSuccess(node);
+ return onMaybeFastPathSuccess(node);
}
}
@@ -159,31 +157,14 @@ public class FastPathTracker extends PreAcceptTracker<FastPathTracker.FastPathSh
public ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node)
{
- ++successes;
- if (!shard.fastPathElectorate.contains(node))
- return quorumIfNewAndHasRejectedFastPath();
-
- ++fastPathFailures;
- if (isNewFastPathReject() && hasReachedQuorum())
- return Success;
+ if (complete)
+ return NoChange;
- if (isNewSlowPathSuccess() && hasRejectedFastPath())
- return Success;
-
- return NoChange;
- }
-
- @Override
- public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
- {
++successes;
if (shard.fastPathElectorate.contains(node))
- {
- ++fastPathAccepts;
- if (isNewFastPathSuccess())
- return NewFastPathSuccess;
- }
- return quorumIfNewAndHasRejectedFastPath();
+ ++fastPathFailures;
+
+ return quorumIfHasRejectedFastPath();
}
}
diff --git a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
index 8a740873..ab2ed46c 100644
--- a/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/AbstractSafeCommandStore.java
@@ -67,10 +67,10 @@ public abstract class AbstractSafeCommandStore<CommandType extends SafeCommand,
protected abstract CommandsForKeyType getCommandsForKeyIfUnsafe(RoutingKey key);
@Override
- protected CommandType ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
+ protected CommandType ifLoadedInternal(TxnId txnId)
{
CommandType command = getIfLoadedUnsafe(txnId, this::getCommandUnsafe, this::addCommandUnsafe, this::getIfLoadedUnsafe);
- if (command == null || command.isUnset())
+ if (command == null)
return null;
return command;
}
diff --git a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
index daa5593c..52fba403 100644
--- a/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
+++ b/accord-core/src/main/java/accord/impl/DefaultLocalListeners.java
@@ -84,7 +84,7 @@ public class DefaultLocalListeners implements LocalListeners
@Override
public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listenerId)
{
- SafeCommand listener = safeStore.ifLoadedAndInitialisedAndNotErased(listenerId);
+ SafeCommand listener = safeStore.ifLoadedAndInitialised(listenerId);
if (listener != null) Commands.listenerUpdate(safeStore, listener, safeCommand);
else
{
diff --git a/accord-core/src/main/java/accord/impl/DurabilityScheduling.java b/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
index 49e57cb1..bd370b93 100644
--- a/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
+++ b/accord-core/src/main/java/accord/impl/DurabilityScheduling.java
@@ -306,7 +306,7 @@ public class DurabilityScheduling implements ConfigurationService.Listener
{
synchronized (ShardScheduler.this)
{
- // TODO (required): try to recover or invalidate prior sync point
+ // TODO (expected): try to recover or invalidate prior sync point
retry();
if (numberOfSplits * 2 <= maxNumberOfSplits)
{
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 69e547c5..e50998b4 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -195,7 +195,7 @@ public abstract class InMemoryCommandStore extends CommandStore
long epoch = cur.executeAt().epoch();
Ranges ranges = safeStore.ranges().allAt(epoch);
Participants<?> participants = cur.route().participants(ranges, Minimal);
- // TODO (required): look forwards also, but we only need to look at ?=ReadyToExecute transactions as they have already run their backwards checks
+ // TODO (testing): look forwards also, but we only need to look at ?=ReadyToExecute transactions as they have already run their backwards checks
Iterator<GlobalCommand> iter = commandsByExecuteAt.descendingMap().tailMap(cur.executeAt(), false).values().iterator();
while (iter.hasNext())
{
@@ -391,16 +391,18 @@ public abstract class InMemoryCommandStore extends CommandStore
List<TxnId> resubmit = new ArrayList<>();
for (TxnId txnId : txnIds)
{
- Command command = commands.get(txnId).value();
- if (!command.hasBeen(PreCommitted)) return;
- if (!command.txnId().isVisible()) return;
+ GlobalCommand globalCommand = commands.get(txnId);
+ if (globalCommand == null) continue;
+ Command command = globalCommand.value();
+ if (!command.hasBeen(PreCommitted)) continue;
+ if (!command.txnId().isVisible()) continue;
Ranges allRanges = unsafeRangesForEpoch().allBetween(txnId.epoch(), command.executeAtOrTxnId().epoch());
boolean done = command.hasBeen(Truncated);
if (!done)
{
if (unsafeGetRedundantBefore().status(txnId, command.route()) == RedundantStatus.PRE_BOOTSTRAP_OR_STALE)
- return;
+ continue;
Route<?> route = command.route().slice(allRanges);
done = !route.isEmpty() && ranges.containsAll(route);
@@ -423,7 +425,7 @@ public abstract class InMemoryCommandStore extends CommandStore
if (!resubmit.isEmpty())
{
txnIds = resubmit;
- node.scheduler().once(this, 5L, TimeUnit.SECONDS);
+ node.scheduler().selfRecurring(this, 5L, TimeUnit.SECONDS);
}
}
}
@@ -1194,7 +1196,7 @@ public abstract class InMemoryCommandStore extends CommandStore
@Override
public void shutdown()
{
- executor.shutdown();
+ executor.shutdownNow();
}
}
@@ -1213,10 +1215,10 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- public InMemorySafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
+ public InMemorySafeCommand ifLoadedInternal(TxnId txnId)
{
assertThread();
- return super.ifLoadedAndInitialisedAndNotErasedInternal(txnId);
+ return super.ifLoadedInternal(txnId);
}
@Override
diff --git a/accord-core/src/main/java/accord/impl/progresslog/BaseTxnState.java b/accord-core/src/main/java/accord/impl/progresslog/BaseTxnState.java
index 2c20c6e6..140d5015 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/BaseTxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/BaseTxnState.java
@@ -39,11 +39,11 @@ import static accord.impl.progresslog.TxnStateKind.Waiting;
* - WaitingState defines the methods for updating the WaitingState state machine (i.e. for local commands that are waiting on a dependency's outcome)
* - HomeState defines the methods for updating the HomeState state machine (i.e. for coordinators ensuring a transaction completes)
* <p>
- * TODO (required): unit test all bit fiddling methods
+ * TODO (testing): unit test all bit fiddling methods
*/
abstract class BaseTxnState extends LogGroupTimers.Timer implements Comparable<BaseTxnState>
{
- private static final int PENDING_TIMER_SHIFT = 53;
+ private static final int PENDING_TIMER_SHIFT = HomeState.HOME_STATE_END_SHIFT;
private static final int PENDING_TIMER_BITS = 9;
private static final int PENDING_TIMER_LOW_BITS = PackedLogLinearInteger.validateLowBits(4, PENDING_TIMER_BITS);
private static final long PENDING_TIMER_MASK = (1L << PENDING_TIMER_BITS) - 1;
@@ -51,22 +51,27 @@ abstract class BaseTxnState extends LogGroupTimers.Timer implements Comparable<B
private static final int SCHEDULED_TIMER_SHIFT = 62;
private static final int CONTACT_ALL_SHIFT = 63;
+ static
+ {
+ Invariants.checkState(PENDING_TIMER_SHIFT + PENDING_TIMER_BITS <= SCHEDULED_TIMER_SHIFT);
+ }
+
public final TxnId txnId;
/**
- * bits [0..40) encode WaitingState
+ * bits [0..43) encode WaitingState
* 2 bits for Progress
* 3 bits for BlockedUntil target
* 3 bits for BlockedUntil that home shard can satisfy
* 32 bits for remote progress key counter [note: if we need to in future we can safely and easily reclaim bits here]
* 3 bits for retry counter
* <p>
- * bits [40..45) encode HomeState
+ * bits [43..51) encode HomeState
* 2 bits for Progress
* 3 bits for CoordinatePhase
* 3 bits for retry counter
* <p>
- * bits [53..62) for pending timer delay
+ * bits [51..60) for pending timer delay
* bit 62 for which kind of timer is scheduled
* bit 63 for whether we should contact all candidate replicas (rather than just our preferred group)
*/
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 41c7c7e7..1d75196e 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -459,6 +459,10 @@ public class DefaultProgressLog implements ProgressLog, Runnable
if (safeCommand == null)
return;
+ // check this after fetching SafeCommand, as doing so can erase the command (and invalidate our state)
+ if (run.isDone(runKind))
+ return;
+
Invariants.checkState(get(run.txnId) == run, "Transaction state for %s does not match expected one %s", run.txnId, run);
Invariants.checkState(run.scheduledTimer() != runKind, "We are actively executing %s, but we are also scheduled to run this same TxnState later. This should not happen.", runKind);
Invariants.checkState(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);
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
index 7f4fceaa..8ca59d6c 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLogs.java
@@ -28,7 +28,7 @@ import accord.local.Node;
// TODO (desired, consider): consider propagating invalidations in the same way as we do applied
// TODO (expected): report transactions not making progress
-// TODO (required): evict to disk
+// TODO (expected): evict to disk
public class DefaultProgressLogs implements ProgressLog.Factory
{
static ConcurrentHashMap<DefaultProgressLog, Runnable> PAUSE_FOR_TEST;
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 d8378f53..f8302f6d 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/HomeState.java
@@ -41,7 +41,7 @@ import static accord.impl.progresslog.Progress.Queued;
import static accord.impl.progresslog.TxnStateKind.Home;
/**
- * TODO (required): describe state machine
+ * TODO (documentation): describe state machine
*
* TODO (expected): do not attempt recovery every run; simply check the coordinator is still active
* TODO (expected): do not attempt execution until all shards are ready; use the WaitingState to achieve this
@@ -52,11 +52,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 = PROGRESS_SHIFT + 3;
+ private static final int RETRY_COUNTER_SHIFT = STATUS_SHIFT + 3;
private static final long RETRY_COUNTER_MASK = 0x7;
private static final long SET_MASK = ~((PROGRESS_MASK << PROGRESS_SHIFT)
| (STATUS_MASK << STATUS_SHIFT));
- final int HOME_STATE_END_SHIFT = RETRY_COUNTER_SHIFT + 3;
+ static final int HOME_STATE_END_SHIFT = RETRY_COUNTER_SHIFT + 3;
HomeState(TxnId txnId)
{
@@ -141,7 +141,7 @@ abstract class HomeState extends WaitingState
Invariants.checkState(command.durability() != null);
// TODO (expected): when invalidated, safer to maintain HomeState until known to be globally invalidated
- // TODO (now): validate that we clear HomeState when we receive a Durable reply, to replace the token check logic
+ // TODO (expected): validate that we clear HomeState when we receive a Durable reply, to replace the token check logic
Invariants.checkState(!command.durability().isDurableOrInvalidated(), "Command is durable or invalidated, but we have not cleared the ProgressLog");
ProgressToken maxProgressToken = instance.savedProgressToken(txnId).merge(command);
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 c189f6a2..e5441995 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
@@ -154,4 +154,9 @@ final class TxnState extends HomeState
{
return txnId + ": " + toStateString();
}
+
+ public boolean isDone(TxnStateKind runKind)
+ {
+ return runKind == TxnStateKind.Home ? isHomeDone() : isWaitingDone();
+ }
}
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 87c90ae3..35e80b8a 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
@@ -87,13 +87,13 @@ 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_BITS = 32;
private static final int AWAIT_SHIFT = 8;
+ private static final int AWAIT_BITS = 32;
private static final long AWAIT_MASK = (1L << AWAIT_BITS) - 1;
private static final long SET_MASK = ~((PROGRESS_MASK << PROGRESS_SHIFT) | (BLOCKED_UNTIL_MASK << BLOCKED_UNTIL_SHIFT));
private static final long INITIALISED_MASK = (PROGRESS_MASK << PROGRESS_SHIFT) | (BLOCKED_UNTIL_MASK << BLOCKED_UNTIL_SHIFT) | (HOME_SATISFIES_MASK << HOME_SATISFIES_SHIFT);
- private static final int RETRY_COUNTER_SHIFT = 40;
+ private static final int RETRY_COUNTER_SHIFT = AWAIT_SHIFT + AWAIT_BITS;
private static final long RETRY_COUNTER_MASK = 0x7;
static final int WAITING_STATE_END_SHIFT = RETRY_COUNTER_SHIFT + 3;
diff --git a/accord-core/src/main/java/accord/local/Bootstrap.java b/accord-core/src/main/java/accord/local/Bootstrap.java
index 2c9799e8..459952aa 100644
--- a/accord-core/src/main/java/accord/local/Bootstrap.java
+++ b/accord-core/src/main/java/accord/local/Bootstrap.java
@@ -44,7 +44,6 @@ import accord.utils.async.AsyncResults;
import static accord.local.PreLoadContext.empty;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-import static accord.primitives.Txn.Kind.LocalOnly;
import static accord.utils.Invariants.illegalState;
/**
@@ -109,7 +108,7 @@ class Bootstrap
boolean fetchCompleted;
boolean completed; // we have finished fetching all the data we are able to, but we may still have in-flight fetches
Throwable fetchOutcome;
- TxnId globalSyncId, localSyncId;
+ TxnId globalSyncId;
Attempt(Ranges ranges)
{
@@ -125,7 +124,6 @@ class Bootstrap
}
globalSyncId = node.nextTxnId(ExclusiveSyncPoint, Routable.Domain.Range);
- localSyncId = globalSyncId.as(LocalOnly).withEpoch(epoch);
Invariants.checkArgument(epoch <= globalSyncId.epoch(), "Attempting to use local epoch %d which is larger than global epoch %d", epoch, globalSyncId.epoch());
if (!node.topology().hasEpoch(globalSyncId.epoch()))
diff --git a/accord-core/src/main/java/accord/local/Cleanup.java b/accord-core/src/main/java/accord/local/Cleanup.java
index 195f9fd6..78a4dea2 100644
--- a/accord-core/src/main/java/accord/local/Cleanup.java
+++ b/accord-core/src/main/java/accord/local/Cleanup.java
@@ -33,6 +33,7 @@ import static accord.local.RedundantBefore.PreBootstrapOrStale.FULLY;
import static accord.local.RedundantStatus.LIVE;
import static accord.local.RedundantStatus.NOT_OWNED;
import static accord.local.RedundantStatus.SHARD_REDUNDANT;
+import static accord.local.RedundantStatus.WAS_OWNED_CLOSED;
import static accord.primitives.SaveStatus.Erased;
import static accord.primitives.SaveStatus.ErasedOrVestigial;
import static accord.primitives.SaveStatus.Invalidated;
@@ -111,7 +112,7 @@ public enum Cleanup
private static Cleanup shouldCleanupInternal(Agent agent, TxnId txnId, SaveStatus saveStatus, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
if (txnId.kind() == EphemeralRead)
- return NO; // TODO (required): clean-up based on timeout
+ return NO;
if (expunge(txnId, saveStatus, durableBefore, redundantBefore))
return EXPUNGE;
@@ -199,17 +200,23 @@ public enum Cleanup
case REDUNDANT_PRE_BOOTSTRAP_OR_STALE:
case LOCALLY_REDUNDANT:
return Cleanup.NO;
+
case SHARD_REDUNDANT:
- if (!isPartial && saveStatus.hasBeen(PreCommitted) && !saveStatus.hasBeen(Applied) && redundantBefore.preBootstrapOrStale(txnId, participants.owns) != FULLY)
+ if (!isPartial)
{
- agent.onViolation(String.format("Loading SHARD_REDUNDANT command %s with status %s (that should have been Applied). Expected to be witnessed and executed by %s.", txnId, saveStatus, redundantBefore.max(participants.route, e -> e.shardAppliedOrInvalidatedBefore)));
- return TRUNCATE;
+ if (!saveStatus.hasBeen(PreCommitted))
+ return INVALIDATE;
+
+ if (!saveStatus.hasBeen(Applied) && redundantBefore.preBootstrapOrStale(txnId, participants.owns) != FULLY)
+ {
+ agent.onViolation(String.format("Loading SHARD_REDUNDANT command %s with status %s (that should have been Applied). Expected to be witnessed and executed by %s.", txnId, saveStatus, redundantBefore.max(participants.route, e -> e.shardAppliedOrInvalidatedBefore)));
+ return TRUNCATE;
+ }
}
- case WAS_OWNED:
- }
+ break;
- if (!isPartial && !saveStatus.hasBeen(PreCommitted))
- return INVALIDATE;
+ case WAS_OWNED_CLOSED:
+ }
Durability min = durableBefore.min(txnId, participants.route);
switch (min)
@@ -219,6 +226,9 @@ public enum Cleanup
throw new AssertionError("Unexpected durability: " + min);
case NotDurable:
+ if (redundant == WAS_OWNED_CLOSED)
+ return NO;
+
if (durability == Majority)
return Cleanup.TRUNCATE;
return Cleanup.TRUNCATE_WITH_OUTCOME;
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index bcd175cb..6fb4ae04 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -165,6 +165,7 @@ public abstract class CommandStore implements AgentExecutor
* But they may still be ordered for other key ranges they participate in.
*
* TODO (expected): merge with redundantBefore
+ * TODO (required): if we ever actually erase data on disk (after losing ownership), make sure to mark the range unsafe to read
*/
private NavigableMap<Timestamp, Ranges> safeToRead = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY);
private final Set<Bootstrap> bootstraps = Collections.synchronizedSet(new DeterministicIdentitySet<>());
@@ -592,7 +593,7 @@ public abstract class CommandStore implements AgentExecutor
e.getValue().ranges = remaining = remaining.without(ranges);
if (e.getValue().ranges.isEmpty())
{
- logger.info("Completed full sync for {} on epoch {} using {}", e.getValue().allRanges, e.getKey(), syncId);
+ logger.debug("Completed full sync for {} on epoch {} using {}", e.getValue().allRanges, e.getKey(), syncId);
e.getValue().whenDone.trySuccess(null);
if (remove == null)
remove = new LongHashSet();
@@ -600,7 +601,7 @@ public abstract class CommandStore implements AgentExecutor
}
else
{
- logger.info("Completed partial sync for {} on epoch {} using {}; {} still to sync", synced, e.getKey(), syncId, remaining);
+ logger.debug("Completed partial sync for {} on epoch {} using {}; {} still to sync", synced, e.getKey(), syncId, remaining);
}
}
if (remove != null)
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 3974d86d..39866a12 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -573,7 +573,7 @@ public class Commands
if (logger.isTraceEnabled())
logger.trace("{}: Maybe executing with status {}. Will notify listeners on noop: {}", command.txnId(), command.status(), alwaysNotifyListeners);
- if (command.status() != Stable && command.status() != PreApplied)
+ if (command.status() != Stable && command.saveStatus() != SaveStatus.PreApplied)
{
if (alwaysNotifyListeners)
safeStore.notifyListeners(safeCommand, command);
@@ -648,7 +648,7 @@ public class Commands
redundantBefore.removeRedundantDependencies(participants, update);
update.forEachWaitingOnId(safeStore, update, waiting, executeAt, (store, upd, w, exec, i) -> {
- SafeCommand dep = store.ifLoadedAndInitialisedAndNotErased(upd.txnId(i));
+ SafeCommand dep = store.ifLoadedAndInitialised(upd.txnId(i));
if (dep == null || !dep.current().hasBeen(PreCommitted))
return;
updateWaitingOn(store, w, exec, upd, dep);
@@ -870,7 +870,7 @@ public class Commands
return result;
}
- private static boolean validateSafeToCleanup(SafeCommandStore safeStore, Command command, StoreParticipants participants)
+ private static boolean validateSafeToCleanup(SafeCommandStore safeStore, Command command, @Nullable StoreParticipants participants)
{
TxnId txnId = command.txnId();
participants = command.participants().supplement(participants);
@@ -879,13 +879,13 @@ public class Commands
{
default: throw new AssertionError("Unhandled RedundantStatus: " + status);
case LIVE:
+ case WAS_OWNED_CLOSED:
case REDUNDANT_PRE_BOOTSTRAP_OR_STALE:
case PARTIALLY_PRE_BOOTSTRAP_OR_STALE:
return false;
case LOCALLY_REDUNDANT:
case SHARD_REDUNDANT:
Invariants.checkState(!command.hasBeen(PreCommitted));
- case WAS_OWNED:
case PRE_BOOTSTRAP_OR_STALE:
case NOT_OWNED:
return true;
@@ -894,7 +894,8 @@ public class Commands
public static boolean maybeCleanup(SafeCommandStore safeStore, SafeCommand safeCommand, Command command, @Nonnull StoreParticipants participants)
{
- Cleanup cleanup = shouldCleanup(safeStore, command, participants);
+ // TODO (desired): cheaper version of supplement for here
+ Cleanup cleanup = shouldCleanup(safeStore, command, command.participants().supplement(participants));
if (cleanup == NO)
return false;
@@ -998,7 +999,7 @@ public class Commands
}
}
- depSafe = safeStore.ifLoadedAndInitialisedAndNotErased(directlyBlockedOn);
+ depSafe = safeStore.ifLoadedAndInitialised(directlyBlockedOn);
if (depSafe == null)
{
loadDepId = directlyBlockedOn;
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index 97697d06..b4d5e15b 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -156,6 +156,13 @@ public interface PreLoadContext
return new Standard(primary, additional, keys, keyHistory);
}
+ default boolean contains(TxnId txnId)
+ {
+ TxnId primaryTxnId = primaryTxnId();
+ return primaryTxnId != null && (txnId.equals(primaryTxnId) || txnId.equals(additionalTxnId()));
+ }
+
+
class Standard implements PreLoadContext
{
private final @Nullable TxnId primary;
diff --git a/accord-core/src/main/java/accord/local/RedundantBefore.java b/accord-core/src/main/java/accord/local/RedundantBefore.java
index 4aef63a2..7dcae0c8 100644
--- a/accord-core/src/main/java/accord/local/RedundantBefore.java
+++ b/accord-core/src/main/java/accord/local/RedundantBefore.java
@@ -38,6 +38,7 @@ import accord.primitives.Ranges;
import accord.primitives.Routables;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
+import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
@@ -53,7 +54,7 @@ import static accord.local.RedundantStatus.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.NOT_OWNED;
import static accord.local.RedundantStatus.PRE_BOOTSTRAP_OR_STALE;
import static accord.local.RedundantStatus.SHARD_REDUNDANT;
-import static accord.local.RedundantStatus.WAS_OWNED;
+import static accord.local.RedundantStatus.WAS_OWNED_CLOSED;
import static accord.utils.Invariants.illegalState;
public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
@@ -155,14 +156,27 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
this.gcBefore = gcBefore;
this.bootstrappedAt = bootstrappedAt;
this.staleUntilAtLeast = staleUntilAtLeast;
- Invariants.checkArgument(locallyWitnessedOrInvalidatedBefore.equals(TxnId.NONE) || locallyWitnessedOrInvalidatedBefore.domain().isRange());
- Invariants.checkArgument(locallyAppliedOrInvalidatedBefore.equals(TxnId.NONE) || locallyAppliedOrInvalidatedBefore.domain().isRange());
- Invariants.checkArgument(locallyDecidedAndAppliedOrInvalidatedBefore.equals(TxnId.NONE) || locallyDecidedAndAppliedOrInvalidatedBefore.domain().isRange());
- Invariants.checkArgument(shardAppliedOrInvalidatedBefore.equals(TxnId.NONE) || shardAppliedOrInvalidatedBefore.domain().isRange());
- Invariants.checkArgument(gcBefore.equals(TxnId.NONE) || gcBefore.domain().isRange());
- Invariants.checkArgument(locallyDecidedAndAppliedOrInvalidatedBefore.compareTo(locallyAppliedOrInvalidatedBefore) <= 0);
- Invariants.checkArgument(shardAppliedOrInvalidatedBefore.compareTo(shardOnlyAppliedOrInvalidatedBefore) <= 0);
- Invariants.checkArgument(gcBefore.compareTo(shardAppliedOrInvalidatedBefore) <= 0);
+ checkNoneOrRX(locallyWitnessedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore, locallyDecidedAndAppliedOrInvalidatedBefore,
+ shardAppliedOrInvalidatedBefore, gcBefore);
+ checkPartiallyOrdered(locallyDecidedAndAppliedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore);
+ checkPartiallyOrdered(shardAppliedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore);
+ checkPartiallyOrdered(gcBefore, shardAppliedOrInvalidatedBefore, shardOnlyAppliedOrInvalidatedBefore);
+ }
+
+ private static void checkPartiallyOrdered(TxnId ... txnIds)
+ {
+ for (int i = 1 ; i < txnIds.length ; ++i)
+ Invariants.checkArgument(txnIds[i - 1].compareTo(txnIds[i]) <= 0);
+ }
+
+ private static void checkNoneOrRX(TxnId ... txnIds)
+ {
+ for (TxnId txnId : txnIds)
+ checkNoneOrRX(txnId);
+ }
+ private static void checkNoneOrRX(TxnId txnId)
+ {
+ Invariants.checkArgument(txnId.equals(TxnId.NONE) || (txnId.domain().isRange() && txnId.is(Txn.Kind.ExclusiveSyncPoint)));
}
public static Entry reduce(Entry a, Entry b)
@@ -201,7 +215,6 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
TxnId gcBefore = cg >= 0 ? cur.gcBefore : add.gcBefore;
TxnId bootstrappedAt = cb >= 0 ? cur.bootstrappedAt : add.bootstrappedAt;
Timestamp staleUntilAtLeast = csu >= 0 ? cur.staleUntilAtLeast : add.staleUntilAtLeast;
- TxnId shardAppliedOrInvalidatedBefore = TxnId.min(shardOnlyAppliedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore);
// if a NEW redundantBefore predates our current bootstrappedAt, we should not update it to avoid erroneously
// treating transactions prior as locally redundant when they may simply have not applied yet, since we may
@@ -212,6 +225,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
// TODO (desired): revisit later as semantics here evolve
if (bootstrappedAt.compareTo(locallyAppliedOrInvalidatedBefore) >= 0)
locallyAppliedOrInvalidatedBefore = cur.locallyAppliedOrInvalidatedBefore;
+
+ TxnId shardAppliedOrInvalidatedBefore = TxnId.min(shardOnlyAppliedOrInvalidatedBefore, locallyAppliedOrInvalidatedBefore);
if (bootstrappedAt.compareTo(shardAppliedOrInvalidatedBefore) >= 0)
locallyDecidedAndAppliedOrInvalidatedBefore = cur.locallyDecidedAndAppliedOrInvalidatedBefore;
if (staleUntilAtLeast != null && bootstrappedAt.compareTo(staleUntilAtLeast) >= 0)
@@ -364,7 +379,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Entry>
RedundantStatus get(TxnId txnId)
{
if (wasOwned(txnId) && isComplete())
- return WAS_OWNED;
+ return WAS_OWNED_CLOSED;
return getIgnoringOwnership(txnId);
}
diff --git a/accord-core/src/main/java/accord/local/RedundantStatus.java b/accord-core/src/main/java/accord/local/RedundantStatus.java
index 3ecdc073..a9944b66 100644
--- a/accord-core/src/main/java/accord/local/RedundantStatus.java
+++ b/accord-core/src/main/java/accord/local/RedundantStatus.java
@@ -30,9 +30,10 @@ public enum RedundantStatus
NOT_OWNED,
/**
- * None of the relevant ranges are owned by the command store anymore
+ * None of the relevant ranges are owned by the command store anymore, and all of those ranges are closed
+ * (i.e. we should not participate in any decisions)
*/
- WAS_OWNED,
+ WAS_OWNED_CLOSED,
/**
* Some of the relevant ranges are owned by the command store and valid for execution
@@ -84,25 +85,25 @@ public enum RedundantStatus
{
NOT_OWNED.merge = new EnumMap<>(RedundantStatus.class);
NOT_OWNED.merge.put(NOT_OWNED, NOT_OWNED);
- NOT_OWNED.merge.put(WAS_OWNED, WAS_OWNED);
+ NOT_OWNED.merge.put(WAS_OWNED_CLOSED, WAS_OWNED_CLOSED);
NOT_OWNED.merge.put(LIVE, LIVE);
NOT_OWNED.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
NOT_OWNED.merge.put(PRE_BOOTSTRAP_OR_STALE, PRE_BOOTSTRAP_OR_STALE);
NOT_OWNED.merge.put(REDUNDANT_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
NOT_OWNED.merge.put(LOCALLY_REDUNDANT, LOCALLY_REDUNDANT);
NOT_OWNED.merge.put(SHARD_REDUNDANT, SHARD_REDUNDANT);
- WAS_OWNED.merge = new EnumMap<>(RedundantStatus.class);
- WAS_OWNED.merge.put(NOT_OWNED, NOT_OWNED);
- WAS_OWNED.merge.put(WAS_OWNED, WAS_OWNED);
- WAS_OWNED.merge.put(LIVE, LIVE);
- WAS_OWNED.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
- WAS_OWNED.merge.put(PRE_BOOTSTRAP_OR_STALE, PRE_BOOTSTRAP_OR_STALE);
- WAS_OWNED.merge.put(REDUNDANT_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
- WAS_OWNED.merge.put(LOCALLY_REDUNDANT, LOCALLY_REDUNDANT);
- WAS_OWNED.merge.put(SHARD_REDUNDANT, SHARD_REDUNDANT);
+ WAS_OWNED_CLOSED.merge = new EnumMap<>(RedundantStatus.class);
+ WAS_OWNED_CLOSED.merge.put(NOT_OWNED, NOT_OWNED);
+ WAS_OWNED_CLOSED.merge.put(WAS_OWNED_CLOSED, WAS_OWNED_CLOSED);
+ WAS_OWNED_CLOSED.merge.put(LIVE, LIVE);
+ WAS_OWNED_CLOSED.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
+ WAS_OWNED_CLOSED.merge.put(PRE_BOOTSTRAP_OR_STALE, PRE_BOOTSTRAP_OR_STALE);
+ WAS_OWNED_CLOSED.merge.put(REDUNDANT_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
+ WAS_OWNED_CLOSED.merge.put(LOCALLY_REDUNDANT, LOCALLY_REDUNDANT);
+ WAS_OWNED_CLOSED.merge.put(SHARD_REDUNDANT, SHARD_REDUNDANT);
LIVE.merge = new EnumMap<>(RedundantStatus.class);
LIVE.merge.put(NOT_OWNED, LIVE);
- LIVE.merge.put(WAS_OWNED, LIVE);
+ LIVE.merge.put(WAS_OWNED_CLOSED, LIVE);
LIVE.merge.put(LIVE, LIVE);
LIVE.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
LIVE.merge.put(PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
@@ -111,7 +112,7 @@ public enum RedundantStatus
LIVE.merge.put(SHARD_REDUNDANT, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge = new EnumMap<>(RedundantStatus.class);
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(NOT_OWNED, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
- PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
+ PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED_CLOSED, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(LIVE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
@@ -120,7 +121,7 @@ public enum RedundantStatus
PARTIALLY_PRE_BOOTSTRAP_OR_STALE.merge.put(SHARD_REDUNDANT, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
PRE_BOOTSTRAP_OR_STALE.merge = new EnumMap<>(RedundantStatus.class);
PRE_BOOTSTRAP_OR_STALE.merge.put(NOT_OWNED, PRE_BOOTSTRAP_OR_STALE);
- PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED, PRE_BOOTSTRAP_OR_STALE);
+ PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED_CLOSED, PRE_BOOTSTRAP_OR_STALE);
PRE_BOOTSTRAP_OR_STALE.merge.put(LIVE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
PRE_BOOTSTRAP_OR_STALE.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, PARTIALLY_PRE_BOOTSTRAP_OR_STALE);
PRE_BOOTSTRAP_OR_STALE.merge.put(PRE_BOOTSTRAP_OR_STALE, PRE_BOOTSTRAP_OR_STALE);
@@ -129,7 +130,7 @@ public enum RedundantStatus
PRE_BOOTSTRAP_OR_STALE.merge.put(SHARD_REDUNDANT, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge = new EnumMap<>(RedundantStatus.class);
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(NOT_OWNED, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
- REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
+ REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(WAS_OWNED_CLOSED, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(LIVE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE); // TODO (expected): should this be an invalid combination?
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
@@ -138,7 +139,7 @@ public enum RedundantStatus
REDUNDANT_PRE_BOOTSTRAP_OR_STALE.merge.put(SHARD_REDUNDANT, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
LOCALLY_REDUNDANT.merge = new EnumMap<>(RedundantStatus.class);
LOCALLY_REDUNDANT.merge.put(NOT_OWNED, LOCALLY_REDUNDANT);
- LOCALLY_REDUNDANT.merge.put(WAS_OWNED, LOCALLY_REDUNDANT);
+ LOCALLY_REDUNDANT.merge.put(WAS_OWNED_CLOSED, LOCALLY_REDUNDANT);
LOCALLY_REDUNDANT.merge.put(LIVE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE); // TODO (expected): should this be an invalid combination?
LOCALLY_REDUNDANT.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
LOCALLY_REDUNDANT.merge.put(PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
@@ -147,7 +148,7 @@ public enum RedundantStatus
LOCALLY_REDUNDANT.merge.put(SHARD_REDUNDANT, LOCALLY_REDUNDANT);
SHARD_REDUNDANT.merge = new EnumMap<>(RedundantStatus.class);
SHARD_REDUNDANT.merge.put(NOT_OWNED, SHARD_REDUNDANT);
- SHARD_REDUNDANT.merge.put(WAS_OWNED, SHARD_REDUNDANT);
+ SHARD_REDUNDANT.merge.put(WAS_OWNED_CLOSED, SHARD_REDUNDANT);
SHARD_REDUNDANT.merge.put(LIVE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE); // TODO (expected): should this be an invalid combination?
SHARD_REDUNDANT.merge.put(PARTIALLY_PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
SHARD_REDUNDANT.merge.put(PRE_BOOTSTRAP_OR_STALE, REDUNDANT_PRE_BOOTSTRAP_OR_STALE);
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 72fb5bed..54639392 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -164,17 +164,18 @@ public abstract class SafeCommandStore
*
* This permits efficient operation when a transaction involved in processing another transaction happens to be in memory.
*/
- public SafeCommand ifLoadedAndInitialisedAndNotErased(TxnId txnId)
+ public SafeCommand ifLoadedAndInitialised(TxnId txnId)
{
SafeCommand safeCommand = getInternal(txnId);
- if (safeCommand != null)
- return safeCommand.current().saveStatus().isUninitialised() ? null : safeCommand;
-
- safeCommand = ifLoadedAndInitialisedAndNotErasedInternal(txnId);
- if (safeCommand == null || safeCommand.current().saveStatus().isUninitialised())
- return null;
+ if (safeCommand == null)
+ {
+ safeCommand = ifLoadedInternal(txnId);
+ if (safeCommand == null)
+ return null;
+ }
- return maybeCleanup(safeCommand, safeCommand.current(), StoreParticipants.empty(txnId));
+ safeCommand = maybeCleanup(safeCommand, safeCommand.current(), StoreParticipants.empty(txnId));
+ return safeCommand.isUnset() ? null : safeCommand;
}
protected SafeCommandsForKey maybeCleanup(SafeCommandsForKey safeCfk)
@@ -220,8 +221,8 @@ public abstract class SafeCommandStore
/** Get anything already referenced (should include anything in PreLoadContext). If returned, should be initialised. */
protected abstract SafeCommand getInternal(TxnId txnId);
- /** Get if available and initialised */
- protected abstract SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId);
+ /** Get if available */
+ protected abstract SafeCommand ifLoadedInternal(TxnId txnId);
/** Get anything already referenced (should include anything in PreLoadContext) */
protected abstract SafeCommandsForKey getInternal(RoutingKey key);
/** Get if available */
@@ -412,7 +413,8 @@ public abstract class SafeCommandStore
CommandStore commandStore = safeStore.commandStore();
Ranges ranges = next.participants().touches.toRanges();
commandStore.registerTransitive(safeStore, next.partialDeps().rangeDeps);
- commandStore.markSynced(safeStore, txnId, ranges);
+ if (txnId.is(Kind.ExclusiveSyncPoint))
+ commandStore.markSynced(safeStore, txnId, ranges);
}
/**
diff --git a/accord-core/src/main/java/accord/local/StoreParticipants.java b/accord-core/src/main/java/accord/local/StoreParticipants.java
index 7c8a319b..564b2b87 100644
--- a/accord-core/src/main/java/accord/local/StoreParticipants.java
+++ b/accord-core/src/main/java/accord/local/StoreParticipants.java
@@ -126,8 +126,9 @@ public class StoreParticipants
return new StoreParticipants(route, owns, touches, touches);
}
- public final StoreParticipants supplement(StoreParticipants that)
+ public final StoreParticipants supplement(@Nullable StoreParticipants that)
{
+ if (that == null) return this;
if (this == that) return this;
return supplement(that.route, that.owns, that.hasTouched);
}
diff --git a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
index 428d2ca0..84d50d77 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -312,13 +312,13 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
{
public static CommandsForKey create(RoutingKey key, TxnInfo[] txns, Unmanaged[] unmanageds, TxnId prunedBefore, RedundantBefore.Entry boundsInfo)
{
- return reconstruct(key, boundsInfo, txns, prunedBefore, unmanageds);
+ return reconstruct(key, boundsInfo, true, txns, prunedBefore, unmanageds);
}
}
interface Updater<O>
{
- O update(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds);
+ O update(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds);
}
/**
@@ -391,6 +391,11 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
return (encodedStatus & 0xf) == status.ordinal();
}
+ boolean hasBeen(InternalStatus status)
+ {
+ return (encodedStatus & 0xf) >= status.ordinal();
+ }
+
boolean is(TestStatus testStatus)
{
InternalStatus status = status();
@@ -677,7 +682,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
{
this.pending = pending;
this.txnId = txnId;
- this.waitingUntil = waitingUntil;
+ this.waitingUntil = Invariants.checkArgument(waitingUntil, waitingUntil != null);
}
@Override
@@ -829,6 +834,10 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
final Unmanaged[] unmanageds;
+ CommandsForKey(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int prunedBeforeById, Unmanaged[] unmanageds)
+ {
+ this(key, boundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, prunedBeforeById, unmanageds);
+ }
CommandsForKey(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int prunedBeforeById, Unmanaged[] unmanageds)
{
this.key = key;
@@ -1335,24 +1344,24 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
CommandsForKeyUpdate update(TxnInfo[] newById, int newMinUndecidedById, TxnInfo[] newCommittedByExecuteAt, int newMaxAppliedWriteByExecuteAt, Object[] newLoadingPruned, int newPrunedBeforeById, @Nullable TxnInfo curInfo, @Nonnull TxnInfo newInfo)
{
Invariants.checkState(prunedBeforeById < 0 || newById[newPrunedBeforeById].equals(byId[prunedBeforeById]));
- return updateAndNotifyUnmanageds(key, boundsInfo,
+ return updateAndNotifyUnmanageds(key, boundsInfo, false,
newById, newCommittedByExecuteAt, newMinUndecidedById, newMaxAppliedWriteByExecuteAt,
newLoadingPruned, newPrunedBeforeById, unmanageds, curInfo, newInfo);
}
- static CommandsForKey reconstruct(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, TxnId prunedBefore, Unmanaged[] unmanageds)
+ static CommandsForKey reconstruct(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, TxnId prunedBefore, Unmanaged[] unmanageds)
{
int prunedBeforeById = Arrays.binarySearch(byId, prunedBefore);
Invariants.checkState(prunedBeforeById >= 0 || prunedBefore.equals(TxnId.NONE));
- return reconstruct(key, boundsInfo, byId, BTree.empty(), prunedBeforeById, unmanageds);
+ return reconstruct(key, boundsInfo, isNewBoundsInfo, byId, BTree.empty(), prunedBeforeById, unmanageds);
}
- static CommandsForKey reconstruct(RoutingKey key, RedundantBefore.Entry newBoundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
+ static CommandsForKey reconstruct(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
{
- return reconstruct(key, newBoundsInfo, byId, loadingPruned, newPrunedBeforeById, unmanageds, CommandsForKey::new);
+ return reconstruct(key, boundsInfo, isNewBoundsInfo, byId, loadingPruned, newPrunedBeforeById, unmanageds, CommandsForKey::new);
}
- static <O> O reconstruct(RoutingKey key, RedundantBefore.Entry newBoundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds, Updater<O> updater)
+ static <O> O reconstruct(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds, Updater<O> updater)
{
int countCommitted = 0;
int minUndecidedById = -1;
@@ -1376,26 +1385,26 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
while (--maxAppliedWriteByExecuteAt >= 0)
{
TxnInfo txn = committedByExecuteAt[maxAppliedWriteByExecuteAt];
- if (txn.kind() == Write && (txn.status() == APPLIED || isPreBootstrap(txn, newBoundsInfo)))
+ if (txn.kind() == Write && (txn.status() == APPLIED || isPreBootstrap(txn, boundsInfo)))
break;
}
- return updater.update(key, newBoundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, newPrunedBeforeById, unmanageds);
+ return updater.update(key, boundsInfo, isNewBoundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, newPrunedBeforeById, unmanageds);
}
- static CommandsForKeyUpdate reconstructAndUpdateUnmanaged(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
+ static CommandsForKeyUpdate reconstructAndUpdateUnmanaged(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
{
- return reconstruct(key, boundsInfo, byId, loadingPruned, newPrunedBeforeById, unmanageds, CommandsForKey::updateAndNotifyUnmanageds);
+ return reconstruct(key, boundsInfo, isNewBoundsInfo, byId, loadingPruned, newPrunedBeforeById, unmanageds, CommandsForKey::updateAndNotifyUnmanageds);
}
- static CommandsForKeyUpdate updateAndNotifyUnmanageds(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
+ static CommandsForKeyUpdate updateAndNotifyUnmanageds(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds)
{
- return updateAndNotifyUnmanageds(key, boundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, newPrunedBeforeById, unmanageds, null, null);
+ return updateAndNotifyUnmanageds(key, boundsInfo, isNewBoundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, newPrunedBeforeById, unmanageds, null, null);
}
- static CommandsForKeyUpdate updateAndNotifyUnmanageds(RoutingKey key, RedundantBefore.Entry boundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds, @Nullable TxnInfo curInfo, @Nullable TxnInfo newInfo)
+ static CommandsForKeyUpdate updateAndNotifyUnmanageds(RoutingKey key, RedundantBefore.Entry boundsInfo, boolean isNewBoundsInfo, TxnInfo[] byId, TxnInfo[] committedByExecuteAt, int minUndecidedById, int maxAppliedWriteByExecuteAt, Object[] loadingPruned, int newPrunedBeforeById, Unmanaged[] unmanageds, @Nullable TxnInfo curInfo, @Nullable TxnInfo newInfo)
{
- NotifyUnmanagedResult notifyUnmanaged = PostProcess.notifyUnmanaged(unmanageds, byId, minUndecidedById, committedByExecuteAt, maxAppliedWriteByExecuteAt, loadingPruned, boundsInfo, curInfo, newInfo);
+ NotifyUnmanagedResult notifyUnmanaged = PostProcess.notifyUnmanaged(unmanageds, byId, minUndecidedById, committedByExecuteAt, maxAppliedWriteByExecuteAt, loadingPruned, boundsInfo, isNewBoundsInfo, curInfo, newInfo);
if (notifyUnmanaged != null)
unmanageds = notifyUnmanaged.newUnmanaged;
CommandsForKey result = new CommandsForKey(key, boundsInfo, byId, committedByExecuteAt, minUndecidedById, maxAppliedWriteByExecuteAt, loadingPruned, newPrunedBeforeById, unmanageds);
@@ -1599,7 +1608,6 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
switch (kind)
{
default: throw new AssertionError("Unhandled Txn.Kind: " + kind);
- case LocalOnly:
case EphemeralRead:
throw illegalState("Invalid Txn.Kind for CommandsForKey: " + kind);
@@ -1620,7 +1628,6 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
switch (kind)
{
default: throw new AssertionError("Unhandled Txn.Kind: " + kind);
- case LocalOnly:
case EphemeralRead:
case ExclusiveSyncPoint:
throw illegalState("Invalid Txn.Kind for CommandsForKey: " + kind);
@@ -1668,7 +1675,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
int newPrunedBeforeById = prunedBeforeId(newById, prunedBefore(), redundantBefore(newBoundsInfo));
Object[] newLoadingPruned = Pruning.removeRedundantLoadingPruned(loadingPruned, redundantBefore(newBoundsInfo));
- return notifyManagedPreBootstrap(this, newBoundsInfo, reconstructAndUpdateUnmanaged(key, newBoundsInfo, newById, newLoadingPruned, newPrunedBeforeById, unmanageds));
+ return notifyManagedPreBootstrap(this, newBoundsInfo, reconstructAndUpdateUnmanaged(key, newBoundsInfo, true, newById, newLoadingPruned, newPrunedBeforeById, unmanageds));
}
/**
@@ -1685,7 +1692,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
int newPrunedBeforeById = prunedBeforeId(newById, prunedBefore(), newRedundantBefore);
Object[] newLoadingPruned = Pruning.removeRedundantLoadingPruned(loadingPruned, newRedundantBefore);
- return reconstruct(key, newBoundsInfo, newById, newLoadingPruned, newPrunedBeforeById, unmanageds);
+ return reconstruct(key, newBoundsInfo, true, newById, newLoadingPruned, newPrunedBeforeById, unmanageds);
}
/**
diff --git a/accord-core/src/main/java/accord/local/cfk/NotifySink.java b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
index b6a415df..a869e8f0 100644
--- a/accord-core/src/main/java/accord/local/cfk/NotifySink.java
+++ b/accord-core/src/main/java/accord/local/cfk/NotifySink.java
@@ -49,7 +49,7 @@ interface NotifySink
@Override
public void notWaiting(SafeCommandStore safeStore, TxnId txnId, RoutingKey key)
{
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
if (safeCommand != null) notWaiting(safeStore, safeCommand, key);
else
{
diff --git a/accord-core/src/main/java/accord/local/cfk/PostProcess.java b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
index 1c187a00..2dc9a916 100644
--- a/accord-core/src/main/java/accord/local/cfk/PostProcess.java
+++ b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
@@ -91,7 +91,7 @@ abstract class PostProcess
for (TxnId txnId : load)
{
safeStore = safeStore; // make it unsafe for use in lambda
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
if (safeCommand != null) load(safeStore, safeCommand, safeCfk, notifySink);
else
safeStore.commandStore().execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(key), SYNC), safeStore0 -> {
@@ -196,7 +196,7 @@ abstract class PostProcess
List<PostProcess> nestedNotify = new ArrayList<>();
for (TxnId txnId : notify)
{
- SafeCommand safeCommand = safeStore.ifLoadedAndInitialisedAndNotErased(txnId);
+ SafeCommand safeCommand = safeStore.ifLoadedAndInitialised(txnId);
if (safeCommand != null)
{
CommandsForKeyUpdate update = updateUnmanaged(cfk, safeCommand, UPDATE, addUnmanageds);
@@ -246,6 +246,7 @@ abstract class PostProcess
int maxAppliedWriteByExecuteAt,
Object[] loadingPruned,
RedundantBefore.Entry boundsInfo,
+ boolean isNewBoundsInfo,
@Nullable TxnInfo curInfo,
@Nullable TxnInfo newInfo)
{
@@ -283,7 +284,18 @@ abstract class PostProcess
TxnInfo maxContiguousApplied = CommandsForKey.maxContiguousManagedApplied(committedByExecuteAt, maxAppliedWriteByExecuteAt, bootstrappedAt);
if (maxContiguousApplied != null)
applyTo = maxContiguousApplied.executeAt;
- // if we're updating bootstrappedAt, we can fire anyone waiting on an executeAt before the bootstrappedAt
+
+ // if we're updating bootstrappedAt, we can notify anyone waiting on an executeAt before the bootstrappedAt
+ if (applyTo == null && bootstrappedAt != null && isNewBoundsInfo)
+ {
+ // there's a special edge case here also, which is where we have removed entries from committedByExecuteAt because they're no longer owned
+ // but they could be waiting on Apply of some transaction with a higher executeAt
+ for (int i = byId.length - 1; i >= 0; --i)
+ {
+ TxnInfo txn = byId[i];
+ applyTo = Timestamp.nonNullOrMax(applyTo, txn.executeAt);
+ }
+ }
applyTo = Timestamp.nonNullOrMax(applyTo, TxnId.nonNullOrMax(redundantBefore, bootstrappedAt));
}
diff --git a/accord-core/src/main/java/accord/local/cfk/Updating.java b/accord-core/src/main/java/accord/local/cfk/Updating.java
index 079f8b45..0fa684d1 100644
--- a/accord-core/src/main/java/accord/local/cfk/Updating.java
+++ b/accord-core/src/main/java/accord/local/cfk/Updating.java
@@ -760,6 +760,7 @@ class Updating
Txn.Kind waitingKind = waitingTxnId.kind();
boolean readyToApply = true; // our dependencies have applied, so we are ready to apply
boolean waitingToApply = true; // our dependencies have committed, so we know when we execute and are waiting
+ boolean hasFutureDependency = false;
Timestamp executesAt = null;
int j = SortedArrays.binarySearch(byId, 0, byId.length, txnIds.get(i), Timestamp::compareTo, FAST);
@@ -774,44 +775,28 @@ class Updating
{
if (txn.status().compareTo(COMMITTED) < 0) waitingToApply = readyToApply = false;
else if (txn.status() != INVALID_OR_TRUNCATED_OR_PRUNED
- && (txn.executeAt.compareTo(waitingExecuteAt) < 0
- || waitingKind == EphemeralRead
- || (waitingKind == ExclusiveSyncPoint && txn.compareTo(waitingTxnId) < 0)
- ))
+ && (txn.executeAt.compareTo(waitingExecuteAt) < 0 || waitingKind.awaitsOnlyDeps))
{
- readyToApply &= txn.is(APPLIED);
- executesAt = Timestamp.nonNullOrMax(executesAt, txn.executeAt);
- }
- }
- ++i;
- ++j;
- }
- else if (c > 0)
- {
- if (waitingKind.isSyncPoint())
- {
- // we special-case sync points to handle pruning of transactions they should have witnessed as dependencies.
- // We require all earlier TxnId to be decided before we can execute; we must be informed of all the necessary TxnId
- // transitively by the "future" dependency provided for this purpose in mapReduceActive, so simply ensuring
- // all earlier TxnId are committed is sufficient to compute the correct executeAt.
- TxnInfo txn = byId[j];
- if (txn.mayExecute())
- {
- if (txn.status().compareTo(COMMITTED) < 0) readyToApply = waitingToApply = false;
- else if (txn.status() != INVALID_OR_TRUNCATED_OR_PRUNED
- && cfk.executes(txn.executeAt)
- && (txn.executeAt.compareTo(waitingExecuteAt) < 0
- || waitingKind == EphemeralRead
- || (waitingKind == ExclusiveSyncPoint && txn.compareTo(waitingTxnId) < 0)
- ))
+ if (waitingKind == ExclusiveSyncPoint && txn.compareTo(waitingTxnId) > 0)
+ {
+ // we don't wait for this txn itself to apply, but we do require it to be transitively committed
+ // (i.e., not just it must be committed, but all of its transitive dependencies;
+ // and then for the highest committed transaction < waitingTxnId to apply
+ if (cfk.minUndecidedById >= 0 && cfk.minUndecidedById <= j)
+ waitingToApply = readyToApply = false;
+ hasFutureDependency = true;
+ }
+ else
{
readyToApply &= txn.is(APPLIED);
executesAt = Timestamp.nonNullOrMax(executesAt, txn.executeAt);
}
}
}
+ ++i;
++j;
}
+ else if (c > 0) ++j;
else if (!cfk.mayExecute(txnIds.get(i))) ++i;
else if (register)
{
@@ -826,8 +811,26 @@ class Updating
}
}
- if (waitingKind.isSyncPoint() && Pruning.isAnyPredecessorWaitingOnPruned(cfk.loadingPruned, waitingTxnId))
- readyToApply = waitingToApply = false;
+ if (waitingKind.isSyncPoint())
+ {
+ if (Pruning.isAnyPredecessorWaitingOnPruned(cfk.loadingPruned, waitingTxnId))
+ {
+ readyToApply = waitingToApply = false;
+ }
+ else if (waitingKind == ExclusiveSyncPoint && hasFutureDependency && readyToApply)
+ {
+ int w = Arrays.binarySearch(byId, waitingTxnId);
+ if (w < 0) w = -1 - w;
+ while (--w >= 0)
+ {
+ TxnInfo txn = byId[w];
+ if (!txn.mayExecute()) continue;
+ readyToApply &= txn.hasBeen(APPLIED);
+ if (txn.isCommittedToExecute())
+ executesAt = Timestamp.nonNullOrMax(executesAt, byId[w].executeAt);
+ }
+ }
+ }
if (!readyToApply)
{
diff --git a/accord-core/src/main/java/accord/messages/Apply.java b/accord-core/src/main/java/accord/messages/Apply.java
index 862e2658..0288dfce 100644
--- a/accord-core/src/main/java/accord/messages/Apply.java
+++ b/accord-core/src/main/java/accord/messages/Apply.java
@@ -57,12 +57,12 @@ public class Apply extends TxnRequest<ApplyReply>
public interface Factory
{
- Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result);
+ Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> scope, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute);
}
public static Factory wrapForExclusiveSyncPoint(Factory factory)
{
- return (kind, to, participates, txnId, route, txn, executeAt, deps, writes, result) -> {
+ return (kind, to, participates, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute) -> {
while (true)
{
long minEpoch = participates.oldestEpoch();
@@ -81,7 +81,7 @@ public class Apply extends TxnRequest<ApplyReply>
break;
}
- return factory.create(kind, to, participates, txnId, route, txn, executeAt, deps, writes, result);
+ return factory.create(kind, to, participates, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute);
};
}
@@ -95,26 +95,26 @@ public class Apply extends TxnRequest<ApplyReply>
public enum Kind { Minimal, Maximal }
- protected Apply(Kind kind, Id to, Topologies participates, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
+ protected Apply(Kind kind, Id to, Topologies participates, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
{
- super(to, participates, route, txnId);
+ super(to, participates, sendTo, txnId);
Invariants.checkState(txnId.kind() != Txn.Kind.Write || writes != null);
// TODO (desired): it's wasteful to encode the full set of ranges owned by the recipient node;
// often it will be cheaper to include the FullRoute for Deps scope (or come up with some other safety-preserving encoding scheme)
this.kind = kind;
this.deps = deps.intersecting(scope);
this.txn = kind == Kind.Maximal ? txn.intersecting(scope, true) : null;
- this.fullRoute = kind == Kind.Maximal ? route : null;
+ this.fullRoute = kind == Kind.Maximal ? fullRoute : null;
this.executeAt = executeAt;
this.writes = writes;
this.result = result;
}
- public static void sendMaximal(Node node, Id to, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result)
+ public static void sendMaximal(Node node, Id to, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute)
{
- Topologies executes = executes(node, route, executeAt);
- Topologies participates = participates(node, route, txnId, executeAt, executes);
- node.send(to, applyMaximal(FACTORY, to, participates, txnId, route, txn, executeAt, deps, writes, result));
+ Topologies executes = executes(node, sendTo, executeAt);
+ Topologies participates = participates(node, sendTo, txnId, executeAt, executes);
+ node.send(to, applyMaximal(FACTORY, to, participates, txnId, sendTo, txn, executeAt, deps, writes, result, fullRoute));
}
public static Topologies executes(Node node, Unseekables<?> route, Timestamp executeAt)
@@ -127,9 +127,9 @@ public class Apply extends TxnRequest<ApplyReply>
return txnId.epoch() == executeAt.epoch() ? executes : node.topology().preciseEpochs(route, txnId.epoch(), executeAt.epoch());
}
- public static Apply applyMaximal(Factory factory, Id to, Topologies participates, TxnId txnId, FullRoute<?> route, Txn txn, Timestamp executeAt, Deps stableDeps, Writes writes, Result result)
+ public static Apply applyMaximal(Factory factory, Id to, Topologies participates, TxnId txnId, Route<?> sendTo, Txn txn, Timestamp executeAt, Deps stableDeps, Writes writes, Result result, FullRoute<?> fullRoute)
{
- return factory.create(Kind.Maximal, to, participates, txnId, route, txn, executeAt, stableDeps, writes, result);
+ return factory.create(Kind.Maximal, to, participates, txnId, sendTo, txn, executeAt, stableDeps, writes, result, fullRoute);
}
protected Apply(Kind kind, TxnId txnId, Route<?> route, long waitForEpoch, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
diff --git a/accord-core/src/main/java/accord/messages/CheckStatus.java b/accord-core/src/main/java/accord/messages/CheckStatus.java
index 24a32e0d..e4e5a0a8 100644
--- a/accord-core/src/main/java/accord/messages/CheckStatus.java
+++ b/accord-core/src/main/java/accord/messages/CheckStatus.java
@@ -64,6 +64,7 @@ import static accord.primitives.Status.Durability.Universal;
import accord.primitives.Known;
import accord.utils.async.Cancellable;
+import static accord.primitives.Status.Invalidated;
import static accord.primitives.Status.NotDefined;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
@@ -310,7 +311,7 @@ public class CheckStatus extends AbstractRequest<CheckStatus.CheckStatusReply>
}
Known validForAll = map.computeValidForAll(routeOrParticipants);
- if (withQuorum == HasQuorum && invalidIf.inferInvalidWithQuorum(finished.maxKnown()))
+ if (withQuorum == HasQuorum && invalidIf.inferInvalidWithQuorum(finished.minKnown()))
validForAll = validForAll.atLeast(Known.Invalidated);
return finished.with(map.with(validForAll));
@@ -447,6 +448,11 @@ public class CheckStatus extends AbstractRequest<CheckStatus.CheckStatusReply>
return map.foldl(Known::atLeast, Known.Nothing, i -> false);
}
+ public Known minKnown()
+ {
+ return map.foldl(Known::nonNullOrMin, null, i -> false);
+ }
+
@Override
public MessageType type()
{
diff --git a/accord-core/src/main/java/accord/messages/InformDurable.java b/accord-core/src/main/java/accord/messages/InformDurable.java
index 061849a9..7a9f76a3 100644
--- a/accord-core/src/main/java/accord/messages/InformDurable.java
+++ b/accord-core/src/main/java/accord/messages/InformDurable.java
@@ -75,6 +75,8 @@ public class InformDurable extends TxnRequest<Reply> implements PreLoadContext
public Reply apply(SafeCommandStore safeStore)
{
StoreParticipants participants = StoreParticipants.update(safeStore, scope, txnId.epoch(), txnId, txnId.epoch());
+ if (txnId.toString().equals("[7,12610,9(RX),2]") && safeStore.commandStore().toString().equals("DelayedCommandStore{id=0,node=1}"))
+ System.out.println();
SafeCommand safeCommand = safeStore.get(txnId, participants);
if (safeCommand.current().is(Status.Truncated))
return Ok;
diff --git a/accord-core/src/main/java/accord/messages/PreAccept.java b/accord-core/src/main/java/accord/messages/PreAccept.java
index 8a5fbf59..0b7a0efb 100644
--- a/accord-core/src/main/java/accord/messages/PreAccept.java
+++ b/accord-core/src/main/java/accord/messages/PreAccept.java
@@ -127,6 +127,9 @@ public class PreAccept extends WithUnsynced<PreAccept.PreAcceptReply>
if (command.saveStatus().compareTo(SaveStatus.PreAccepted) > 0)
return PreAcceptNack.INSTANCE;
+ if (command.executeAt().isRejected())
+ return new PreAcceptOk(txnId, command.executeAt(), Deps.NONE);
+
Deps deps = calculateDeps(safeStore, txnId, participants, EpochSupplier.constant(minEpoch), txnId);
// NOTE: we CANNOT test whether we adopt a future dependency here because it might be that this command
// is guaranteed to not reach agreement, but that this replica is unaware of that fact and has pruned
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index f872d08b..c6df62bf 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -417,7 +417,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
{
default: throw new AssertionError("Unhandled RedundantStatus: " + status);
case NOT_OWNED:
- case WAS_OWNED:
+ case WAS_OWNED_CLOSED:
case SHARD_REDUNDANT:
case LOCALLY_REDUNDANT:
purge(safeStore, safeCommand, null, ERASE, true);
diff --git a/accord-core/src/main/java/accord/messages/ReadData.java b/accord-core/src/main/java/accord/messages/ReadData.java
index 5e1d4635..1d40b8f9 100644
--- a/accord-core/src/main/java/accord/messages/ReadData.java
+++ b/accord-core/src/main/java/accord/messages/ReadData.java
@@ -63,7 +63,7 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
{
private static final Logger logger = LoggerFactory.getLogger(ReadData.class);
- private enum State { PENDING, RETURNED, OBSOLETE }
+ private enum State { PENDING, PENDING_OBSOLETE, RETURNED, OBSOLETE }
protected enum StoreAction { WAIT, EXECUTE, OBSOLETE }
public enum ReadType
@@ -176,6 +176,7 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
this.cancel = cancel;
this.timeout = timeout;
return;
+ case PENDING_OBSOLETE:
case OBSOLETE:
timeout = null;
case RETURNED:
@@ -228,7 +229,7 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
return status.compareTo(SaveStatus.Stable) >= 0 ? null : Insufficient;
case OBSOLETE:
- state = State.OBSOLETE;
+ state = State.PENDING_OBSOLETE;
return Redundant;
case EXECUTE:
@@ -400,6 +401,7 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
case RETURNED:
throw illegalState("ReadOk was sent, yet ack called again");
+ case PENDING_OBSOLETE:
case OBSOLETE:
logger.debug("Before the read completed successfully for txn {}, the result was marked obsolete", txnId);
return null;
@@ -418,7 +420,7 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
Cancellable clear;
synchronized (this)
{
- if (state != State.PENDING)
+ if (state.compareTo(State.PENDING_OBSOLETE) > 0)
return false;
state = State.OBSOLETE;
diff --git a/accord-core/src/main/java/accord/primitives/AbstractRanges.java b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
index 141bded3..873e8c52 100644
--- a/accord-core/src/main/java/accord/primitives/AbstractRanges.java
+++ b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
@@ -181,13 +181,6 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
}
}
- public Ranges without(Unseekables<?> keysOrRanges)
- {
- if (keysOrRanges.domain() == Routable.Domain.Key)
- keysOrRanges = ((AbstractUnseekableKeys)keysOrRanges).toRanges();
- return without((AbstractRanges) keysOrRanges);
- }
-
// returns ri in low 32 bits, ki in top, or -1 if no match found
@Override
public final long findNextIntersection(int ri, AbstractKeys<?> keys, int ki)
@@ -240,21 +233,13 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
/**
* Subtracts the given set of ranges from this
*/
- public Ranges without(Ranges that)
- {
- return without((AbstractRanges) that);
- }
-
- /**
- * Subtracts the given set of ranges from this
- */
- private Ranges without(AbstractRanges that)
+ Range[] withoutInternal(AbstractRanges that)
{
if (that.isEmpty())
- return this instanceof Ranges ? (Ranges)this : ofSortedAndDeoverlappedUnchecked(ranges);
+ return ranges;
if (isEmpty() || that == this)
- return EMPTY;
+ return EMPTY.ranges;
ObjectBuffers<Range> cachedRanges = cachedRanges();
Range[] result = null;
@@ -310,9 +295,9 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
}
if (count == 0)
- return EMPTY;
+ return EMPTY.ranges;
- return ofSortedAndDeoverlappedUnchecked(cachedRanges.completeAndDiscard(result, count));
+ return cachedRanges.completeAndDiscard(result, count);
}
/**
diff --git a/accord-core/src/main/java/accord/primitives/KeyRoute.java b/accord-core/src/main/java/accord/primitives/KeyRoute.java
index 7970ccad..b03b64c2 100644
--- a/accord-core/src/main/java/accord/primitives/KeyRoute.java
+++ b/accord-core/src/main/java/accord/primitives/KeyRoute.java
@@ -57,6 +57,37 @@ public abstract class KeyRoute extends AbstractUnseekableKeys implements Route<R
return new PartialKeyRoute(homeKey, mergedKeys);
}
+ @Override
+ public KeyRoute without(Unseekables<?> keysOrRanges)
+ {
+ switch (keysOrRanges.domain())
+ {
+ default: throw new AssertionError("Unhandled domain: " + keysOrRanges.domain());
+ case Key:
+ {
+ AbstractUnseekableKeys that = (AbstractUnseekableKeys) keysOrRanges;
+ RoutingKey[] newKeys = SortedArrays.linearSubtract(this.keys, that.keys, cachedRoutingKeys());
+ return newKeys == keys ? this : new PartialKeyRoute(homeKey, newKeys);
+ }
+ case Range:
+ {
+ return without((AbstractRanges)keysOrRanges);
+ }
+ }
+ }
+
+ @Override
+ public KeyRoute without(Ranges ranges)
+ {
+ return without((AbstractRanges) ranges);
+ }
+
+ private KeyRoute without(AbstractRanges ranges)
+ {
+ RoutingKey[] newKeys = subtract(ranges, keys, RoutingKey[]::new);
+ return newKeys == keys ? this : new PartialKeyRoute(homeKey, newKeys);
+ }
+
@SuppressWarnings("unchecked")
@Override
public Participants<RoutingKey> participants()
diff --git a/accord-core/src/main/java/accord/primitives/Keys.java b/accord-core/src/main/java/accord/primitives/Keys.java
index abfd62b5..3469db7d 100644
--- a/accord-core/src/main/java/accord/primitives/Keys.java
+++ b/accord-core/src/main/java/accord/primitives/Keys.java
@@ -262,16 +262,22 @@ public class Keys extends AbstractKeys<Key> implements Seekables<Key, Keys>
public static Keys of(Set<? extends Key> keys)
{
+ if (keys.size() == 0)
+ return EMPTY;
return ofUnique(keys.toArray(new Key[0]));
}
public static Keys ofSorted(Key ... keys)
{
+ if (keys.length == 0)
+ return EMPTY;
return new Keys(SortedArrays.toUnique(keys));
}
public static Keys ofSortedUnique(Key ... keys)
{
+ if (keys.length == 0)
+ return EMPTY;
if (!isSortedUnique(keys))
throw new IllegalArgumentException(Arrays.toString(keys) + " is not sorted");
return new Keys(keys);
@@ -282,11 +288,6 @@ public class Keys extends AbstractKeys<Key> implements Seekables<Key, Keys>
return ofSortedUnique(keys.toArray(new Key[0]));
}
- static Keys ofSortedUnchecked(Key ... keys)
- {
- return new Keys(keys);
- }
-
private Keys wrap(Key[] wrap, AbstractKeys<Key> that)
{
return wrap == keys ? this : wrap == that.keys && that instanceof Keys ? (Keys)that : new Keys(wrap);
diff --git a/accord-core/src/main/java/accord/primitives/Known.java b/accord-core/src/main/java/accord/primitives/Known.java
index 4b5fc6c5..0b934fa0 100644
--- a/accord-core/src/main/java/accord/primitives/Known.java
+++ b/accord-core/src/main/java/accord/primitives/Known.java
@@ -96,6 +96,10 @@ public class Known
return selectOrCreate(that, maxRoute, maxDefinition, maxExecuteAt, maxDeps, maxOutcome);
}
+ public static Known nonNullOrMin(Known a, Known b)
+ {
+ return a == null | b == null ? a == null ? b : a : a.min(b);
+ }
public Known min(Known that)
{
Invariants.checkArgument(compatibleWith(that));
@@ -613,7 +617,7 @@ public class Known
public boolean isOrWasKnown()
{
- return this == DefinitionKnown;
+ return this != DefinitionUnknown;
}
public Definition atLeast(Definition that)
diff --git a/accord-core/src/main/java/accord/primitives/LatestDeps.java b/accord-core/src/main/java/accord/primitives/LatestDeps.java
index 313eddb3..adae0e37 100644
--- a/accord-core/src/main/java/accord/primitives/LatestDeps.java
+++ b/accord-core/src/main/java/accord/primitives/LatestDeps.java
@@ -381,7 +381,6 @@ public class LatestDeps extends ReducingRangeMap<LatestDeps.LatestEntry>
};
}
- // TODO (now): override tryMergeEqual
static class MergeBuilder extends AbstractIntervalBuilder<RoutingKey, MergeEntry, Merge>
{
protected MergeBuilder(boolean inclusiveEnds, int capacity)
diff --git a/accord-core/src/main/java/accord/primitives/RangeDeps.java b/accord-core/src/main/java/accord/primitives/RangeDeps.java
index 4f8a8811..2ea91e46 100644
--- a/accord-core/src/main/java/accord/primitives/RangeDeps.java
+++ b/accord-core/src/main/java/accord/primitives/RangeDeps.java
@@ -80,7 +80,7 @@ import static accord.utils.SortedArrays.Search.FAST;
* TODO (expected): keep only the latest exclusive sync point for a given range
* TODO (expected): permit building out-of-order
* TODO (expected): currently permitting duplicates
- * TODO (required): randomised testing of all iteration methods (just found a double increment bug)
+ * TODO (testing): randomised testing of all iteration methods
*/
public class RangeDeps implements Iterable<Map.Entry<Range, TxnId>>
{
diff --git a/accord-core/src/main/java/accord/primitives/RangeRoute.java b/accord-core/src/main/java/accord/primitives/RangeRoute.java
index 808461fc..cbc74d3e 100644
--- a/accord-core/src/main/java/accord/primitives/RangeRoute.java
+++ b/accord-core/src/main/java/accord/primitives/RangeRoute.java
@@ -19,6 +19,7 @@
package accord.primitives;
import accord.api.RoutingKey;
+import accord.primitives.Routable.Domain;
import accord.utils.Invariants;
import javax.annotation.Nonnull;
@@ -59,6 +60,24 @@ public abstract class RangeRoute extends AbstractRanges implements Route<Range>,
return with((Unseekables<Range>) Ranges.of(withKey.asRange()));
}
+ @Override
+ public RangeRoute without(Unseekables<?> keysOrRanges)
+ {
+ return without(keysOrRanges.domain() == Domain.Range ? (AbstractRanges) keysOrRanges : ((AbstractUnseekableKeys)keysOrRanges).toRanges());
+ }
+
+ @Override
+ public RangeRoute without(Ranges that)
+ {
+ return without((AbstractRanges) that);
+ }
+
+ private RangeRoute without(AbstractRanges that)
+ {
+ Range[] newRanges = withoutInternal(that);
+ return newRanges == ranges ? this : new PartialRangeRoute(homeKey, newRanges);
+ }
+
@Override
public Route<Range> withHomeKey()
{
diff --git a/accord-core/src/main/java/accord/primitives/Ranges.java b/accord-core/src/main/java/accord/primitives/Ranges.java
index ed536abb..9e87c38f 100644
--- a/accord-core/src/main/java/accord/primitives/Ranges.java
+++ b/accord-core/src/main/java/accord/primitives/Ranges.java
@@ -175,6 +175,21 @@ public class Ranges extends AbstractRanges implements Iterable<Range>, Seekables
return with(Ranges.of(withKey.asRange()));
}
+ public Ranges without(Unseekables<?> keysOrRanges)
+ {
+ if (keysOrRanges.domain() == Routable.Domain.Key)
+ keysOrRanges = ((AbstractUnseekableKeys)keysOrRanges).toRanges();
+ return ofSortedAndDeoverlappedUnchecked(withoutInternal((AbstractRanges) keysOrRanges));
+ }
+
+ /**
+ * Subtracts the given set of ranges from this
+ */
+ public Ranges without(Ranges that)
+ {
+ return ofSortedAndDeoverlappedUnchecked(withoutInternal(that));
+ }
+
@Override
public UnseekablesKind kind()
{
diff --git a/accord-core/src/main/java/accord/primitives/Route.java b/accord-core/src/main/java/accord/primitives/Route.java
index 63d85808..fb75ad4a 100644
--- a/accord-core/src/main/java/accord/primitives/Route.java
+++ b/accord-core/src/main/java/accord/primitives/Route.java
@@ -38,6 +38,8 @@ public interface Route<K extends Unseekable> extends Participants<K>
*/
@Override
Route<K> with(Participants<K> participants);
+ @Override Route<K> without(Ranges ranges);
+ @Override Route<K> without(Unseekables<?> without);
@Override
Route<K> slice(int from, int to);
diff --git a/accord-core/src/main/java/accord/primitives/RoutingKeys.java b/accord-core/src/main/java/accord/primitives/RoutingKeys.java
index c9f67cac..25e40d97 100644
--- a/accord-core/src/main/java/accord/primitives/RoutingKeys.java
+++ b/accord-core/src/main/java/accord/primitives/RoutingKeys.java
@@ -49,6 +49,8 @@ public class RoutingKeys extends AbstractUnseekableKeys implements Unseekables<R
public static RoutingKeys of(RoutingKey ... keys)
{
+ if (keys.length == 0)
+ return EMPTY;
return new RoutingKeys(toUnique(sort(keys)));
}
@@ -68,6 +70,8 @@ public class RoutingKeys extends AbstractUnseekableKeys implements Unseekables<R
public static RoutingKeys ofSortedUnique(RoutingKey ... keys)
{
+ if (keys.length == 0)
+ return EMPTY;
checkArgument(isSortedUnique(keys));
return new RoutingKeys(keys);
}
diff --git a/accord-core/src/main/java/accord/primitives/Timestamp.java b/accord-core/src/main/java/accord/primitives/Timestamp.java
index da8cb733..0a9de552 100644
--- a/accord-core/src/main/java/accord/primitives/Timestamp.java
+++ b/accord-core/src/main/java/accord/primitives/Timestamp.java
@@ -37,7 +37,7 @@ public class Timestamp implements Comparable<Timestamp>, EpochSupplier
* which we may also want to retain when merging in other contexts (such as in Deps).
*/
private static final int MERGE_FLAGS = 0x8000;
- // TODO (required): is this the correct set of identity bits?
+ // TODO (testing): is this the correct set of identity bits?
private static final long IDENTITY_LSB = 0xFFFFFFFFFFFF001EL;
public static final int IDENTITY_FLAGS = 0x001E;
public static final long MAX_EPOCH = (1L << 48) - 1;
diff --git a/accord-core/src/main/java/accord/primitives/Txn.java b/accord-core/src/main/java/accord/primitives/Txn.java
index 99e18979..55645602 100644
--- a/accord-core/src/main/java/accord/primitives/Txn.java
+++ b/accord-core/src/main/java/accord/primitives/Txn.java
@@ -103,12 +103,7 @@ public interface Txn
// TODO (expected): introduce a special kind of visible ExclusiveSyncPoint that creates a precise moment,
// and is therefore visible to all transactions.
ExclusiveSyncPoint('X', true, true, true, true),
-
- /**
- * Used for local book-keeping only, not visible to any other replica or directly to other transactions.
- * This is used to create pseudo transactions that take the place of dependencies that will be fulfilled by a bootstrap.
- */
- LocalOnly('L', false, false, true, false);
+ ;
public enum Kinds
{
@@ -198,11 +193,6 @@ public interface Txn
return this == Read;
}
- public boolean isLocal()
- {
- return this == LocalOnly;
- }
-
public boolean isDurable()
{
return this != EphemeralRead;
diff --git a/accord-core/src/main/java/accord/topology/TopologyManager.java b/accord-core/src/main/java/accord/topology/TopologyManager.java
index da310314..8825fceb 100644
--- a/accord-core/src/main/java/accord/topology/TopologyManager.java
+++ b/accord-core/src/main/java/accord/topology/TopologyManager.java
@@ -638,7 +638,7 @@ public class TopologyManager
return new Single(sorter, epochs.get(epoch).global);
}
- // TODO (required): test all of these methods when asking for epochs that have been cleaned up (and other code paths)
+ // TODO (testing): test all of these methods when asking for epochs that have been cleaned up (and other code paths)
public Topologies withUnsyncedEpochs(Unseekables<?> select, Timestamp min, Timestamp max)
{
return withUnsyncedEpochs(select, min.epoch(), max.epoch());
diff --git a/accord-core/src/main/java/accord/utils/SortedArrays.java b/accord-core/src/main/java/accord/utils/SortedArrays.java
index f6d5e714..a1cfe6ef 100644
--- a/accord-core/src/main/java/accord/utils/SortedArrays.java
+++ b/accord-core/src/main/java/accord/utils/SortedArrays.java
@@ -748,15 +748,15 @@ public class SortedArrays
while (keepFrom < keepTo && subtractFrom < subtractTo)
{
- I leftKey = keep[keepFrom];
- I rightKey = subtract[subtractFrom];
- int cmp = leftKey == rightKey ? 0 : comparator.compare(leftKey, rightKey);
+ I keepKey = keep[keepFrom];
+ I subtractKey = subtract[subtractFrom];
+ int cmp = keepKey == subtractKey ? 0 : comparator.compare(keepKey, subtractKey);
- if (cmp > 0)
+ if (cmp < 0)
{
result[resultSize++] = keep[keepFrom++];
}
- else if (cmp < 0)
+ else if (cmp > 0)
{
++subtractFrom;
}
diff --git a/accord-core/src/main/java/accord/utils/ThreadPoolScheduler.java b/accord-core/src/main/java/accord/utils/ThreadPoolScheduler.java
index f26a94fd..f2409f41 100644
--- a/accord-core/src/main/java/accord/utils/ThreadPoolScheduler.java
+++ b/accord-core/src/main/java/accord/utils/ThreadPoolScheduler.java
@@ -101,7 +101,7 @@ public class ThreadPoolScheduler implements Scheduler
public void stop()
{
- exec.shutdown();
+ exec.shutdownNow();
try
{
if (!exec.awaitTermination(1L, TimeUnit.MINUTES))
diff --git a/accord-core/src/main/java/accord/utils/Timestamped.java b/accord-core/src/main/java/accord/utils/Timestamped.java
index fad231f2..3b7f60a3 100644
--- a/accord-core/src/main/java/accord/utils/Timestamped.java
+++ b/accord-core/src/main/java/accord/utils/Timestamped.java
@@ -73,4 +73,9 @@ public class Timestamped<T>
{
return toString.apply(data);
}
+
+ public boolean equals(Timestamped<T> that, BiPredicate<T, T> equality)
+ {
+ return this.timestamp.equals(that.timestamp) && equality.test(this.data, that.data);
+ }
}
diff --git a/accord-core/src/test/java/accord/KeysTest.java b/accord-core/src/test/java/accord/KeysTest.java
index 138de0d3..dcc3c3c6 100644
--- a/accord-core/src/test/java/accord/KeysTest.java
+++ b/accord-core/src/test/java/accord/KeysTest.java
@@ -19,19 +19,32 @@
package accord;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.IntFunction;
import java.util.stream.IntStream;
import accord.api.Key;
+import accord.api.RoutingKey;
import accord.impl.IntKey;
import accord.impl.IntKey.Raw;
+import accord.primitives.AbstractKeys;
+import accord.primitives.Participants;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Keys;
+import accord.primitives.RoutableKey;
+import accord.primitives.Routables;
+import accord.primitives.RoutingKeys;
import accord.utils.Gen;
import accord.utils.Gens;
+import accord.utils.RandomTestRunner;
+import org.agrona.collections.IntHashSet;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -225,6 +238,51 @@ public class KeysTest
});
}
+ @Test
+ public void keysWithout()
+ {
+ without(IntKey::key, Key[]::new, Keys::of, Keys::without);
+ }
+
+ @Test
+ public void routingKeysWithout()
+ {
+ without(IntKey::routing, RoutingKey[]::new, RoutingKeys::of, RoutingKeys::without);
+ }
+
+ private <K extends RoutableKey, KS extends AbstractKeys<K>> void without(IntFunction<K> keyFactory,
+ IntFunction<K[]> arrayFactory,
+ Function<K[], KS> keysFactory,
+ BiFunction<KS, KS, Routables<K>> without)
+ {
+ final int maxSize = 100;
+ final int maxKey = 1000;
+ for (int test = 0 ; test < 10000 ; ++test)
+ {
+ RandomTestRunner.test().check(rs -> {
+ K[] keys = arrayFactory.apply(rs.nextInt(2, maxSize));
+ IntHashSet used = new IntHashSet();
+ for (int i = 0 ; i < keys.length ; ++i)
+ {
+ int k = rs.nextInt(maxKey);
+ while (!used.add(k)) k = rs.nextInt(maxKey);
+ keys[i] = keyFactory.apply(k);
+ }
+ int addTo = rs.nextInt(1, keys.length);
+ int removeFrom = addTo - rs.nextInt(addTo);
+ KS add = keysFactory.apply(Arrays.copyOf(keys, addTo));
+ KS remove = keysFactory.apply(Arrays.copyOfRange(keys, removeFrom, keys.length));
+ Routables<K> result = without.apply(add, remove);
+ Assertions.assertEquals(removeFrom, result.size());
+ if (removeFrom == addTo)
+ Assertions.assertSame(result, add);
+ Arrays.sort(keys, 0, removeFrom);
+ for (int i = 0 ; i < removeFrom ; ++i)
+ Assertions.assertEquals(keys[i], result.get(i));
+ });
+ }
+ }
+
@Test
public void foldl()
{
diff --git a/accord-core/src/test/java/accord/burn/BurnTest.java b/accord-core/src/test/java/accord/burn/BurnTest.java
index 4678cec3..bfbb7c7b 100644
--- a/accord-core/src/test/java/accord/burn/BurnTest.java
+++ b/accord-core/src/test/java/accord/burn/BurnTest.java
@@ -314,7 +314,7 @@ public class BurnTest
RandomSource retryRandom = random.fork();
retryBootstrap = retry -> {
long delay = retryRandom.nextInt(1, 15);
- queue.add((PendingRunnable) retry::run, delay, TimeUnit.SECONDS);
+ queue.add(PendingRunnable.create(retry::run), delay, TimeUnit.SECONDS);
};
}
IntSupplier coordinationDelays, progressDelays, timeoutDelays;
diff --git a/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java b/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java
index 95517ae0..ab39e5ad 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java
@@ -28,7 +28,7 @@ import java.util.ArrayList;
public class FastPathTrackerReconciler extends TrackerReconciler<FastPathShardTracker, FastPathTracker, FastPathTrackerReconciler.Rsp>
{
- enum Rsp { FAST, SLOW, FAIL }
+ enum Rsp { FAST, SLOW, DELAY, FAIL }
FastPathTrackerReconciler(RandomSource random, Topologies topologies)
{
@@ -48,6 +48,7 @@ public class FastPathTrackerReconciler extends TrackerReconciler<FastPathShardTr
default: throw new AssertionError();
case FAST: inflight.remove(from); return tracker.recordSuccess(from, true);
case SLOW: inflight.remove(from); return tracker.recordSuccess(from, false);
+ case DELAY: return tracker.recordDelayed(from);
case FAIL: inflight.remove(from); return tracker.recordFailure(from);
}
}
@@ -64,7 +65,7 @@ public class FastPathTrackerReconciler extends TrackerReconciler<FastPathShardTr
case Success:
Assertions.assertTrue(tracker.all(FastPathShardTracker::hasReachedQuorum));
- Assertions.assertTrue(tracker.all(shard -> shard.hasRejectedFastPath() || shard.hasMetFastPathCriteria()));
+ Assertions.assertTrue(tracker.all(shard -> shard.hasRejectedFastPath() || shard.hasMetFastPathCriteria() || shard.isFastPathDelayed()));
Assertions.assertFalse(tracker.any(FastPathShardTracker::hasFailed));
break;
diff --git a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java
index 0f53bd48..13810a8e 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java
@@ -37,6 +37,7 @@ public class ReadTrackerReconciler extends TrackerReconciler<ReadShardTracker, R
public InFlightCapturingReadTracker(Topologies topologies)
{
super(topologies);
+ initialise();
}
@Override
diff --git a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java
index 04eecf14..348cc6eb 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java
@@ -62,6 +62,7 @@ public class ReadTrackerTest
public TestReadTracker(Topologies topologies)
{
super(topologies);
+ Assertions.assertTrue(initialise());
}
@Override
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index 2b12039c..7f1f81b8 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -414,7 +414,7 @@ public class RemoteListenersTest
@Override public CommandStore commandStore() { return commandStore; }
@Override protected SafeCommand getInternal(TxnId txnId) { return null; }
- @Override protected SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId) { return null; }
+ @Override protected SafeCommand ifLoadedInternal(TxnId txnId) { return null; }
@Override protected SafeCommandsForKey getInternal(RoutingKey key) { return null;}
@Override protected SafeCommandsForKey ifLoadedInternal(RoutingKey key) { return null;}
@Override public SafeTimestampsForKey timestampsForKey(RoutingKey key) { return null; }
diff --git a/accord-core/src/test/java/accord/impl/basic/Cluster.java b/accord-core/src/test/java/accord/impl/basic/Cluster.java
index 56b7a970..3dd73e6f 100644
--- a/accord-core/src/test/java/accord/impl/basic/Cluster.java
+++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java
@@ -29,7 +29,9 @@ import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.NavigableMap;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -53,6 +55,7 @@ import accord.api.LocalConfig;
import accord.api.MessageSink;
import accord.api.RoutingKey;
import accord.api.Scheduler;
+import accord.api.Scheduler.Scheduled;
import accord.burn.BurnTestConfigurationService;
import accord.burn.TopologyUpdates;
import accord.burn.random.FrequentLargeRange;
@@ -82,7 +85,9 @@ import accord.local.CommandStore;
import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
+import accord.local.RedundantBefore;
import accord.local.TimeService;
+import accord.primitives.RoutableKey;
import accord.primitives.SaveStatus;
import accord.local.ShardDistributor;
import accord.primitives.Seekables;
@@ -105,8 +110,10 @@ import accord.topology.TopologyRandomizer;
import accord.utils.Gens;
import accord.utils.Invariants;
import accord.utils.RandomSource;
+import accord.utils.Timestamped;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
+import org.agrona.collections.Int2ObjectHashMap;
import static accord.impl.basic.Cluster.OverrideLinksKind.NONE;
import static accord.impl.basic.Cluster.OverrideLinksKind.RANDOM_BIDIRECTIONAL;
@@ -122,7 +129,7 @@ import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toList;
-public class Cluster implements Scheduler
+public class Cluster
{
public static final Logger trace = LoggerFactory.getLogger("accord.impl.basic.Trace");
@@ -164,6 +171,56 @@ public class Cluster implements Scheduler
}
}
+ public class ClusterScheduler implements Scheduler
+ {
+ final int source;
+
+ ClusterScheduler(int source)
+ {
+ this.source = source;
+ }
+
+ @Override
+ public void now(Runnable run)
+ {
+ run.run();
+ }
+
+ @Override
+ public Scheduled recurring(Runnable run, long delay, TimeUnit units)
+ {
+ return recurring(run, () -> delay, units);
+ }
+
+ @Override
+ public Scheduled once(Runnable run, long delay, TimeUnit units)
+ {
+ RecurringPendingRunnable result = new RecurringPendingRunnable(source, null, run, () -> delay, units, false);
+ pending.add(result, delay, units);
+ return result;
+ }
+
+ @Override
+ public Scheduled selfRecurring(Runnable run, long delay, TimeUnit units)
+ {
+ RecurringPendingRunnable result = new RecurringPendingRunnable(source, null, run, () -> delay, units, true);
+ pending.add(result, delay, units);
+ return result;
+ }
+
+ public Scheduled recurring(Runnable run, LongSupplier delay, TimeUnit units)
+ {
+ RecurringPendingRunnable result = new RecurringPendingRunnable(source, pending, run, delay, units, true);
+ pending.add(result, delay.getAsLong(), units);
+ return result;
+ }
+
+ public void onDone(Runnable run)
+ {
+ Cluster.this.onDone(run);
+ }
+ }
+
final Map<MessageType, Stats> statsMap = new HashMap<>();
final RandomSource random;
@@ -177,7 +234,6 @@ public class Cluster implements Scheduler
final Map<Id, NodeSink> sinks = new HashMap<>();
final MessageListener messageListener;
int clock;
- int recurring;
BiFunction<Id, Id, Link> links;
public Cluster(RandomSource random, MessageListener messageListener, Supplier<PendingQueue> queueSupplier, Runnable checkFailures, Function<Id, Node> lookup, Function<Id, Journal> journalLookup, IntSupplier rf, Consumer<Packet> responseSink)
@@ -224,34 +280,36 @@ public class Cluster implements Scheduler
boolean hasNonRecurring()
{
- boolean hasNonRecurring = false;
+ if (pending.hasNonRecurring())
+ return true;
+
for (Pending p : pending)
{
if (!(p instanceof RecurringPendingRunnable))
continue;
- RecurringPendingRunnable r = (RecurringPendingRunnable) p;
- if (r.requeue.hasNonRecurring())
- {
- hasNonRecurring = true;
- break;
- }
+
+ RecurringPendingRunnable r = (RecurringPendingRunnable) p.origin();
+ if (r.requeue != null && r.requeue.hasNonRecurring())
+ return true;
}
- return hasNonRecurring;
+ return false;
}
public boolean processPending()
{
checkFailures.run();
// All remaining tasks are recurring
- if (recurring > 0 && pending.size() == recurring && !hasNonRecurring())
+ if (!hasNonRecurring())
return false;
Pending next = pending.poll();
if (next == null)
return false;
+ Pending.Global.setActiveOrigin(next);
processNext(next);
+ Pending.Global.clearActiveOrigin();
checkFailures.run();
return true;
@@ -308,49 +366,11 @@ public class Cluster implements Scheduler
trace.trace("{} DROP[{}] (from:{}, to:{}, {}:{}, body:{})", clock++, lookup.apply(to).epoch(), from, to, message instanceof Reply ? "replyTo" : "id", id, message);
}
- @Override
- public Scheduled recurring(Runnable run, long delay, TimeUnit units)
- {
- return recurring(run, () -> delay, units);
- }
-
- @Override
- public Scheduled once(Runnable run, long delay, TimeUnit units)
- {
- RecurringPendingRunnable result = new RecurringPendingRunnable(null, run, () -> delay, units, false);
- pending.add(result, delay, units);
- return result;
- }
-
- @Override
- public Scheduled selfRecurring(Runnable run, long delay, TimeUnit units)
- {
- RecurringPendingRunnable result = new RecurringPendingRunnable(null, run, () -> delay, units, true);
- pending.add(result, delay, units);
- return result;
- }
-
- public Scheduled recurring(Runnable run, LongSupplier delay, TimeUnit units)
- {
- RecurringPendingRunnable result = new RecurringPendingRunnable(pending, run, delay, units, true);
- ++recurring;
- result.onCancellation(() -> --recurring);
- pending.add(result, delay.getAsLong(), units);
- return result;
-
- }
-
public void onDone(Runnable run)
{
onDone.add(run);
}
- @Override
- public void now(Runnable run)
- {
- run.run();
- }
-
// TODO (expected): merge with BurnTest.burn
public static Map<MessageType, Cluster.Stats> run(Supplier<RandomSource> randomSupplier,
List<Node.Id> nodes,
@@ -364,7 +384,7 @@ public class Cluster implements Scheduler
RandomSource rnd = randomSupplier.get();
retryBootstrap = retry -> {
long delay = rnd.nextInt(1, 15);
- queue.add((PendingRunnable) retry::run, delay, TimeUnit.SECONDS);
+ queue.add(PendingRunnable.create(retry::run), delay, TimeUnit.SECONDS);
};
}
IntSupplier coordinationDelays, progressDelays, timeoutDelays;
@@ -376,7 +396,7 @@ public class Cluster implements Scheduler
RandomSource nowRandom = randomSupplier.get();
Supplier<LongSupplier> nowSupplier = () -> {
RandomSource forked = nowRandom.fork();
- // TODO (now): meta-randomise scale of clock drift
+ // TODO (testing): meta-randomise scale of clock drift
return FrequentLargeRange.builder(forked)
.ratio(1, 5)
.small(50, 5000, TimeUnit.MICROSECONDS)
@@ -467,6 +487,7 @@ public class Cluster implements Scheduler
List<Service> services = new ArrayList<>();
for (Id id : nodes)
{
+ ClusterScheduler scheduler = sinks.new ClusterScheduler(id.id);
MessageSink messageSink = sinks.create(id, randomSupplier.get());
LongSupplier nowSupplier = nowSupplierSupplier.get();
LocalConfig localConfig = LocalConfig.DEFAULT;
@@ -478,9 +499,9 @@ public class Cluster implements Scheduler
BurnTestConfigurationService configService = new BurnTestConfigurationService(id, nodeExecutor, randomSupplier, topology, nodeMap::get, topologyUpdates);
BooleanSupplier isLoadedCheck = Gens.supplier(Gens.bools().mixedDistribution().next(random), random);
Node node = new Node(id, messageSink, configService, TimeService.ofNonMonotonic(nowSupplier, MILLISECONDS),
- () -> new ListStore(sinks, random, id), new ShardDistributor.EvenSplit<>(8, ignore -> new PrefixedIntHashKey.Splitter()),
+ () -> new ListStore(scheduler, random, id), new ShardDistributor.EvenSplit<>(8, ignore -> new PrefixedIntHashKey.Splitter()),
nodeExecutor.agent(),
- randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new,
+ randomSupplier.get(), scheduler, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultTimeouts::new,
DefaultProgressLogs::new, DefaultLocalListeners.Factory::new, DelayedCommandStores.factory(sinks.pending, isLoadedCheck, journal), new CoordinationAdapter.DefaultFactory(),
DurableBefore.NOOP_PERSISTER, localConfig);
DurabilityScheduling durability = node.durabilityScheduling();
@@ -516,44 +537,35 @@ public class Cluster implements Scheduler
while (sinks.processPending());
Assertions.assertTrue(startup.isDone());
+ ClusterScheduler clusterScheduler = sinks.new ClusterScheduler(-1);
List<Id> nodesList = new ArrayList<>(Arrays.asList(nodes));
- Scheduled chaos = sinks.recurring(() -> {
+ Scheduled chaos = clusterScheduler.recurring(() -> {
sinks.links = sinks.linkConfig.overrideLinks.apply(nodesList);
if (random.decide(0.1f))
updateDurabilityRate.run();
}, 5L, SECONDS);
- Scheduled reconfigure = sinks.recurring(configRandomizer::maybeUpdateTopology, 1, SECONDS);
-
- Scheduled purge = sinks.recurring(() -> {
- trace.debug("Triggering purge.");
- int numNodes = random.nextInt(1, nodeMap.size());
- for (int i = 0; i < numNodes; i++)
- {
- Id id = random.pick(nodes);
- Node node = nodeMap.get(id);
+ Scheduled reconfigure = clusterScheduler.recurring(configRandomizer::maybeUpdateTopology, 1, SECONDS);
- Journal journal = journalMap.get(node.id());
- CommandStore[] stores = nodeMap.get(node.id()).commandStores().all();
- journal.purge((j) -> stores[j]);
- }
- }, () -> random.nextInt(1, 10), SECONDS);
+ Purge purge = new Purge(clusterScheduler, random, nodesList, nodeMap, journalMap);
- Scheduled restart = sinks.recurring(() -> {
+ Scheduled restart = clusterScheduler.recurring(() -> {
Id id = random.pick(nodes);
CommandStore[] stores = nodeMap.get(id).commandStores().all();
((DelayedCommandStore)stores[0]).unsafeRunIn(() -> {
- Predicate<Pending> pred = getPendingPredicate(stores);
+ Predicate<Pending> pred = getPendingPredicate(id.id, stores);
while (sinks.drain(pred));
// Journal cleanup is a rough equivalent of a node restart.
trace.debug("Triggering journal cleanup for node " + id);
CommandsForKey.disableLinearizabilityViolationsReporting();
ListStore listStore = (ListStore) nodeMap.get(id).commandStores().dataStore();
+ NavigableMap<RoutableKey, Timestamped<int[]>> prevData = listStore.copyOfCurrentData();
listStore.clear();
listStore.restoreFromSnapshot();
Journal journal = journalMap.get(id);
+ Int2ObjectHashMap<NavigableMap<TxnId, Command>> beforeStores = copyCommands(stores);
for (CommandStore s : stores)
{
DelayedCommandStores.DelayedCommandStore store = (DelayedCommandStores.DelayedCommandStore) s;
@@ -562,6 +574,8 @@ public class Cluster implements Scheduler
}
while (sinks.drain(pred));
CommandsForKey.enableLinearizabilityViolationsReporting();
+ Invariants.checkState(listStore.equals(prevData));
+ verifyConsistentRestore(beforeStores, stores);
trace.debug("Done with cleanup.");
});
}, () -> random.nextInt(1, 10), SECONDS);
@@ -612,7 +626,128 @@ public class Cluster implements Scheduler
}
}
- private static Predicate<Pending> getPendingPredicate(CommandStore[] stores)
+ private static class Purge
+ {
+ Scheduled scheduled;
+
+ Purge(Scheduler clusterScheduler, RandomSource rs, List<Id> nodes, Map<Id, Node> nodeMap, Map<Id, Journal> journalMap)
+ {
+ schedule(clusterScheduler, rs, nodes, nodeMap, journalMap);
+ }
+
+ void cancel()
+ {
+ scheduled.cancel();
+ }
+
+ private void schedule(Scheduler clusterScheduler, RandomSource rs, List<Id> nodes, Map<Id, Node> nodeMap, Map<Id, Journal> journalMap)
+ {
+ scheduled = clusterScheduler.selfRecurring(() -> run(clusterScheduler, rs, nodes, nodeMap, journalMap), rs.nextInt(1, 30), SECONDS);
+ }
+
+ private void run(Scheduler clusterScheduler, RandomSource rs, List<Id> nodes, Map<Id, Node> nodeMap, Map<Id, Journal> journalMap)
+ {
+ Id id = rs.pick(nodes);
+ Node node = nodeMap.get(id);
+
+ Journal journal = journalMap.get(node.id());
+ CommandStore[] stores = nodeMap.get(node.id()).commandStores().all();
+ // run on node scheduler so doesn't run during replay
+ scheduled = node.scheduler().selfRecurring(() -> {
+ journal.purge(j -> j < stores.length ? stores[j] : null);
+ schedule(clusterScheduler, rs, nodes, nodeMap, journalMap);
+ }, 0, SECONDS);
+ }
+
+ }
+
+ private static Int2ObjectHashMap<NavigableMap<TxnId, Command>> copyCommands(CommandStore[] stores)
+ {
+ Int2ObjectHashMap<NavigableMap<TxnId, Command>> result = new Int2ObjectHashMap<>();
+ for (CommandStore s : stores)
+ {
+ DelayedCommandStores.DelayedCommandStore store = (DelayedCommandStores.DelayedCommandStore) s;
+ NavigableMap<TxnId, Command> commands = new TreeMap<>();
+ result.put(store.id(), commands);
+ for (Map.Entry<TxnId, GlobalCommand> e : store.unsafeCommands().entrySet())
+ commands.put(e.getKey(), e.getValue().value());
+ }
+ return result;
+ }
+
+ private static void verifyConsistentRestore(Int2ObjectHashMap<NavigableMap<TxnId, Command>> beforeStores, CommandStore[] stores)
+ {
+ for (CommandStore s : stores)
+ {
+ DelayedCommandStores.DelayedCommandStore store = (DelayedCommandStores.DelayedCommandStore) s;
+ NavigableMap<TxnId, Command> before = beforeStores.get(store.id());
+ for (Map.Entry<TxnId, GlobalCommand> e : store.unsafeCommands().entrySet())
+ {
+ Command beforeCommand = before.get(e.getKey());
+ Command afterCommand = e.getValue().value();
+ if (beforeCommand == null)
+ {
+ Invariants.checkArgument(afterCommand.is(Status.NotDefined));
+ continue;
+ }
+ if (afterCommand.hasBeen(Status.Truncated))
+ {
+ if (afterCommand.is(Status.Invalidated))
+ Invariants.checkState(beforeCommand.hasBeen(Status.Truncated) || !beforeCommand.hasBeen(Status.PreCommitted)
+ && store.unsafeGetRedundantBefore().max(beforeCommand.participants().touches(), RedundantBefore.Entry::shardRedundantBefore).compareTo(beforeCommand.txnId()) >= 0);
+ continue;
+ }
+ if (beforeCommand.hasBeen(Status.Truncated))
+ {
+ Invariants.checkState(!beforeCommand.is(Status.Invalidated) || afterCommand.is(Status.Invalidated));
+ Invariants.checkState(beforeCommand.is(Status.Invalidated) || afterCommand.is(Status.Truncated) || afterCommand.is(Status.Applied));
+ continue;
+ }
+ Invariants.checkState(isConsistent(beforeCommand.saveStatus(), afterCommand.saveStatus())
+ && beforeCommand.executeAtOrTxnId().equals(afterCommand.executeAtOrTxnId())
+ && beforeCommand.acceptedOrCommitted().equals(afterCommand.acceptedOrCommitted())
+ && beforeCommand.promised().equals(afterCommand.promised())
+ && beforeCommand.durability().equals(afterCommand.durability()));
+ }
+ if (before.size() > store.unsafeCommands().size())
+ {
+ for (TxnId txnId : before.keySet())
+ {
+ if (!store.unsafeCommands().containsKey(txnId))
+ {
+ Command beforeCommand = before.get(txnId);
+ if (beforeCommand.saveStatus() == SaveStatus.Erased)
+ continue;
+
+ if (store.unsafeGetRedundantBefore().min(beforeCommand.participants().owns(), RedundantBefore.Entry::shardRedundantBefore).compareTo(txnId) > 0)
+ continue;
+
+ Invariants.checkState(false);
+ }
+ }
+ }
+ }
+ }
+
+ private static boolean isConsistent(SaveStatus before, SaveStatus after)
+ {
+ if (before == after)
+ return true;
+
+ if (before == SaveStatus.Uninitialised || before == SaveStatus.NotDefined)
+ return after == SaveStatus.Uninitialised || after == SaveStatus.NotDefined;
+
+ // depending on arrival order, an unmanaged txn may be ready to execute immediately or have to wait for another transaction to commit
+ if (before == SaveStatus.Stable || before == SaveStatus.ReadyToExecute)
+ return after == SaveStatus.Stable || after == SaveStatus.ReadyToExecute;
+
+ if (before == SaveStatus.PreApplied || before == SaveStatus.Applying || before == SaveStatus.Applied)
+ return after == SaveStatus.PreApplied || after == SaveStatus.Applying || after == SaveStatus.Applied;
+
+ return false;
+ }
+
+ private static Predicate<Pending> getPendingPredicate(int nodeId, CommandStore[] stores)
{
Set<CommandStore> nodeStores = new HashSet<>(Arrays.asList(stores));
return item -> {
@@ -621,6 +756,12 @@ public class Cluster implements Scheduler
DelayedCommandStore.DelayedTask<?> task = (DelayedCommandStore.DelayedTask<?>) item;
if (nodeStores.contains(task.parent()))
return true;
+ item = item.origin();
+ }
+ if (item instanceof RecurringPendingRunnable)
+ {
+ RecurringPendingRunnable recurring = (RecurringPendingRunnable) item;
+ return recurring.source == nodeId && (!recurring.isRecurring || recurring.origin() != recurring);
}
return false;
};
diff --git a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
index e5cf16d5..04ac085d 100644
--- a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
+++ b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
@@ -29,12 +29,8 @@ import java.util.function.BiConsumer;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Function;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.google.common.collect.Iterables;
import accord.api.Agent;
import accord.api.DataStore;
@@ -74,40 +70,6 @@ import static accord.utils.Invariants.ParanoiaCostFactor.HIGH;
public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
{
- private static final Logger logger = LoggerFactory.getLogger(DelayedCommandStores.class);
-
- //TODO (correctness): remove once we have a Journal integration that does not change the values of a command...
- // There are known issues due to Journal, so exclude known problamtic fields
- private static class ValidationHack
- {
- private static Pattern field(String path)
- {
- path = path.replace(".", "\\.");
- path += ".*";
- return Pattern.compile(path);
- }
-
- private static final List<Pattern> KNOWN_ISSUES = List.of(
- // when a new epoch is detected and the execute ranges have more than the coordinating ranges,
- // and the coordinating ranges doesn't include the home key... we drop the query...
- // The logic to stitch messages together is not able to handle this as it doesn't know the original Topologies
- // TODO (required): test int if this is still true
-// field(".partialTxn.query."),
- // cmd.mutable().build() != cmd. This is due to Command.durability changing NotDurable to Local depending on the status
- field(".durability."));
- }
-
- private static boolean hasKnownIssue(String path)
- {
- for (Pattern p : ValidationHack.KNOWN_ISSUES)
- {
- Matcher m = p.matcher(path);
- if (m.find())
- return true;
- }
- return false;
- }
-
private DelayedCommandStores(NodeCommandStoreService time, Agent agent, DataStore store, RandomSource random, ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenersFactory, SimulatedDelayedExecutorService executorService, BooleanSupplier isLoadedCheck, Journal journal)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, listenersFactory, DelayedCommandStore.factory(executorService, isLoadedCheck, journal));
@@ -193,8 +155,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
// Journal will not have result persisted. This part is here for test purposes and ensuring that we have strict object equality.
Command reconstructed = journal.reconstruct(id, current.txnId());
List<Difference<?>> diff = ReflectionUtils.recursiveEquals(current, reconstructed);
- List<String> filteredDiff = diff.stream().filter(d -> !DelayedCommandStores.hasKnownIssue(d.path)).map(Object::toString).collect(Collectors.toList());
- Invariants.checkState(filteredDiff.isEmpty(), "Commands did not match: expected %s, given %s, node %s, store %d, diff %s", current, reconstructed, node, id(), new LazyToString(() -> String.join("\n", filteredDiff)));
+ Invariants.checkState(diff.isEmpty(), "Commands did not match: expected %s, given %s, node %s, store %d, diff %s", current, reconstructed, node, id(), new LazyToString(() -> String.join("\n", Iterables.transform(diff, Object::toString))));
}
@Override
diff --git a/accord-core/src/test/java/accord/impl/basic/Journal.java b/accord-core/src/test/java/accord/impl/basic/Journal.java
index 66052112..27f9cb19 100644
--- a/accord-core/src/test/java/accord/impl/basic/Journal.java
+++ b/accord-core/src/test/java/accord/impl/basic/Journal.java
@@ -18,7 +18,9 @@
package accord.impl.basic;
+import java.util.AbstractList;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -53,6 +55,8 @@ import accord.primitives.Writes;
import accord.utils.Invariants;
import org.agrona.collections.Long2ObjectHashMap;
+import static accord.primitives.SaveStatus.Erased;
+import static accord.primitives.SaveStatus.NotDefined;
import static accord.primitives.SaveStatus.Stable;
import static accord.primitives.Status.Invalidated;
import static accord.primitives.Status.Truncated;
@@ -76,6 +80,8 @@ public class Journal
int commandStoreId = e.getKey().intValue();
Map<TxnId, List<Diff>> localJournal = e.getValue();
CommandStore store = storeSupplier.apply(commandStoreId);
+ if (store == null)
+ continue;
Map<TxnId, List<Diff>> updates = new HashMap<>();
List<TxnId> removals = new ArrayList<>();
@@ -94,55 +100,114 @@ public class Journal
case INVALIDATE:
case TRUNCATE_WITH_OUTCOME:
case TRUNCATE:
+ case ERASE:
+ if (commandStoreId == 0 && id.id == 1 && txnId.toString().equals("[7,12610,9(RX),2]"))
+ cleanup = Cleanup.shouldCleanup(store.agent(), command, store.unsafeGetRedundantBefore(), store.durableBefore());
command = Commands.purge(command, command.participants(), cleanup);
Invariants.checkState(command.saveStatus() != SaveStatus.Uninitialised);
- List<Diff> arr = new ArrayList<>();
- arr.add(diff(null, command));
- updates.put(txnId, arr);
+ Diff diff = diff(null, command);
+ e2.setValue(cleanup == Cleanup.ERASE ? new ErasedList(diff) : new TruncatedList(diff));
break;
- case ERASE:
- removals.add(txnId);
+
+ case EXPUNGE:
+ e2.setValue(new PurgedList());
break;
}
}
-
- for (TxnId removal : removals)
- localJournal.remove(removal);
- localJournal.putAll(updates);
}
}
- // TODO (required): this might be a good first approximation, but maybe we need to make a better distinction between
- // when we want to produce side-effects.
- private boolean loading = false;
-
public void reconstructAll(InMemoryCommandStore.Loader loader, int commandStoreId)
{
- Map<TxnId, List<Diff>> diffs = diffsPerCommandStore.get(commandStoreId);
-
// Nothing to do here, journal is empty for this command store
- if (diffs == null)
+ if (!diffsPerCommandStore.containsKey(commandStoreId))
return;
- loading = true;
- try
+ // copy to avoid concurrent modification when appending to journal
+ Map<TxnId, List<Diff>> diffs = new TreeMap<>(diffsPerCommandStore.get(commandStoreId));
+ for (Map.Entry<TxnId, List<Diff>> e : diffs.entrySet())
+ e.setValue(new ArrayList<>(e.getValue()));
+
+ List<Command> toApply = new ArrayList<>();
+ for (Map.Entry<TxnId, List<Diff>> e : diffs.entrySet())
{
- List<Command> toApply = new ArrayList<>();
- for (TxnId txnId : diffs.keySet())
- {
- Command command = reconstruct(commandStoreId, txnId);
- if (command.saveStatus().compareTo(Stable) >= 0 && !command.hasBeen(Truncated))
- toApply.add(command);
- loader.load(command);
- }
+ if (e.getValue().isEmpty()) continue;
+ Command command = reconstruct(commandStoreId, e.getKey(), e.getValue());
+ if (command.saveStatus().compareTo(Stable) >= 0 && !command.hasBeen(Truncated))
+ toApply.add(command);
+ loader.load(command);
+ }
+
+ toApply.sort(Comparator.comparing(Command::executeAt));
+ for (Command command : toApply)
+ loader.apply(command);
+ }
+
+ static class ErasedList extends AbstractList<Diff>
+ {
+ final Diff erased;
+
+ ErasedList(Diff erased)
+ {
+ Invariants.checkArgument(erased.saveStatus.value == SaveStatus.Erased);
+ this.erased = erased;
+ }
+
+ @Override
+ public Diff get(int index)
+ {
+ if (index != 0)
+ throw new IndexOutOfBoundsException();
+ return erased;
+ }
+
+ @Override
+ public int size()
+ {
+ return 1;
+ }
- toApply.sort(Comparator.comparing(Command::executeAt));
- for (Command command : toApply)
- loader.apply(command);
+ @Override
+ public boolean add(Diff diff)
+ {
+ if (diff.saveStatus != null && diff.saveStatus.value == SaveStatus.Erased)
+ return false;
+ throw illegalState();
}
- finally
+ }
+
+ static class TruncatedList extends ArrayList<Diff>
+ {
+ TruncatedList(Diff truncated)
{
- loading = false;
+ add(truncated);
+ }
+ }
+
+ static class PurgedList extends AbstractList<Diff>
+ {
+ PurgedList()
+ {
+ }
+
+ @Override
+ public Diff get(int index)
+ {
+ throw new IndexOutOfBoundsException();
+ }
+
+ @Override
+ public int size()
+ {
+ return 0;
+ }
+
+ @Override
+ public boolean add(Diff diff)
+ {
+ if (diff.saveStatus != null && diff.saveStatus.value == SaveStatus.Erased)
+ return false;
+ throw illegalState();
}
}
@@ -154,7 +219,11 @@ public class Journal
public Command reconstruct(int commandStoreId, TxnId txnId)
{
- List<Diff> diffs = this.diffsPerCommandStore.get(commandStoreId).get(txnId);
+ return reconstruct(commandStoreId, txnId, diffsPerCommandStore.get(commandStoreId).get(txnId));
+ }
+
+ public Command reconstruct(int commandStoreId, TxnId txnId, List<Diff> diffs)
+ {
return reconstruct(diffs, Reconstruct.Last).get(0);
}
@@ -167,7 +236,7 @@ public class Journal
TxnId txnId = null;
Timestamp executeAt = null;
Timestamp executesAtLeast = null;
- SaveStatus saveStatus = null;
+ SaveStatus saveStatus = NotDefined;
Status.Durability durability = Status.Durability.NotDurable;
Ballot acceptedOrCommitted = Ballot.ZERO;
@@ -185,7 +254,7 @@ public class Journal
{
Diff diff = diffs.get(i);
if (diff.txnId != null)
- txnId = diff.txnId.get();
+ txnId = diff.txnId;
if (diff.executeAt != null)
executeAt = diff.executeAt.get();
if (diff.executesAtLeast != null)
@@ -229,6 +298,7 @@ public class Journal
{
t.printStackTrace();
}
+
switch (saveStatus.known.outcome)
{
case Erased:
@@ -329,15 +399,9 @@ public class Journal
public void onExecute(int commandStoreId, Command before, Command after, boolean isPrimary)
{
- if (loading || (before == null && after == null))
+ if (before == null && after == null)
return;
- if (after.saveStatus() == SaveStatus.Erased)
- {
- diffsPerCommandStore.computeIfAbsent(commandStoreId, (k) -> new TreeMap<>())
- .remove(after.txnId());
- return;
- }
Diff diff = diff(before, after);
if (!isPrimary)
diff = diff.asNonPrimary();
@@ -352,7 +416,7 @@ public class Journal
private static class Diff
{
- public final NewValue<TxnId> txnId;
+ public final TxnId txnId;
public final NewValue<Timestamp> executeAt;
public final NewValue<Timestamp> executesAtLeast;
@@ -371,7 +435,7 @@ public class Journal
public final NewValue<Result> result; // temporarily here for sakes for reloads
- public Diff(NewValue<TxnId> txnId,
+ public Diff(TxnId txnId,
NewValue<Timestamp> executeAt,
NewValue<Timestamp> executesAtLeast,
NewValue<SaveStatus> saveStatus,
@@ -470,7 +534,7 @@ public class Journal
if (Objects.equals(before, after))
return null;
- Diff diff = new Diff(ifNotEqual(before, after, Command::txnId, false),
+ Diff diff = new Diff(after.txnId(),
ifNotEqual(before, after, Command::executeAt, true),
ifNotEqual(before, after, Command::executesAtLeast, true),
ifNotEqual(before, after, Command::saveStatus, false),
@@ -543,6 +607,21 @@ public class Journal
{
return "" + value;
}
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ NewValue<?> newValue = (NewValue<?>) o;
+ return Objects.equals(value, newValue.value);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ throw new UnsupportedOperationException();
+ }
}
}
\ No newline at end of file
diff --git a/accord-core/src/test/java/accord/impl/basic/NodeSink.java b/accord-core/src/test/java/accord/impl/basic/NodeSink.java
index bfa4b3c9..8bd61a03 100644
--- a/accord-core/src/test/java/accord/impl/basic/NodeSink.java
+++ b/accord-core/src/test/java/accord/impl/basic/NodeSink.java
@@ -79,14 +79,14 @@ public class NodeSink implements MessageSink
callbacks.put(messageId, sc);
if (maybeEnqueue(to, messageId, send, sc))
{
- parent.pending.add((PendingRunnable) () -> {
+ parent.pending.add(PendingRunnable.create(() -> {
if (sc == callbacks.get(messageId))
sc.slowResponse(to);
- }, 100 + random.nextInt(200), MILLISECONDS);
- parent.pending.add((PendingRunnable) () -> {
+ }), 100 + random.nextInt(200), MILLISECONDS);
+ parent.pending.add(PendingRunnable.create(() -> {
if (sc == callbacks.remove(messageId))
sc.timeout(to);
- }, 1000 + random.nextInt(1000), MILLISECONDS);
+ }), 1000 + random.nextInt(1000), MILLISECONDS);
}
}
@@ -120,12 +120,12 @@ public class NodeSink implements MessageSink
parent.notifyDropped(self, to, id, message);
if (callback != null)
{
- parent.pending.add((PendingRunnable) () -> {
+ parent.pending.add(PendingRunnable.create(() -> {
if (callback == callbacks.remove(id))
{
try
{
- // TODO (now): we MUST drop any ACTUAL reply now
+ // TODO (required): we MUST drop any ACTUAL reply now
callback.failure(to, new SimulatedFault("Simulation Failure; src=" + self + ", to=" + to + ", id=" + id + ", message=" + message));
}
catch (Throwable t)
@@ -134,11 +134,11 @@ public class NodeSink implements MessageSink
lookup.apply(self).agent().onUncaughtException(t);
}
}
- }, 1000 + random.nextInt(1000), MILLISECONDS);
+ }), 1000 + random.nextInt(1000), MILLISECONDS);
}
return false;
case DROP:
- // TODO (consistency): parent.notifyDropped is a trace logger that is very similar in spirit to MessageListener; can we unify?
+ // TODO (desired): parent.notifyDropped is a trace logger that is very similar in spirit to MessageListener; can we unify?
parent.notifyDropped(self, to, id, message);
return true;
default:
diff --git a/accord-core/src/test/java/accord/impl/basic/Packet.java b/accord-core/src/test/java/accord/impl/basic/Packet.java
index dc7fd567..698552b6 100644
--- a/accord-core/src/test/java/accord/impl/basic/Packet.java
+++ b/accord-core/src/test/java/accord/impl/basic/Packet.java
@@ -29,6 +29,7 @@ public class Packet implements Pending, ReplyContext
{
static final int SENTINEL_MESSAGE_ID = Integer.MIN_VALUE;
+ final Pending origin = Pending.Global.activeOrigin();
public final Id src;
public final Id dst;
public final long requestId; // if message is Reply, this is the id of the message we are replying to
@@ -53,6 +54,12 @@ public class Packet implements Pending, ReplyContext
this.message = reply;
}
+ @Override
+ public Pending origin()
+ {
+ return origin;
+ }
+
@Override
public String toString()
{
diff --git a/accord-core/src/test/java/accord/impl/basic/Pending.java b/accord-core/src/test/java/accord/impl/basic/Pending.java
index 7aadb5c9..8acfd5b4 100644
--- a/accord-core/src/test/java/accord/impl/basic/Pending.java
+++ b/accord-core/src/test/java/accord/impl/basic/Pending.java
@@ -18,6 +18,28 @@
package accord.impl.basic;
+import accord.utils.Invariants;
+
public interface Pending
{
+ Pending origin();
+
+ // a slightly hacky but easy way to track transitive recurring tasks
+ class Global
+ {
+ private static Pending activeOrigin;
+ public static void setActiveOrigin(Pending newActive)
+ {
+ Invariants.checkState(activeOrigin == null);
+ activeOrigin = newActive.origin();
+ }
+ public static void clearActiveOrigin()
+ {
+ activeOrigin = null;
+ }
+ public static Pending activeOrigin()
+ {
+ return activeOrigin;
+ }
+ }
}
diff --git a/accord-core/src/test/java/accord/impl/basic/PendingRunnable.java b/accord-core/src/test/java/accord/impl/basic/PendingRunnable.java
index 0ef0c989..15a0eb7f 100644
--- a/accord-core/src/test/java/accord/impl/basic/PendingRunnable.java
+++ b/accord-core/src/test/java/accord/impl/basic/PendingRunnable.java
@@ -20,4 +20,22 @@ package accord.impl.basic;
public interface PendingRunnable extends Pending, Runnable
{
+ static PendingRunnable create(Runnable run)
+ {
+ Pending origin = Global.activeOrigin();
+ return new PendingRunnable()
+ {
+ @Override
+ public Pending origin()
+ {
+ return origin;
+ }
+
+ @Override
+ public void run()
+ {
+ run.run();
+ }
+ };
+ }
}
diff --git a/accord-core/src/test/java/accord/impl/basic/RandomDelayQueue.java b/accord-core/src/test/java/accord/impl/basic/RandomDelayQueue.java
index a90f150a..3708c8cf 100644
--- a/accord-core/src/test/java/accord/impl/basic/RandomDelayQueue.java
+++ b/accord-core/src/test/java/accord/impl/basic/RandomDelayQueue.java
@@ -32,6 +32,7 @@ import accord.impl.basic.DelayedCommandStores.DelayedCommandStore.DelayedTask;
import accord.utils.DefaultRandom;
import accord.utils.RandomSource;
+import static accord.impl.basic.RecurringPendingRunnable.isRecurring;
import static accord.utils.Invariants.illegalArgument;
import static accord.utils.Invariants.illegalState;
@@ -169,7 +170,9 @@ public class RandomDelayQueue implements PendingQueue
for (Item item : items)
{
if (toDrain.test(item.item))
+ {
ret.add(item.item);
+ }
else
{
queue.add(item);
@@ -301,9 +304,4 @@ public class RandomDelayQueue implements PendingQueue
this.seed = seed;
}
}
-
- private static boolean isRecurring(Pending pending)
- {
- return pending instanceof RecurringPendingRunnable && ((RecurringPendingRunnable) pending).isRecurring;
- }
}
diff --git a/accord-core/src/test/java/accord/impl/basic/RecurringPendingRunnable.java b/accord-core/src/test/java/accord/impl/basic/RecurringPendingRunnable.java
index 6dbfd754..257f6387 100644
--- a/accord-core/src/test/java/accord/impl/basic/RecurringPendingRunnable.java
+++ b/accord-core/src/test/java/accord/impl/basic/RecurringPendingRunnable.java
@@ -25,6 +25,7 @@ import accord.api.Scheduler.Scheduled;
class RecurringPendingRunnable implements PendingRunnable, Scheduled
{
+ final int source;
final PendingQueue requeue;
final LongSupplier delay;
final TimeUnit units;
@@ -32,8 +33,9 @@ class RecurringPendingRunnable implements PendingRunnable, Scheduled
Runnable run;
Runnable onCancellation;
- RecurringPendingRunnable(PendingQueue requeue, Runnable run, LongSupplier delay, TimeUnit units, boolean isRecurring)
+ RecurringPendingRunnable(int source, PendingQueue requeue, Runnable run, LongSupplier delay, TimeUnit units, boolean isRecurring)
{
+ this.source = source;
this.requeue = requeue;
this.run = run;
this.delay = delay;
@@ -41,6 +43,12 @@ class RecurringPendingRunnable implements PendingRunnable, Scheduled
this.isRecurring = isRecurring;
}
+ @Override
+ public Pending origin()
+ {
+ return this;
+ }
+
@Override
public void run()
{
@@ -87,4 +95,10 @@ class RecurringPendingRunnable implements PendingRunnable, Scheduled
return run + " with " + delay + " " + units + " delay";
}
+
+ public static boolean isRecurring(Pending pending)
+ {
+ Pending origin = pending.origin();
+ return origin instanceof RecurringPendingRunnable && ((RecurringPendingRunnable) origin).isRecurring;
+ }
}
diff --git a/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java b/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
index 0e733095..286d34da 100644
--- a/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
+++ b/accord-core/src/test/java/accord/impl/basic/TaskExecutorService.java
@@ -34,11 +34,17 @@ public abstract class TaskExecutorService extends AbstractExecutorService implem
{
public static class Task<T> extends AsyncResults.RunnableResult<T> implements Pending, RunnableFuture<T>
{
+ final Pending origin = Pending.Global.activeOrigin();
public Task(Callable<T> fn)
{
super(fn);
}
+ public Pending origin()
+ {
+ return origin;
+ }
+
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
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 bb1b04c8..6fcfb181 100644
--- a/accord-core/src/test/java/accord/impl/list/ListAgent.java
+++ b/accord-core/src/test/java/accord/impl/list/ListAgent.java
@@ -152,7 +152,7 @@ public class ListAgent implements Agent
@Override
public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
{
- // TODO (required): meta randomise
+ // TODO (testing): meta randomise
return units.convert(rnd.nextInt(100, 1000), MILLISECONDS);
}
@@ -172,7 +172,7 @@ public class ListAgent implements Agent
@Override
public long expiresAt(ReplyContext replyContext, TimeUnit unit)
{
- // TODO (required): implement this (add it to the replyContext)
+ // TODO (testing): implement this (add it to the replyContext)
return -1;
}
diff --git a/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java b/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
index c7ffc2c6..340b6077 100644
--- a/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
+++ b/accord-core/src/test/java/accord/impl/list/ListFetchCoordinator.java
@@ -65,7 +65,7 @@ public class ListFetchCoordinator extends AbstractFetchCoordinator
ListData listData = (ListData) data;
persisting.add(commandStore.execute(PreLoadContext.empty(), safeStore -> {
listData.forEach((key, value) -> listStore.data.merge(key, value, Timestamped::merge));
- }).flatMap(ignore -> listStore.snapshot(received, syncPoint.syncId)).addCallback((success, fail) -> {
+ }).flatMap(ignore -> listStore.snapshot(true, received, syncPoint.syncId)).addCallback((success, fail) -> {
if (fail == null) success(from, received);
else fail(from, received, fail);
}).beginAsResult());
diff --git a/accord-core/src/test/java/accord/impl/list/ListRequest.java b/accord-core/src/test/java/accord/impl/list/ListRequest.java
index 46ac4c2b..acb9270e 100644
--- a/accord-core/src/test/java/accord/impl/list/ListRequest.java
+++ b/accord-core/src/test/java/accord/impl/list/ListRequest.java
@@ -226,7 +226,15 @@ public class ListRequest implements Request
node.reply(client, replyContext, ListResult.lost(client, ((Packet)replyContext).requestId, txnId), null);
break;
case Other:
- node.reply(client, replyContext, ListResult.other(client, ((Packet)replyContext).requestId, txnId), null);
+ if (attempt < 100)
+ {
+ node.reply(client, replyContext, ListResult.heartBeat(client, ((Packet)replyContext).requestId, txnId), null);
+ node.scheduler().once(() -> checkOnResult(finalHomeKey, txnId, attempt + 1, null), 5L, TimeUnit.MINUTES);
+ }
+ else
+ {
+ node.reply(client, replyContext, ListResult.other(client, ((Packet)replyContext).requestId, txnId), null);
+ }
break;
default:
node.agent().onUncaughtException(new AssertionError("Unknown outcome: " + s));
diff --git a/accord-core/src/test/java/accord/impl/list/ListResult.java b/accord-core/src/test/java/accord/impl/list/ListResult.java
index c9da1719..adfa1fe3 100644
--- a/accord-core/src/test/java/accord/impl/list/ListResult.java
+++ b/accord-core/src/test/java/accord/impl/list/ListResult.java
@@ -155,7 +155,7 @@ public class ListResult implements Result, Reply
+ "requestId:" + requestId + ", "
+ "txnId:" + txnId + ", "
+ (responseKeys == null
- ? (read == null ? "invalidated!}" : read.length == 0 ? "unknown}" : "lost}")
+ ? status.toString() + '}'
: "reads:" + IntStream.range(0, responseKeys.size())
.mapToObj(i -> responseKeys.get(i) + ":" + Arrays.toString(read[i]))
.collect(Collectors.joining(", ", "{", "}")) + ", "
diff --git a/accord-core/src/test/java/accord/impl/list/ListStore.java b/accord-core/src/test/java/accord/impl/list/ListStore.java
index 3671e3b6..179d00d1 100644
--- a/accord-core/src/test/java/accord/impl/list/ListStore.java
+++ b/accord-core/src/test/java/accord/impl/list/ListStore.java
@@ -22,6 +22,7 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Deque;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@@ -179,27 +180,44 @@ public class ListStore implements DataStore
private static final class PendingSnapshot
{
+ // we don't restart bootstraps after clearing journal state and replaying, so must finish snapshots
+ final boolean runBeforeRestart;
final long delay;
final Consumer<Boolean> onCompletion;
- private PendingSnapshot(long delay, Consumer<Boolean> onCompletion)
+ private PendingSnapshot(boolean runBeforeRestart, long delay, Consumer<Boolean> onCompletion)
{
+ this.runBeforeRestart = runBeforeRestart;
this.delay = delay;
this.onCompletion = onCompletion;
}
}
private Snapshot snapshot;
+ private Scheduler.Scheduled scheduled;
private final Deque<PendingSnapshot> pendingSnapshots = new ArrayDeque<>();
private long pendingDelay = 0;
+ @Override
public AsyncResult<Void> snapshot(Ranges ranges, TxnId before)
+ {
+ return snapshot(false, ranges, before);
+ }
+
+ public AsyncResult<Void> snapshot(boolean runBeforeRestart, Ranges ranges, TxnId before)
{
Snapshot snapshot = new Snapshot(data, addedAts, removedAts, purgedAts, fetchCompletes, pendingRemoves);
AsyncResult.Settable<Void> result = new AsyncResults.SettableResult<>();
long delay = Math.max(1, random.nextBiasedLong(100, 1000, 5000) - pendingDelay);
pendingDelay += delay;
- pendingSnapshots.add(new PendingSnapshot(delay, success -> {
+
+ if (scheduled != null && runBeforeRestart && !pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart))
+ {
+ scheduled.cancel();
+ scheduled = null;
+ }
+
+ pendingSnapshots.add(new PendingSnapshot(runBeforeRestart, delay, success -> {
if (success)
{
this.snapshot = snapshot;
@@ -211,7 +229,7 @@ public class ListStore implements DataStore
}
}));
- if (pendingSnapshots.size() == 1)
+ if (scheduled == null)
scheduleRunSnapshot();
return result;
}
@@ -219,7 +237,9 @@ public class ListStore implements DataStore
private void scheduleRunSnapshot()
{
Invariants.checkState(!pendingSnapshots.isEmpty());
- scheduler.once(() -> {
+ // schedule as recurring so that we don't run them
+ Runnable run = () -> {
+ scheduled = null;
if (pendingSnapshots.isEmpty())
return;
@@ -228,7 +248,10 @@ public class ListStore implements DataStore
pendingDelay -= pendingSnapshot.delay;
if (!pendingSnapshots.isEmpty())
scheduleRunSnapshot();
- }, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
+ };
+
+ if (pendingSnapshots.stream().anyMatch(p -> p.runBeforeRestart)) scheduled = scheduler.once(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
+ else scheduled = scheduler.selfRecurring(run, pendingSnapshots.peekFirst().delay, TimeUnit.MILLISECONDS);
}
public void restoreFromSnapshot()
@@ -297,10 +320,16 @@ public class ListStore implements DataStore
private void checkAccess(Timestamp executeAt, Key key)
{
if (!allowed.contains(key))
- throw new IllegalStateException(String.format("Attempted to access key %s on node %s, which is not in the range %s;\nexecuteAt = %s\n%s",
- key, node, allowed,
- executeAt,
- history(key)));
+ {
+ // TODO (expected): improve this validation logic
+ // but in the meantime, whitelist valid things, e.g. ephemeral reads can access data that has been retired
+ if (executeAt instanceof TxnId && ((TxnId) executeAt).awaitsOnlyDeps()
+ && removedAts.stream().anyMatch(r -> r.ranges.contains(key) && r.epoch > executeAt.epoch()))
+ return;
+
+ throw illegalState("Attempted to access key %s on node %s, which is not in the range %s;\nexecuteAt = %s\n%s",
+ key, node, allowed, executeAt, history(key));
+ }
}
private void checkAccess(Timestamp executeAt, Range range)
@@ -462,6 +491,34 @@ public class ListStore implements DataStore
return Timestamped.mergeEqual(a, b, Arrays::equals);
}
+ public NavigableMap<RoutableKey, Timestamped<int[]>> copyOfCurrentData()
+ {
+ return new TreeMap<>(data);
+ }
+
+ public boolean equals(NavigableMap<RoutableKey, Timestamped<int[]>> a)
+ {
+ return equal(a, data);
+ }
+
+ public static boolean equal(NavigableMap<RoutableKey, Timestamped<int[]>> a, NavigableMap<RoutableKey, Timestamped<int[]>> b)
+ {
+ Iterator<Map.Entry<RoutableKey, Timestamped<int[]>>> aiter = a.entrySet().iterator();
+ Iterator<Map.Entry<RoutableKey, Timestamped<int[]>>> biter = b.entrySet().iterator();
+ while (aiter.hasNext() && biter.hasNext())
+ {
+ Map.Entry<RoutableKey, Timestamped<int[]>> av = aiter.next();
+ Map.Entry<RoutableKey, Timestamped<int[]>> bv = biter.next();
+ if (!av.getKey().equals(bv.getKey()))
+ return false;
+ if (!av.getValue().equals(bv.getValue(), Arrays::equals))
+ return false;
+ }
+ if (aiter.hasNext() || biter.hasNext())
+ return false;
+ return true;
+ }
+
private static boolean isStrictPrefix(int[] a, int[] b)
{
if (a.length >= b.length)
@@ -531,7 +588,7 @@ public class ListStore implements DataStore
private void runWhenReady(Node node, long epoch, Runnable whenKnown)
{
if (node.topology().epoch() >= epoch) whenKnown.run();
- else node.scheduler().once(() -> runWhenReady(node, epoch, whenKnown), 10, TimeUnit.SECONDS);
+ else node.scheduler().selfRecurring(() -> runWhenReady(node, epoch, whenKnown), 10, TimeUnit.SECONDS);
}
private void removeLocalWhenReady(Node node, long epoch, Ranges removed)
@@ -567,16 +624,9 @@ public class ListStore implements DataStore
pendingRemoves.remove(epoch);
this.allowed = this.allowed.without(removed);
purgedAts.add(new PurgeAt(s, epoch, removed));
- // C* encodes keyspace/table within a Range, so Ranges being added/removed to the cluster are expected behaviors and not just local range movements. To better simulate that
- // this logic attempts to purge the data once we know its "safe" (no read/writes pending).
- // The assumption in the C* case is that we will be able to do similar logic when integrating with Transactional Cluster Metadata (TCM). The case there is that Epoch is owned by TCM
- // and C* will start applying changes outside the context of Accord (hence why this logic uses callbacks before Accord sees the topology), so there exists an assumption: we will be able
- // to delay local removal of data (drop keyspace/table) on accord txn state!
- for (Range range : removed)
- {
- NavigableMap<RoutableKey, Timestamped<int[]>> historicData = data.subMap(range.start(), range.startInclusive(), range.end(), range.endInclusive());
- historicData.clear();
- }
+ // C* encodes keyspace/table within a Range, so Ranges being added/removed to the cluster are expected behaviors and not just local range movements.
+ // however, there is no reason to erase old data as it being used should be detected as a violation due to being stale,
+ // and clearing data unnecessarily complicates journal replay
List<Runnable> callbacks = new ArrayList<>(onRemovalDone);
onRemovalDone.clear();
diff --git a/accord-core/src/test/java/accord/impl/mock/MockCluster.java b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
index c125f9e1..7604548b 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockCluster.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
@@ -107,7 +107,11 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
@Override
public void close()
{
- nodes.values().forEach(Node::shutdown);
+ nodes.values().forEach(n -> {
+ n.shutdown();
+ ((ThreadPoolScheduler)n.scheduler()).stop();
+ });
+
}
private synchronized Id nextNodeId()
diff --git a/accord-core/src/test/java/accord/local/CommandsTest.java b/accord-core/src/test/java/accord/local/CommandsTest.java
index 073b2f7b..a823e0aa 100644
--- a/accord-core/src/test/java/accord/local/CommandsTest.java
+++ b/accord-core/src/test/java/accord/local/CommandsTest.java
@@ -83,7 +83,7 @@ class CommandsTest
{
{
// the node selected does not matter and should not impact determinism, they all share the same scheduler
- ((Cluster) nodeMap.values().stream().findFirst().get().scheduler()).onDone(() -> checkState(0, nodeMap.values()));
+ ((Cluster.ClusterScheduler) nodeMap.values().stream().findFirst().get().scheduler()).onDone(() -> checkState(0, nodeMap.values()));
}
@Override
@@ -142,6 +142,9 @@ class CommandsTest
if (cmd.value() == null) // empty isn't public
continue;
Command command = cmd.value();
+ if (command.txnId().isSyncPoint())
+ continue;
+
if (command.status() == Status.Invalidated || command.status() == Status.AcceptedInvalidate)
continue;
// current limitation of SimpleProgressLog is that only the home shard will attempt to recover, so
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
index b5958276..58cb6cbb 100644
--- a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
+++ b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -118,7 +118,7 @@ public class ImmutableCommandTest
void noConflictWitnessTest()
{
CommandStoreSupport support = new CommandStoreSupport();
- InMemoryCommandStore commands = createStore(support);
+ InMemoryCommandStore.Synchronized commands = (InMemoryCommandStore.Synchronized) createStore(support);
MockCluster.Clock clock = new MockCluster.Clock(100);
TxnId txnId = clock.idForNode(1, 1);
Keys keys = Keys.of(KEY);
@@ -130,13 +130,23 @@ public class ImmutableCommandTest
Assertions.assertEquals(Status.NotDefined, command.status());
Assertions.assertNull(command.executeAt());
}
- SafeCommandStore safeStore = commands.beginOperation(PreLoadContext.contextFor(txnId, keys.toParticipants()));
- StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, txnId.epoch());
- SafeCommand safeCommand = safeStore.get(txnId, participants);
- Commands.preaccept(safeStore, safeCommand, participants, txnId, txnId.epoch(), txn.slice(FULL_RANGES, true), ROUTE);
- Command command = safeStore.get(txnId).current();
- Assertions.assertEquals(Status.PreAccepted, command.status());
- Assertions.assertEquals(txnId, command.executeAt());
+
+ commands.execute(() -> {
+ SafeCommandStore safeStore = commands.beginOperation(PreLoadContext.contextFor(txnId, keys.toParticipants()));
+ try
+ {
+ StoreParticipants participants = StoreParticipants.update(safeStore, ROUTE, txnId.epoch(), txnId, txnId.epoch());
+ SafeCommand safeCommand = safeStore.get(txnId, participants);
+ Commands.preaccept(safeStore, safeCommand, participants, txnId, txnId.epoch(), txn.slice(FULL_RANGES, true), ROUTE);
+ Command command = safeStore.get(txnId).current();
+ Assertions.assertEquals(Status.PreAccepted, command.status());
+ Assertions.assertEquals(txnId, command.executeAt());
+ }
+ finally
+ {
+ commands.completeOperation(safeStore);
+ }
+ });
}
@Test
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 dfe37867..50b2b97c 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -719,7 +719,7 @@ public class CommandsForKeyTest
}
@Override
- protected SafeCommand ifLoadedAndInitialisedAndNotErasedInternal(TxnId txnId)
+ protected SafeCommand ifLoadedInternal(TxnId txnId)
{
return getInternal(txnId);
}
@@ -757,7 +757,7 @@ public class CommandsForKeyTest
}
@Override
- public SafeCommand ifLoadedAndInitialisedAndNotErased(TxnId txnId)
+ public SafeCommand ifLoadedAndInitialised(TxnId txnId)
{
if (txnId.compareTo(cfk.current.prunedBefore()) < 0)
return null;
diff --git a/accord-core/src/test/java/accord/messages/PreAcceptTest.java b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
index c07e8252..fb5e5667 100644
--- a/accord-core/src/test/java/accord/messages/PreAcceptTest.java
+++ b/accord-core/src/test/java/accord/messages/PreAcceptTest.java
@@ -165,7 +165,7 @@ public class PreAcceptTest
invalidate.process(node, ID2, REPLY_CONTEXT);
messageSink.assertHistorySizes(0, 1);
- assertThat(messageSink.responses.get(0).payload).isEqualTo(new BeginInvalidation.InvalidateReply(null, Ballot.ZERO, SaveStatus.NotDefined, false, null, null, null));
+ assertThat(messageSink.responses.get(0).payload).isEqualTo(new BeginInvalidation.InvalidateReply(null, Ballot.ZERO, SaveStatus.Uninitialised, false, null, null, null));
messageSink.clearHistory();
PreAccept preAccept = preAccept(txnId, txn, key.toUnseekable());
diff --git a/accord-core/src/test/java/accord/messages/ReadDataTest.java b/accord-core/src/test/java/accord/messages/ReadDataTest.java
index d118bf8e..c72022ff 100644
--- a/accord-core/src/test/java/accord/messages/ReadDataTest.java
+++ b/accord-core/src/test/java/accord/messages/ReadDataTest.java
@@ -185,8 +185,9 @@ class ReadDataTest
store = stores.get(1);
check(store.execute(PreLoadContext.contextFor(state.txnId, state.route), safeStore -> {
StoreParticipants participants = StoreParticipants.invalidate(safeStore, state.route, state.txnId);
- SafeCommand command = safeStore.get(state.txnId, participants);
- command.commitInvalidated(safeStore);
+ SafeCommand safeCommand = safeStore.get(state.txnId, participants);
+ Command prev = safeCommand.current();
+ safeCommand.commitInvalidated(safeStore);
}));
ReplyContext replyContext = state.process();
diff --git a/accord-core/src/test/java/accord/primitives/AbstractRangesTest.java b/accord-core/src/test/java/accord/primitives/AbstractRangesTest.java
index bb243399..68965df7 100644
--- a/accord-core/src/test/java/accord/primitives/AbstractRangesTest.java
+++ b/accord-core/src/test/java/accord/primitives/AbstractRangesTest.java
@@ -127,6 +127,12 @@ class AbstractRangesTest
throw new UnsupportedOperationException();
}
+ @Override
+ public Routables<Range> without(accord.primitives.Ranges ranges)
+ {
+ return new Ranges(withoutInternal(ranges));
+ }
+
@Override
public Ranges intersecting(Seekables<?, ?> intersecting)
{
diff --git a/accord-core/src/test/java/accord/primitives/RangeDepsTest.java b/accord-core/src/test/java/accord/primitives/RangeDepsTest.java
index cf03e41c..dd3ca9f9 100644
--- a/accord-core/src/test/java/accord/primitives/RangeDepsTest.java
+++ b/accord-core/src/test/java/accord/primitives/RangeDepsTest.java
@@ -237,7 +237,7 @@ public class RangeDepsTest
}
- // TODO (now): broaden patterns of random contents, sizes of collection etc.
+ // TODO (testing): broaden patterns of random contents, sizes of collection etc.
@Test
public void testRandom()
{
@@ -296,7 +296,7 @@ public class RangeDepsTest
{
for (int txnIdCount = 1; txnIdCount < 100 ; txnIdCount *= 2)
{
- // TODO (now): verify that we write only a small number of checkpoints
+ // TODO (testing): verify that we write only a small number of checkpoints
generateNemesisRanges(width, txnIdCount, 1000, nonNemesisEntryPerNemesisEntry)
.validate(random);
}
diff --git a/accord-core/src/test/java/accord/topology/TopologyRandomizer.java b/accord-core/src/test/java/accord/topology/TopologyRandomizer.java
index 135eb91b..4784d304 100644
--- a/accord-core/src/test/java/accord/topology/TopologyRandomizer.java
+++ b/accord-core/src/test/java/accord/topology/TopologyRandomizer.java
@@ -56,7 +56,7 @@ import static accord.burn.BurnTest.HASH_RANGE_END;
import static accord.burn.BurnTest.HASH_RANGE_START;
-// TODO (required, testing): add change replication factor
+// TODO (testing): add change replication factor
public class TopologyRandomizer
{
public interface Listener
@@ -453,7 +453,6 @@ public class TopologyRandomizer
state.shards = newShards;
Shard[] testShards = type.apply(state, random);
if (!everyShardHasOverlaps(current.epoch, oldShards, testShards)
- // TODO (now): I don't think it is necessary to prevent re-replicating ranges any longer
|| reassignsRanges(current, testShards, previouslyReplicated)
)
{
diff --git a/accord-core/src/test/java/accord/utils/AccordGens.java b/accord-core/src/test/java/accord/utils/AccordGens.java
index eda147a8..560caa11 100644
--- a/accord-core/src/test/java/accord/utils/AccordGens.java
+++ b/accord-core/src/test/java/accord/utils/AccordGens.java
@@ -585,7 +585,6 @@ public class AccordGens
break;
case ExclusiveSyncPoint:
case SyncPoint:
- case LocalOnly:
//TODO (coverage, now):
keyDepsGen = i -> KeyDeps.NONE;
rangeDepsGen = i -> RangeDeps.NONE;
diff --git a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
index bb09b65b..8fbadf04 100644
--- a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
+++ b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
@@ -38,7 +38,7 @@ public class LogGroupTimersTest
public void test()
{
Random random = new Random();
- for (int i = 0 ; i < 10000 ; ++i)
+ for (int i = 0 ; i < 1000 ; ++i)
testOne(random.nextLong(), 1000, 100);
}
@@ -95,6 +95,7 @@ public class LogGroupTimersTest
private void testOne(long seed, int rounds, int maxBatchSize)
{
+ System.out.printf("Seed %d\n", seed);
RandomTestRunner.test().withSeed(seed).check(rnd -> {
TimeUnit scale = rnd.randomWeightedPicker(new TimeUnit[] { MINUTES, HOURS, DAYS }).get();
long globalMaxDeadline = scale.toMicros(rnd.nextInt(1, 8));
diff --git a/accord-core/src/test/java/accord/verify/StrictSerializabilityVerifier.java b/accord-core/src/test/java/accord/verify/StrictSerializabilityVerifier.java
index c0cb6b95..a39c566d 100644
--- a/accord-core/src/test/java/accord/verify/StrictSerializabilityVerifier.java
+++ b/accord-core/src/test/java/accord/verify/StrictSerializabilityVerifier.java
@@ -53,7 +53,7 @@ import static java.util.stream.Collectors.joining;
* find a path of predecessors that would witness us.
* <p>
* TODO (low priority): find and report a path when we encounter a violation
- * TODO (required): validate we only witness values we may have written (this is a post-processing step to just validate all steps have an associated write that may have succeeded)
+ * TODO (testing): validate we only witness values we may have written (this is a post-processing step to just validate all steps have an associated write that may have succeeded)
*/
public class StrictSerializabilityVerifier implements Verifier
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org