You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/10/07 20:44:55 UTC
(cassandra-accord) 01/01: Do not contact faulty replicas, and support reporting slow replies for preaccept/read. Do not wait for stale or left nodes for durability.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit aa70db4b72f74e370dc8f3ec01697d6296cbdaa0
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Sat Oct 5 21:08:35 2024 +0100
Do not contact faulty replicas, and support reporting slow replies for preaccept/read. Do not wait for stale or left nodes for durability.
---
.../src/main/java/accord/api/ProgressLog.java | 5 +
.../src/main/java/accord/api/RequestTimeouts.java | 1 +
.../src/main/java/accord/api/TopologySorter.java | 5 +
.../coordinate/AbstractCoordinatePreAccept.java | 10 +
.../accord/coordinate/CollectCalculatedDeps.java | 7 +-
.../coordinate/CoordinateGloballyDurable.java | 5 +-
.../accord/coordinate/CoordinatePreAccept.java | 7 +
.../accord/coordinate/CoordinateShardDurable.java | 17 +-
.../accord/coordinate/CoordinateTransaction.java | 2 +
.../java/accord/coordinate/ExecuteSyncPoint.java | 5 +-
.../main/java/accord/coordinate/Invalidate.java | 5 +-
.../src/main/java/accord/coordinate/Persist.java | 14 +-
.../coordinate/PersistExclusiveSyncPoint.java | 21 +-
.../src/main/java/accord/coordinate/Propose.java | 5 +-
.../java/accord/coordinate/ReadCoordinator.java | 19 +-
.../coordinate/tracking/AbstractTracker.java | 60 ++++-
.../accord/coordinate/tracking/AllTracker.java | 2 +-
.../coordinate/tracking/FastPathTracker.java | 185 ++++++++-------
.../accord/coordinate/tracking/ReadTracker.java | 37 +--
.../java/accord/impl/AbstractRequestTimeouts.java | 253 +++++++++++++++++++++
.../java/accord/impl/DefaultRequestTimeouts.java | 148 +-----------
.../java/accord/impl/InMemoryCommandStore.java | 3 +-
.../main/java/accord/impl/RequestCallbacks.java | 251 ++++++++++++++++++++
.../java/accord/impl/SizeOfIntersectionSorter.java | 6 +
.../impl/progresslog/DefaultProgressLog.java | 51 ++---
.../java/accord/impl/progresslog/TxnState.java | 8 +-
.../src/main/java/accord/local/CommandStore.java | 10 +-
.../src/main/java/accord/local/CommandStores.java | 7 +
.../src/main/java/accord/local/Commands.java | 2 +-
accord-core/src/main/java/accord/local/Node.java | 40 ++--
.../java/accord/local/NodeCommandStoreService.java | 7 +-
.../main/java/accord/local/SafeCommandStore.java | 2 +-
.../{NodeTimeService.java => TimeService.java} | 30 ++-
.../main/java/accord/local/cfk/CommandsForKey.java | 4 +-
.../java/accord/messages/GetEphemeralReadDeps.java | 2 +-
.../main/java/accord/messages/GetMaxConflict.java | 2 +-
.../src/main/java/accord/topology/Shard.java | 7 +-
.../src/main/java/accord/topology/Topologies.java | 20 +-
.../src/main/java/accord/topology/Topology.java | 20 +-
.../main/java/accord/topology/TopologyManager.java | 16 +-
.../src/main/java/accord/utils/ArrayBuffers.java | 16 ++
.../java/accord/utils/IntrusivePriorityHeap.java | 13 +-
.../src/main/java/accord/utils/LogGroupTimers.java | 236 +++++++++----------
.../src/main/java/accord/utils/SortedArrays.java | 54 ++++-
accord-core/src/test/java/accord/Utils.java | 4 +-
.../tracking/FastPathTrackerReconciler.java | 16 +-
.../coordinate/tracking/ReadTrackerReconciler.java | 2 +-
.../coordinate/tracking/ReadTrackerTest.java | 14 +-
.../test/java/accord/impl/RemoteListenersTest.java | 4 +-
.../src/test/java/accord/impl/basic/Cluster.java | 4 +-
.../test/java/accord/impl/mock/MockCluster.java | 26 ++-
.../java/accord/local/ImmutableCommandTest.java | 6 +-
.../src/test/java/accord/local/NodeTest.java | 6 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 4 +-
.../test/java/accord/utils/LogGroupTimersTest.java | 115 ++--------
.../src/main/java/accord/maelstrom/Cluster.java | 5 +-
.../src/main/java/accord/maelstrom/Main.java | 4 +-
57 files changed, 1169 insertions(+), 661 deletions(-)
diff --git a/accord-core/src/main/java/accord/api/ProgressLog.java b/accord-core/src/main/java/accord/api/ProgressLog.java
index 5dd1c2c0..3f1263cb 100644
--- a/accord-core/src/main/java/accord/api/ProgressLog.java
+++ b/accord-core/src/main/java/accord/api/ProgressLog.java
@@ -187,6 +187,11 @@ public interface ProgressLog
void clear();
+ /**
+ * Should be thread-safe
+ */
+ default void maybeNotify() {}
+
class NoOpProgressLog implements ProgressLog
{
@Override public void update(SafeCommandStore safeStore, TxnId txnId, Command before, Command after) {}
diff --git a/accord-core/src/main/java/accord/api/RequestTimeouts.java b/accord-core/src/main/java/accord/api/RequestTimeouts.java
index 6ae723ee..83c3d2bc 100644
--- a/accord-core/src/main/java/accord/api/RequestTimeouts.java
+++ b/accord-core/src/main/java/accord/api/RequestTimeouts.java
@@ -34,4 +34,5 @@ public interface RequestTimeouts
}
RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units);
+ void maybeNotify();
}
diff --git a/accord-core/src/main/java/accord/api/TopologySorter.java b/accord-core/src/main/java/accord/api/TopologySorter.java
index 257e2e38..01871e96 100644
--- a/accord-core/src/main/java/accord/api/TopologySorter.java
+++ b/accord-core/src/main/java/accord/api/TopologySorter.java
@@ -39,10 +39,15 @@ public interface TopologySorter
default TopologySorter get(Topology topologies) { return this; }
@Override
default TopologySorter get(Topologies topologies) { return this; }
+
+ @Override
+ default boolean isFaulty(Node.Id node) { return false; }
}
/**
* Compare two nodes that occur in some topologies, so that the one that is *least* preferable sorts first
*/
int compare(Node.Id node1, Node.Id node2, ShardSelection shards);
+
+ boolean isFaulty(Node.Id node);
}
diff --git a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
index ea3d2197..1dd04ad6 100644
--- a/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java
@@ -126,6 +126,7 @@ abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> imple
abstract void contact(Collection<Id> nodes, Topologies topologies, Callback<R> callback);
abstract void onSuccessInternal(Id from, R reply);
+ void onSlowResponseInternal(Id from) {}
/**
* The tracker for the extra rounds only is provided by the AbstractCoordinatePreAccept, so we expect a boolean back
* indicating if the "success" reply was actually a good response or a failure (i.e. preempted)
@@ -155,6 +156,7 @@ abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> imple
}
@Override
+ // TODO (expected): shouldn't need synchronized
public final synchronized void onSuccess(Id from, R reply)
{
if (debug != null) debug.putIfAbsent(from, reply);
@@ -162,6 +164,13 @@ abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> imple
onSuccessInternal(from, reply);
}
+ @Override
+ public final synchronized void onSlowResponse(Id from)
+ {
+ if (!initialRoundIsDone)
+ onSlowResponseInternal(from);
+ }
+
@Override
public final boolean tryFailure(Throwable failure)
{
@@ -228,6 +237,7 @@ abstract class AbstractCoordinatePreAccept<T, R> extends SettableResult<T> imple
// TODO (desired, efficiency): check if we have already have a valid quorum for the future epoch
// (noting that nodes may have adopted new ranges, in which case they should be discounted, and quorums may have changed shape)
node.withEpoch(latestEpoch, (ignored, withEpochFailure) -> {
+ // TODO (required): this isn't synchronized, and isn't submitted back to the calling executor.
if (withEpochFailure != null)
{
tryFailure(CoordinationFailed.wrap(withEpochFailure));
diff --git a/accord-core/src/main/java/accord/coordinate/CollectCalculatedDeps.java b/accord-core/src/main/java/accord/coordinate/CollectCalculatedDeps.java
index 7e48ccb7..f8394f04 100644
--- a/accord-core/src/main/java/accord/coordinate/CollectCalculatedDeps.java
+++ b/accord-core/src/main/java/accord/coordinate/CollectCalculatedDeps.java
@@ -30,6 +30,7 @@ import accord.messages.CalculateDeps;
import accord.messages.CalculateDeps.CalculateDepsOk;
import accord.primitives.*;
import accord.topology.Topologies;
+import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.SortedListMap;
import static accord.coordinate.tracking.RequestStatus.Failed;
@@ -65,8 +66,10 @@ public class CollectCalculatedDeps implements Callback<CalculateDepsOk>
CommandStore store = CommandStore.maybeCurrent();
if (store == null)
store = node.commandStores().select(fullRoute);
- node.send(collect.tracker.nodes(), to -> new CalculateDeps(to, topologies, fullRoute, txnId, executeAt),
- store, collect);
+
+ SortedArrayList<Id> contact = collect.tracker.filterAndRecordFaulty();
+ if (contact == null) callback.accept(null, new Exhausted(txnId, collect.homeKey, null));
+ else node.send(contact, to -> new CalculateDeps(to, topologies, fullRoute, txnId, executeAt), store, collect);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateGloballyDurable.java b/accord-core/src/main/java/accord/coordinate/CoordinateGloballyDurable.java
index 7751c35b..5b0dd834 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateGloballyDurable.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateGloballyDurable.java
@@ -27,6 +27,7 @@ import accord.messages.QueryDurableBefore;
import accord.messages.QueryDurableBefore.DurableBeforeReply;
import accord.messages.SetGloballyDurable;
import accord.topology.Topologies;
+import accord.utils.SortedArrays;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults.SettableResult;
@@ -55,7 +56,9 @@ public class CoordinateGloballyDurable extends SettableResult<Void> implements C
private void start()
{
- node.send(tracker.nodes(), to -> new QueryDurableBefore(tracker.topologies().currentEpoch()), this);
+ SortedArrays.SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ if (contact == null) tryFailure(new Exhausted(null, null, null));
+ else node.send(contact, to -> new QueryDurableBefore(tracker.topologies().currentEpoch()), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
index 06a023e0..8090697e 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java
@@ -119,6 +119,13 @@ abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, Pre
}
}
+ @Override
+ void onSlowResponseInternal(Id from)
+ {
+ if (tracker.recordDelayed(from) == Success)
+ onPreAcceptedOrNewEpoch();
+ }
+
@Override
boolean onExtraSuccessInternal(Id from, PreAcceptReply reply)
{
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateShardDurable.java b/accord-core/src/main/java/accord/coordinate/CoordinateShardDurable.java
index ec6e8a85..f5193a56 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateShardDurable.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateShardDurable.java
@@ -20,6 +20,7 @@ package accord.coordinate;
import accord.coordinate.ExecuteSyncPoint.ExecuteExclusiveSyncPoint;
import accord.coordinate.tracking.AllTracker;
+import accord.coordinate.tracking.RequestStatus;
import accord.local.Node;
import accord.messages.Callback;
import accord.messages.ReadData.ReadReply;
@@ -27,6 +28,8 @@ import accord.messages.SetShardDurable;
import accord.messages.WaitUntilApplied;
import accord.primitives.Range;
import accord.primitives.SyncPoint;
+import accord.utils.Invariants;
+import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncResult;
public class CoordinateShardDurable extends ExecuteExclusiveSyncPoint implements Callback<ReadReply>
@@ -52,6 +55,18 @@ public class CoordinateShardDurable extends ExecuteExclusiveSyncPoint implements
protected void start()
{
- node.send(tracker.nodes(), to -> new WaitUntilApplied(to, tracker.topologies(), syncPoint.syncId, syncPoint.route, syncPoint.syncId.epoch()), this);
+ SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ SortedArrayList<Node.Id> allStaleNodes = tracker.topologies().staleNodes();
+ SortedArrayList<Node.Id> allCurrentNodes = tracker.topologies().current().nodes();
+ SortedArrayList<Node.Id> staleNodes = contact.without(id -> !allStaleNodes.contains(id))
+ .without(id -> !allCurrentNodes.contains(id));
+ contact = contact.without(staleNodes);
+ for (Node.Id id :staleNodes)
+ {
+ RequestStatus newStatus = tracker.recordSuccess(id);
+ Invariants.checkState(newStatus != RequestStatus.Success);
+ }
+ if (contact == null) tryFailure(new Exhausted(syncPoint.syncId, syncPoint.route.homeKey(), null));
+ else node.send(contact, to -> new WaitUntilApplied(to, tracker.topologies(), syncPoint.syncId, syncPoint.route, syncPoint.syncId.epoch()), this);
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
index 592b3a2e..5cc380c1 100644
--- a/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
+++ b/accord-core/src/main/java/accord/coordinate/CoordinateTransaction.java
@@ -68,6 +68,8 @@ public class CoordinateTransaction extends CoordinatePreAccept<Result>
@Override
void onPreAccepted(Topologies topologies, Timestamp executeAt, List<PreAcceptOk> oks)
{
+ // TODO (expected): this condition is too strong - we only need fast path _votes_ in the proposal epoch
+ // but must have replied from a fast path quorum in earlier epochs
if (tracker.hasFastPathAccepted())
{
Deps deps = Deps.merge(oks, oks.size(), List::get, ok -> ok.witnessedAt.equals(txnId) ? ok.deps : null);
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
index b1979f2c..0fe5029a 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
@@ -40,6 +40,7 @@ import accord.primitives.Unseekable;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncResults.SettableResult;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
@@ -94,7 +95,9 @@ public abstract class ExecuteSyncPoint<U extends Unseekable> extends SettableRes
@Override
protected void start()
{
- node.send(tracker.nodes(), to -> new WaitUntilApplied(to, tracker.topologies(), syncPoint.syncId, syncPoint.route, syncPoint.syncId.epoch()), this);
+ SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ if (contact == null) tryFailure(new Exhausted(syncPoint.syncId, syncPoint.route.homeKey(), null));
+ else node.send(contact, to -> new WaitUntilApplied(to, tracker.topologies(), syncPoint.syncId, syncPoint.route, syncPoint.syncId.epoch()), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/Invalidate.java b/accord-core/src/main/java/accord/coordinate/Invalidate.java
index 36db93cb..b59f0065 100644
--- a/accord-core/src/main/java/accord/coordinate/Invalidate.java
+++ b/accord-core/src/main/java/accord/coordinate/Invalidate.java
@@ -35,6 +35,7 @@ import accord.messages.BeginInvalidation.InvalidateReply;
import accord.messages.Callback;
import accord.topology.Topology;
import accord.utils.Invariants;
+import accord.utils.SortedArrays;
import javax.annotation.Nullable;
@@ -92,7 +93,9 @@ public class Invalidate implements Callback<InvalidateReply>
private void start()
{
- node.send(tracker.nodes(), to -> new BeginInvalidation(to, tracker.topologies(), txnId, invalidateWith, ballot), this);
+ SortedArrays.SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ if (contact == null) callback.accept(null, new Exhausted(null, null, null));
+ else node.send(contact, to -> new BeginInvalidation(to, tracker.topologies(), txnId, invalidateWith, ballot), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/Persist.java b/accord-core/src/main/java/accord/coordinate/Persist.java
index 1027dde2..042cb228 100644
--- a/accord-core/src/main/java/accord/coordinate/Persist.java
+++ b/accord-core/src/main/java/accord/coordinate/Persist.java
@@ -37,6 +37,7 @@ import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.topology.Topologies;
import accord.topology.Topology;
+import accord.utils.SortedArrays;
import static accord.coordinate.tracking.RequestStatus.Success;
import static accord.primitives.Status.Durability.Majority;
@@ -103,18 +104,27 @@ public abstract class Persist implements Callback<ApplyReply>
public void onFailure(Id from, Throwable failure)
{
// TODO (desired, consider): send knowledge of partial persistence?
+ // TODO (expected): we should presumably report total request failure somewhere?
}
@Override
public void onCallbackFailure(Id from, Throwable failure)
{
- // TODO (required): handle exception
+ // TODO (expected): handle exception
}
public void start(Apply.Factory factory, Apply.Kind kind, Topologies all, Writes writes, Result result)
{
// applyMinimal is used for transaction execution by the original coordinator so it's important to use
// Node's Apply factory in case the factory has to do synchronous Apply.
- node.send(all.nodes(), to -> factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result), this);
+ SortedArrays.SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ if (contact == null)
+ {
+ // TODO (expected): we should presumably report this somewhere?
+ }
+ else
+ {
+ node.send(contact, to -> factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result), this);
+ }
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/PersistExclusiveSyncPoint.java b/accord-core/src/main/java/accord/coordinate/PersistExclusiveSyncPoint.java
index ab22888d..cc628428 100644
--- a/accord-core/src/main/java/accord/coordinate/PersistExclusiveSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/PersistExclusiveSyncPoint.java
@@ -28,6 +28,7 @@ import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.topology.Topologies;
+import accord.utils.SortedArrays;
public class PersistExclusiveSyncPoint extends Persist
{
@@ -40,13 +41,21 @@ public class PersistExclusiveSyncPoint extends Persist
public void start(Apply.Factory factory, Apply.Kind kind, Topologies all, Writes writes, Result result)
{
factory = Apply.wrapForExclusiveSyncPoint(factory);
- for (Node.Id to : topologies.nodes())
+ SortedArrays.SortedArrayList<Node.Id> contact = tracker.filterAndRecordFaulty();
+ if (contact == null)
{
- Apply apply = factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result);
- if (apply == null)
- tracker.recordSuccess(to);
- else
- node.send(to, apply, this);
+ // TODO (expected): we should report this somewhere?
+ }
+ else
+ {
+ for (Node.Id to : contact)
+ {
+ Apply apply = factory.create(kind, to, all, txnId, route, txn, executeAt, stableDeps, writes, result);
+ if (apply == null)
+ tracker.recordSuccess(to);
+ else
+ node.send(to, apply, this);
+ }
}
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/Propose.java b/accord-core/src/main/java/accord/coordinate/Propose.java
index b4d4cf5c..8c477e95 100644
--- a/accord-core/src/main/java/accord/coordinate/Propose.java
+++ b/accord-core/src/main/java/accord/coordinate/Propose.java
@@ -41,6 +41,7 @@ import accord.primitives.TxnId;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.utils.Invariants;
+import accord.utils.SortedArrays;
import accord.utils.SortedListMap;
import static accord.coordinate.tracking.AbstractTracker.ShardOutcomes.Fail;
@@ -81,7 +82,9 @@ abstract class Propose<R> implements Callback<AcceptReply>
void start()
{
- node.send(acceptTracker.nodes(), to -> new Accept(to, acceptTracker.topologies(), ballot, txnId, route, executeAt, deps), this);
+ SortedArrays.SortedArrayList<Node.Id> contact = acceptTracker.filterAndRecordFaulty();
+ if (contact == null) callback.accept(null, new Timeout(null, null));
+ else node.send(contact, to -> new Accept(to, acceptTracker.topologies(), ballot, txnId, route, executeAt, deps), this);
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java b/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
index 68910f8d..99ee9b8f 100644
--- a/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
+++ b/accord-core/src/main/java/accord/coordinate/ReadCoordinator.java
@@ -138,7 +138,7 @@ public abstract class ReadCoordinator<Reply extends accord.messages.Reply> exten
break;
case Reject:
- handle(recordReadFailure(from));
+ handle(recordFailure(from));
break;
case ApproveIfQuorum:
@@ -173,7 +173,7 @@ public abstract class ReadCoordinator<Reply extends accord.messages.Reply> exten
if (this.failure == null) this.failure = failure;
else this.failure.addSuppressed(failure);
- handle(recordReadFailure(from));
+ handle(recordFailure(from));
}
@Override
@@ -263,10 +263,17 @@ public abstract class ReadCoordinator<Reply extends accord.messages.Reply> exten
public void start()
{
- List<Id> contact = new ArrayList<>(maxShardsPerEpoch());
- if (trySendMore(List::add, contact) != RequestStatus.NoChange)
- throw new IllegalStateException();
- start(contact);
+ if (!initialise())
+ {
+ finishOnExhaustion();
+ }
+ else
+ {
+ List<Id> contact = new ArrayList<>(maxShardsPerEpoch());
+ if (trySendMore(List::add, contact) != RequestStatus.NoChange)
+ throw new IllegalStateException();
+ start(contact);
+ }
}
@Override
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
index 183b4879..02d5c691 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
@@ -18,27 +18,31 @@
package accord.coordinate.tracking;
+import accord.api.TopologySorter;
import accord.local.Node.Id;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
import accord.utils.Invariants;
+import accord.utils.SortedArrays.SortedArrayList;
+
import com.google.common.annotations.VisibleForTesting;
-import java.util.*;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.IntFunction;
import java.util.function.Predicate;
import static accord.coordinate.tracking.AbstractTracker.ShardOutcomes.NoChange;
+import static accord.coordinate.tracking.RequestStatus.Failed;
+import static accord.utils.ArrayBuffers.cachedAny;
public abstract class AbstractTracker<ST extends ShardTracker>
{
public enum ShardOutcomes implements ShardOutcome<AbstractTracker<?>>
{
- Fail(RequestStatus.Failed),
+ Fail(Failed),
Success(RequestStatus.Success),
SendMore(null),
NoChange(RequestStatus.NoChange);
@@ -115,6 +119,8 @@ public abstract class AbstractTracker<ST extends ShardTracker>
this.waitingOnShards = shardCount;
}
+ protected abstract RequestStatus recordFailure(Id from);
+
protected int topologyOffset(int topologyIdx)
{
return topologyIdx * maxShardsPerEpoch();
@@ -175,14 +181,24 @@ public abstract class AbstractTracker<ST extends ShardTracker>
return true;
}
- public Collection<Id> nodes()
+ public SortedArrayList<Id> nodes()
{
return topologies.nodes();
}
- public Collection<Id> nonStaleNodes()
+ public SortedArrayList<Id> filterAndRecordFaulty()
+ {
+ return filterAndRecordFaulty(topologies.nodes());
+ }
+
+ public SortedArrayList<Id> filterAndRecordFaulty(SortedArrayList<Id> ids)
{
- return topologies.nonStaleNodes();
+ return filterAndRecordFaulty(ids, topologies, this);
+ }
+
+ public SortedArrayList<Id> filterAndRecordStale(SortedArrayList<Id> ids)
+ {
+ return filterAndRecordFaulty(ids, topologies, this);
}
public ST get(int shardIndex)
@@ -203,4 +219,38 @@ public abstract class AbstractTracker<ST extends ShardTracker>
{
return maxShardsPerEpoch;
}
+
+ public static SortedArrayList<Id> filterAndRecordFaulty(SortedArrayList<Id> nodes, TopologySorter sorter, AbstractTracker<?> reportTo)
+ {
+ Object[] buffer = null;
+ int bufferCount = 0;
+ int previ = 0;
+ for (int i = 0 ; i < nodes.size() ; ++i)
+ {
+ Id node = nodes.get(i);
+ if (sorter.isFaulty(node))
+ {
+ if (Failed == reportTo.recordFailure(node))
+ return null;
+
+ if (buffer == null)
+ buffer = cachedAny().get(nodes.size());
+
+ int count = i - previ;
+ System.arraycopy(nodes.backingArrayUnsafe(), previ, buffer, bufferCount, count);
+ previ = i + 1;
+ bufferCount += count;
+ }
+ }
+ if (buffer == null)
+ return nodes;
+
+ int count = nodes.size() - previ;
+ System.arraycopy(nodes.backingArrayUnsafe(), previ, buffer, bufferCount, count);
+ bufferCount += count;
+
+ Id[] newIds = new Id[bufferCount];
+ System.arraycopy(buffer, 0, newIds, 0, bufferCount);
+ return SortedArrayList.ofSorted(newIds);
+ }
}
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/AllTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/AllTracker.java
index 86d7aa5e..07628bc1 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/AllTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/AllTracker.java
@@ -70,7 +70,7 @@ public class AllTracker extends AbstractSimpleTracker<AllTracker.AllShardTracker
public AllTracker(Topologies topologies)
{
- super(topologies, AllShardTracker[]::new, shard -> new AllShardTracker(shard));
+ super(topologies, AllShardTracker[]::new, AllShardTracker::new);
}
public RequestStatus recordSuccess(Node.Id node)
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 7d30955c..187c052f 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/FastPathTracker.java
@@ -31,67 +31,111 @@ import java.util.function.BiFunction;
import static accord.coordinate.tracking.AbstractTracker.ShardOutcomes.*;
// TODO (desired, efficiency): if any shard *cannot* take the fast path, and all shards have accepted, terminate
-public class FastPathTracker extends AbstractTracker<FastPathTracker.FastPathQuorumShardTracker>
+public class FastPathTracker extends AbstractTracker<FastPathTracker.FastPathShardTracker>
{
private static final ShardOutcome<FastPathTracker> NewFastPathSuccess = (tracker, shardIndex) -> {
--tracker.waitingOnFastPathSuccess;
return --tracker.waitingOnShards == 0 ? Success : NoChange;
};
- public static class FastPathQuorumShardTracker extends QuorumShardTracker
+ public static abstract class FastPathShardTracker extends QuorumShardTracker
{
- public FastPathQuorumShardTracker(Shard shard)
+ protected int fastPathAccepts;
+ protected int fastPathFailures;
+ protected int fastPathDelayed;
+
+ public FastPathShardTracker(Shard shard)
{
super(shard);
}
- public ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node)
+ // 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);
+
+ public final ShardOutcome<? super FastPathTracker> onFailure(@Nonnull Node.Id from)
{
- return super.onSuccess(node);
+ if (++failures > shard.maxFailures)
+ return Fail;
+
+ if (shard.fastPathElectorate.contains(from))
+ {
+ ++fastPathFailures;
+
+ if (isNewFastPathReject() && hasReachedQuorum())
+ return Success;
+ }
+
+ return NoChange;
}
- public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
+ public final ShardOutcome<? super FastPathTracker> onDelayed(@Nonnull Node.Id from)
{
- return super.onSuccess(node);
+ if (shard.fastPathElectorate.contains(from))
+ {
+ ++fastPathDelayed;
+
+ if (isNewFastPathDelayed() && hasReachedQuorum())
+ return Success;
+ }
+
+ return NoChange;
}
- public ShardOutcome<? super FastPathTracker> onFailure(@Nonnull Node.Id from)
+ final ShardOutcome<? super FastPathTracker> quorumIfNewAndHasRejectedFastPath()
{
- return super.onFailure(from);
+ return isNewSlowPathSuccess() && hasRejectedFastPath() ? Success : NoChange;
}
- public boolean hasRejectedFastPath()
+ final boolean isNewSlowPathSuccess()
{
- return false;
+ return successes == shard.slowPathQuorumSize;
}
- public boolean hasMetFastPathCriteria()
+ final boolean isNewFastPathReject()
{
- return true;
+ return fastPathFailures == shard.fastPathRejectSize && fastPathDelayed < shard.fastPathRejectSize;
+ }
+
+ final boolean isNewFastPathDelayed()
+ {
+ return fastPathDelayed == shard.fastPathRejectSize && fastPathFailures < shard.fastPathRejectSize;
+ }
+
+ final boolean isNewFastPathSuccess()
+ {
+ return fastPathAccepts == shard.fastPathQuorumSize && fastPathDelayed < shard.fastPathRejectSize;
+ }
+
+ @VisibleForTesting
+ public final boolean hasMetFastPathCriteria()
+ {
+ return fastPathAccepts >= shard.fastPathQuorumSize;
+ }
+
+ @VisibleForTesting
+ public final boolean hasRejectedFastPath()
+ {
+ return Math.max(fastPathFailures, fastPathDelayed) >= shard.fastPathRejectSize;
}
}
- public static class FastPathShardTracker extends FastPathQuorumShardTracker
+ public static final class PriorFastPathShardTracker extends FastPathShardTracker
{
- protected int fastPathAccepts;
- protected int fastPathFailures;
-
- public FastPathShardTracker(Shard shard)
+ public PriorFastPathShardTracker(Shard shard)
{
super(shard);
}
- // return NewQuorumSuccess ONLY once fast path is rejected
- @Override
public ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node)
{
++successes;
if (!shard.fastPathElectorate.contains(node))
- return quorumIfRejectsFastPath();
+ return quorumIfNewAndHasRejectedFastPath();
- ++fastPathFailures;
- if (isNewFastPathReject() && hasReachedQuorum())
- return Success;
+ ++fastPathAccepts;
+ if (isNewFastPathSuccess())
+ return NewFastPathSuccess;
if (isNewSlowPathSuccess() && hasRejectedFastPath())
return Success;
@@ -102,88 +146,73 @@ public class FastPathTracker extends AbstractTracker<FastPathTracker.FastPathQuo
@Override
public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
{
- ++successes;
- if (shard.fastPathElectorate.contains(node))
- {
- ++fastPathAccepts;
- if (isNewFastPathSuccess())
- return NewFastPathSuccess;
- }
- return quorumIfRejectsFastPath();
- }
-
- @Override
- public ShardOutcome<? super FastPathTracker> onFailure(@Nonnull Node.Id from)
- {
- if (++failures > shard.maxFailures)
- return Fail;
-
- if (shard.fastPathElectorate.contains(from)) {
- ++fastPathFailures;
-
- if (isNewFastPathReject() && successes >= shard.slowPathQuorumSize)
- return Success;
- }
-
- return NoChange;
+ return onQuorumSuccess(node);
}
+ }
- private ShardOutcome<? super FastPathTracker> quorumIfRejectsFastPath()
+ public static final class CurrentFastPathShardTracker extends FastPathShardTracker
+ {
+ public CurrentFastPathShardTracker(Shard shard)
{
- return isNewSlowPathSuccess() && hasRejectedFastPath() ? Success : NoChange;
+ super(shard);
}
- private boolean isNewSlowPathSuccess()
+ public ShardOutcome<? super FastPathTracker> onQuorumSuccess(Node.Id node)
{
- return successes == shard.slowPathQuorumSize;
- }
+ ++successes;
+ if (!shard.fastPathElectorate.contains(node))
+ return quorumIfNewAndHasRejectedFastPath();
- private boolean isNewFastPathReject()
- {
- return fastPathFailures == 1 + shard.fastPathElectorate.size() - shard.fastPathQuorumSize;
- }
+ ++fastPathFailures;
+ if (isNewFastPathReject() && hasReachedQuorum())
+ return Success;
- private boolean isNewFastPathSuccess()
- {
- return fastPathAccepts == shard.fastPathQuorumSize;
- }
+ if (isNewSlowPathSuccess() && hasRejectedFastPath())
+ return Success;
- @Override
- @VisibleForTesting
- public boolean hasMetFastPathCriteria()
- {
- return fastPathAccepts >= shard.fastPathQuorumSize;
+ return NoChange;
}
@Override
- @VisibleForTesting
- public boolean hasRejectedFastPath()
+ public ShardOutcome<? super FastPathTracker> onMaybeFastPathSuccess(Node.Id node)
{
- return shard.rejectsFastPath(fastPathFailures);
+ ++successes;
+ if (shard.fastPathElectorate.contains(node))
+ {
+ ++fastPathAccepts;
+ if (isNewFastPathSuccess())
+ return NewFastPathSuccess;
+ }
+ return quorumIfNewAndHasRejectedFastPath();
}
}
int waitingOnFastPathSuccess; // if we reach zero, we have succeeded on the fast path outcome for every shard
public FastPathTracker(Topologies topologies)
{
- super(topologies, FastPathQuorumShardTracker[]::new, (i, shard) -> i == 0 ? new FastPathShardTracker(shard) : new FastPathQuorumShardTracker(shard));
- this.waitingOnFastPathSuccess = topologies.current().size();
+ super(topologies, FastPathShardTracker[]::new, (i, shard) -> i == 0 ? new CurrentFastPathShardTracker(shard) : new PriorFastPathShardTracker(shard));
+ this.waitingOnFastPathSuccess = super.waitingOnShards;
}
public RequestStatus recordSuccess(Node.Id from, boolean withFastPathTimestamp)
{
if (withFastPathTimestamp)
- return recordResponse(from, FastPathQuorumShardTracker::onMaybeFastPathSuccess);
+ return recordResponse(from, FastPathShardTracker::onMaybeFastPathSuccess);
- return recordResponse(from, FastPathQuorumShardTracker::onQuorumSuccess);
+ return recordResponse(from, FastPathShardTracker::onQuorumSuccess);
}
public RequestStatus recordFailure(Node.Id from)
{
- return recordResponse(from, FastPathQuorumShardTracker::onFailure);
+ return recordResponse(from, FastPathShardTracker::onFailure);
+ }
+
+ public RequestStatus recordDelayed(Node.Id from)
+ {
+ return recordResponse(from, FastPathShardTracker::onDelayed);
}
- protected RequestStatus recordResponse(Node.Id node, BiFunction<? super FastPathQuorumShardTracker, Node.Id, ? extends ShardOutcome<? super FastPathTracker>> function)
+ protected RequestStatus recordResponse(Node.Id node, BiFunction<? super FastPathShardTracker, Node.Id, ? extends ShardOutcome<? super FastPathTracker>> function)
{
return recordResponse(this, node, function, node);
}
@@ -195,16 +224,16 @@ public class FastPathTracker extends AbstractTracker<FastPathTracker.FastPathQuo
public boolean hasFailed()
{
- return any(FastPathQuorumShardTracker::hasFailed);
+ return any(FastPathShardTracker::hasFailed);
}
public boolean hasInFlight()
{
- return any(FastPathQuorumShardTracker::hasInFlight);
+ return any(FastPathShardTracker::hasInFlight);
}
public boolean hasReachedQuorum()
{
- return all(FastPathQuorumShardTracker::hasReachedQuorum);
+ return all(FastPathShardTracker::hasReachedQuorum);
}
}
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/ReadTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/ReadTracker.java
index 3a442ca5..f80c23d2 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/ReadTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/ReadTracker.java
@@ -31,11 +31,12 @@ import accord.utils.Invariants;
import accord.local.Node.Id;
import accord.topology.Topologies;
+import accord.utils.SortedArrays.SortedArrayList;
+import org.agrona.collections.IntHashSet;
import static accord.coordinate.tracking.AbstractTracker.ShardOutcomes.*;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.Invariants.illegalState;
-import static com.google.common.collect.Sets.newHashSetWithExpectedSize;
public class ReadTracker extends AbstractTracker<ReadTracker.ReadShardTracker>
{
@@ -201,23 +202,33 @@ public class ReadTracker extends AbstractTracker<ReadTracker.ReadShardTracker>
}
}
- final Set<Id> inflight; // TODO (easy, efficiency): use Agrona's IntHashSet as soon as Node.Id switches from long to int
- final List<Id> candidates; // TODO (easy, efficiency): use Agrona's IntArrayList as soon as Node.Id switches from long to int
- private Set<Id> slow; // TODO (easy, efficiency): use Agrona's IntHashSet as soon as Node.Id switches from long to int
+ final IntHashSet inflight;
+ final List<Id> candidates; // TODO (easy, efficiency): use Agrona's IntArrayList
+ private IntHashSet slow;
protected int waitingOnData;
public ReadTracker(Topologies topologies)
{
super(topologies, ReadShardTracker[]::new, ReadShardTracker::new);
- this.candidates = new ArrayList<>(topologies.nodes()); // TODO (low priority, efficiency): copyOfNodesAsList to avoid unnecessary copies
- this.inflight = newHashSetWithExpectedSize(maxShardsPerEpoch());
+ this.candidates = new ArrayList<>(topologies.nodes().size());
+ this.inflight = new IntHashSet(topologies.nodes().size());
this.waitingOnData = waitingOnShards;
}
+ public boolean initialise()
+ {
+ SortedArrayList<Id> candidates = filterAndRecordFaulty();
+ if (candidates == null)
+ return false;
+
+ this.candidates.addAll(candidates);
+ return true;
+ }
+
@VisibleForTesting
protected void recordInFlightRead(Id node)
{
- if (!inflight.add(node))
+ if (!inflight.add(node.id))
throw illegalState(node + " already in flight");
recordResponse(this, node, ReadShardTracker::recordInFlightRead, false);
@@ -225,10 +236,10 @@ public class ReadTracker extends AbstractTracker<ReadTracker.ReadShardTracker>
private boolean receiveResponseIsSlow(Id node)
{
- if (!inflight.remove(node))
+ if (!inflight.remove(node.id))
throw illegalState("Nothing in flight for " + node);
- return slow != null && slow.remove(node);
+ return slow != null && slow.remove(node.id);
}
/**
@@ -236,13 +247,13 @@ public class ReadTracker extends AbstractTracker<ReadTracker.ReadShardTracker>
*/
protected RequestStatus recordSlowResponse(Id from)
{
- if (!inflight.contains(from))
+ if (!inflight.contains(from.id))
throw new IllegalStateException();
if (slow == null)
- slow = newHashSetWithExpectedSize(maxShardsPerEpoch());
+ slow = new IntHashSet(topologies.nodes().size());
- if (!slow.add(from)) // we can mark slow responses due to ReadCoordinator.TryAlternative OR onSlowResponse
+ if (!slow.add(from.id)) // we can mark slow responses due to ReadCoordinator.TryAlternative OR onSlowResponse
return RequestStatus.NoChange;
return recordResponse(this, from, ReadShardTracker::recordSlowResponse, true);
@@ -276,7 +287,7 @@ public class ReadTracker extends AbstractTracker<ReadTracker.ReadShardTracker>
/**
* Record a failure response
*/
- protected RequestStatus recordReadFailure(Id from)
+ protected RequestStatus recordFailure(Id from)
{
return recordResponse(from, ReadShardTracker::recordReadFailure);
}
diff --git a/accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java b/accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java
new file mode 100644
index 00000000..e909fe2c
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/AbstractRequestTimeouts.java
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.impl;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Function;
+import java.util.function.IntFunction;
+
+import accord.api.RequestTimeouts;
+import accord.local.TimeService;
+import accord.utils.ArrayBuffers.BufferList;
+import accord.utils.LogGroupTimers;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+public class AbstractRequestTimeouts<S extends AbstractRequestTimeouts.Stripe> implements RequestTimeouts
+{
+ protected interface Expiring
+ {
+ Expiring prepareToExpire(Stripe stripe, long now);
+ void onExpire(long now);
+ }
+
+ protected static class Stripe implements Runnable, Function<Stripe.AbstractRegistered, Runnable>
+ {
+ protected abstract class AbstractRegistered extends LogGroupTimers.Timer implements RegisteredTimeout, Expiring
+ {
+ @Override
+ public void cancel()
+ {
+ lock.lock();
+ try
+ {
+ if (isInHeap())
+ timeouts.remove(this);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+ }
+
+ protected class Registered extends AbstractRegistered
+ {
+ final Timeout timeout;
+
+ Registered(Timeout timeout)
+ {
+ this.timeout = timeout;
+ }
+
+
+ @Override
+ public void onExpire(long now)
+ {
+ timeout.timeout();
+ }
+
+ @Override
+ public Expiring prepareToExpire(Stripe stripe, long now)
+ {
+ return this;
+ }
+ }
+
+ private final ReentrantLock lock = new ReentrantLock();
+ final TimeService time;
+ final LogGroupTimers<AbstractRegistered> timeouts = new LogGroupTimers<>(TimeUnit.MICROSECONDS);
+
+ public Stripe(TimeService time)
+ {
+ this.time = time;
+ }
+
+ @Override
+ public void run()
+ {
+ long now = 0;
+ try (BufferList<Expiring> collect = new BufferList<>())
+ {
+ int i = 0;
+ try
+ {
+ lock.lock();
+ try
+ {
+ now = time.elapsed(MICROSECONDS);
+ timeouts.advance(now, collect, BufferList::add);
+
+ // prepare expiration while we hold the lock - this is to permit expiring objects to
+ // reschedule themselves while returning some immediate expiry work to do
+ for (int j = 0 ; j < collect.size() ; ++j)
+ {
+ Expiring in = collect.get(j);
+ Expiring out = in.prepareToExpire(this, now);
+ if (in != out)
+ collect.set(j, out);
+ }
+ }
+ finally
+ {
+ lock.unlock();
+ }
+
+ while (i < collect.size())
+ collect.get(i++).onExpire(now);
+ }
+ catch (Throwable t)
+ {
+ while (i < collect.size())
+ {
+ try
+ {
+ collect.get(i++).onExpire(now);
+ }
+ catch (Throwable t2)
+ {
+ t.addSuppressed(t2);
+ }
+ }
+ throw t;
+ }
+ }
+ }
+
+ protected void register(AbstractRegistered register, long deadline)
+ {
+ timeouts.add(deadline, register);
+ }
+
+ @Override
+ public Runnable apply(AbstractRegistered registered)
+ {
+ return this;
+ }
+
+ RegisteredTimeout tryRegister(Timeout timeout, long now, long deadline)
+ {
+ if (!tryLock())
+ return null;
+
+ try
+ {
+ Registered registered = new Registered(timeout);
+ register(registered, deadline);
+ return registered;
+ }
+ finally
+ {
+ unlock(now);
+ }
+ }
+
+ protected void lock()
+ {
+ lock.lock();
+ }
+
+ protected boolean tryLock()
+ {
+ return lock.tryLock();
+ }
+
+ protected void unlock()
+ {
+ try
+ {
+ if (timeouts.shouldWake(time.elapsed(MICROSECONDS)))
+ run();
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ protected void unlock(long now)
+ {
+ try
+ {
+ if (timeouts.shouldWake(now))
+ run();
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+ }
+
+ final TimeService time;
+ final S[] stripes;
+
+ protected AbstractRequestTimeouts(TimeService time, IntFunction<S[]> arrayAllocator, Function<TimeService, S> stripeFactory)
+ {
+ this(time, 8, arrayAllocator, stripeFactory);
+ }
+
+ public AbstractRequestTimeouts(TimeService time, int stripeCount, IntFunction<S[]> allocator, Function<TimeService, S> stripeFactory)
+ {
+ if (stripeCount > 1024)
+ throw new IllegalArgumentException("Far too many stripes requested");
+ this.stripes = allocator.apply(stripeCount);
+ for (int i = 0 ; i < stripes.length ; ++i)
+ stripes[i] = stripeFactory.apply(time);
+ this.time = time;
+ }
+
+ @Override
+ public RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units)
+ {
+ long now = time.elapsed(MICROSECONDS);
+ long deadline = now + Math.max(1, MICROSECONDS.convert(delay, units));
+ int i = timeout.stripe() & (stripes.length - 1);
+ while (true)
+ {
+ RegisteredTimeout result = stripes[i].tryRegister(timeout, now, deadline);
+ if (result != null)
+ return result;
+ i = (i + 1) & (stripes.length - 1);
+ }
+ }
+
+ // a fallback method to be invoked at some frequency, to notify pending timeouts in the event that a stripe has been idle too long
+ @Override
+ public void maybeNotify()
+ {
+ long nowMicros = time.elapsed(MICROSECONDS);
+ for (Stripe stripe : stripes)
+ {
+ if (stripe.timeouts.shouldWake(nowMicros) && stripe.tryLock())
+ stripe.unlock(nowMicros);
+ }
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java b/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
index 7365a314..10ff3bb4 100644
--- a/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
+++ b/accord-core/src/main/java/accord/impl/DefaultRequestTimeouts.java
@@ -18,156 +18,12 @@
package accord.impl;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Function;
-
-import accord.api.RequestTimeouts;
import accord.local.Node;
-import accord.utils.ArrayBuffers.BufferList;
-import accord.utils.LogGroupTimers;
-
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.agrona.BitUtil.findNextPositivePowerOfTwo;
-public class DefaultRequestTimeouts implements RequestTimeouts
+public class DefaultRequestTimeouts extends AbstractRequestTimeouts<AbstractRequestTimeouts.Stripe>
{
- static class Stripe implements Runnable, Function<Stripe.Registered, Runnable>
- {
- class Registered extends LogGroupTimers.Timer implements RegisteredTimeout
- {
- final Timeout timeout;
-
- Registered(Timeout timeout)
- {
- this.timeout = timeout;
- }
-
- @Override
- public void cancel()
- {
- lock.lock();
- try
- {
- if (isInHeap())
- timeouts.remove(this);
- }
- finally
- {
- lock.unlock();
- }
- }
- }
-
- final ReentrantLock lock = new ReentrantLock();
- final Node node;
- final LogGroupTimers<Registered> timeouts = new LogGroupTimers<>(TimeUnit.MILLISECONDS);
- final LogGroupTimers<Registered>.Scheduling timeoutScheduling;
-
- public Stripe(Node node)
- {
- this.node = node;
- this.timeoutScheduling = timeouts.new Scheduling(node.scheduler(), this,
- 100, 1L, SECONDS.toMillis(1));
- }
-
- public RegisteredTimeout tryRegister(Timeout timeout, long now, long deadline)
- {
- if (!lock.tryLock())
- return null;
-
- try
- {
- Registered registered = new Registered(timeout);
- timeouts.add(deadline, registered);
- timeoutScheduling.maybeReschedule(now, deadline);
- return registered;
- }
- finally
- {
- lock.unlock();
- }
- }
-
- @Override
- public void run()
- {
- try (BufferList<Registered> collect = new BufferList<>())
- {
- int i = 0;
- try
- {
- lock.lock();
- try
- {
- long now = node.elapsed(MILLISECONDS);
- // TODO (expected): should we handle reentrancy? Or at least throw an exception?
- timeouts.advance(now, collect, BufferList::add);
- }
- finally
- {
- lock.unlock();
- }
-
- while (i < collect.size())
- collect.get(i++).timeout.timeout();
- }
- catch (Throwable t)
- {
- while (i < collect.size())
- {
- try
- {
- collect.get(i++).timeout.timeout();
- }
- catch (Throwable t2)
- {
- t.addSuppressed(t2);
- }
- }
- throw t;
- }
- }
- }
-
- @Override
- public Runnable apply(Registered registered)
- {
- return this;
- }
- }
-
- final Node node;
- final Stripe[] stripes;
-
public DefaultRequestTimeouts(Node node)
{
- this(node, 8);
- }
-
- public DefaultRequestTimeouts(Node node, int stripeCount)
- {
- if (stripeCount > 1024)
- throw new IllegalArgumentException("Far too many stripes requested");
- this.stripes = new Stripe[findNextPositivePowerOfTwo(stripeCount)];
- for (int i = 0 ; i < stripes.length; ++i)
- stripes[i] = new Stripe(node);
- this.node = node;
- }
-
- @Override
- public RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units)
- {
- long now = node.elapsed(MILLISECONDS);
- long deadline = now + Math.max(1, MILLISECONDS.convert(delay, units));
- int i = timeout.stripe() & (stripes.length - 1);
- while (true)
- {
- RegisteredTimeout result = stripes[i].tryRegister(timeout, now, deadline);
- if (result != null)
- return result;
- i = (i + 1) & (stripes.length - 1);
- }
+ super(node, Stripe[]::new, Stripe::new);
}
}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 67bd325f..c8fab6b0 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -62,7 +62,6 @@ import accord.local.Commands;
import accord.local.KeyHistory;
import accord.local.Node;
import accord.local.NodeCommandStoreService;
-import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.RedundantStatus;
@@ -811,7 +810,7 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- public NodeTimeService time()
+ public NodeCommandStoreService node()
{
return commandStore.node;
}
diff --git a/accord-core/src/main/java/accord/impl/RequestCallbacks.java b/accord-core/src/main/java/accord/impl/RequestCallbacks.java
new file mode 100644
index 00000000..f6ea93cc
--- /dev/null
+++ b/accord-core/src/main/java/accord/impl/RequestCallbacks.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package accord.impl;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+
+import accord.local.AgentExecutor;
+import accord.local.Node;
+import accord.local.TimeService;
+import accord.messages.Callback;
+import accord.utils.Invariants;
+import org.agrona.collections.Long2ObjectHashMap;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+public class RequestCallbacks extends AbstractRequestTimeouts<RequestCallbacks.CallbackStripe>
+{
+ public interface CallbackEntry
+ {
+ long registeredAt(TimeUnit units);
+ }
+
+ static class CallbackStripe extends Stripe
+ {
+ protected class RegisteredCallback<T> extends AbstractRegistered implements CallbackEntry
+ {
+ final AgentExecutor executor;
+ final long callbackId;
+ final Callback<T> callback;
+ final Node.Id to;
+ final long registeredAt;
+ final long reportSlowAt;
+ final long reportFailAt;
+
+ public RegisteredCallback(AgentExecutor executor, long callbackId, Callback<T> callback, Node.Id to, long registeredAt, long reportSlowAt, long reportFailAt)
+ {
+ this.callbackId = callbackId;
+ this.executor = executor;
+ this.callback = callback;
+ this.to = to;
+ this.registeredAt = registeredAt;
+ this.reportSlowAt = reportSlowAt;
+ this.reportFailAt = reportFailAt;
+ }
+
+ public long registeredAt(TimeUnit units)
+ {
+ return units.convert(registeredAt, MICROSECONDS);
+ }
+
+ // expects lock to already be held
+ void cancelUnsafe()
+ {
+ if (isInHeap())
+ timeouts.remove(this);
+ }
+
+ @Override
+ public Expiring prepareToExpire(Stripe stripe, long now)
+ {
+ Invariants.checkState(((CallbackStripe)stripe).callbacks.containsKey(callbackId));
+ if (now >= reportFailAt)
+ {
+ ((CallbackStripe)stripe).callbacks.remove(callbackId);
+ return this;
+ }
+
+ stripe.register(this, reportFailAt);
+ return new Expiring()
+ {
+ @Override
+ public Expiring prepareToExpire(Stripe stripe, long now)
+ {
+ return this;
+ }
+
+ @Override
+ public void onExpire(long now)
+ {
+ safeInvoke(RegisteredCallback::unsafeOnSlow, null);
+ }
+ };
+ }
+
+ @Override
+ public void onExpire(long now)
+ {
+ safeInvoke(RegisteredCallback::unsafeOnFailure, new accord.coordinate.Timeout(null, null));
+ }
+
+ private void unsafeOnSuccess(T reply)
+ {
+ callback.onSuccess(to, reply);
+ }
+
+ private void unsafeOnFailure(Throwable reply)
+ {
+ callback.onFailure(to, reply);
+ }
+
+ private void unsafeOnSlow(Object ignore)
+ {
+ callback.onSlowResponse(to);
+ }
+
+ <P> void safeInvoke(BiConsumer<RegisteredCallback<T>, P> invoker, P param)
+ {
+ executor.execute(() -> {
+ try
+ {
+ invoker.accept(this, param);
+ }
+ catch (Throwable t)
+ {
+ try
+ {
+ callback.onCallbackFailure(to, t);
+ }
+ catch (Throwable t2)
+ {
+ t.addSuppressed(t2);
+ executor.agent().onUncaughtException(t);
+ }
+ }
+ });
+ }
+ }
+
+ final Long2ObjectHashMap<RegisteredCallback> callbacks = new Long2ObjectHashMap<>();
+
+ public CallbackStripe(TimeService time)
+ {
+ super(time);
+ }
+
+ <T> RegisteredCallback<T> register(long callbackId, AgentExecutor executor, Callback<T> callback, Node.Id to, long now, long failDeadline)
+ {
+ return register(callbackId, executor, callback, to, now, Long.MAX_VALUE, failDeadline);
+ }
+
+ <T> RegisteredCallback<T> register(long callbackId, AgentExecutor executor, Callback<T> callback, Node.Id to, long now, long reportSlowAt, long reportFailAt)
+ {
+ lock();
+ try
+ {
+ RegisteredCallback<T> registered = new RegisteredCallback<>(executor, callbackId, callback, to, now, reportSlowAt, reportFailAt);
+ Object existing = callbacks.putIfAbsent(callbackId, registered);
+ timeouts.add(Math.min(reportSlowAt, reportFailAt), registered);
+ Invariants.checkState(existing == null);
+ return registered;
+ }
+ finally
+ {
+ unlock(now);
+ }
+ }
+
+ <T> RegisteredCallback<T> onSuccess(long callbackId, Node.Id from, T reply, boolean remove)
+ {
+ return safeInvoke(callbackId, from, reply, RegisteredCallback::unsafeOnSuccess, remove);
+ }
+
+ RegisteredCallback onFailure(long callbackId, Node.Id from, Throwable reply)
+ {
+ return safeInvoke(callbackId, from, reply, RegisteredCallback::unsafeOnFailure, true);
+ }
+
+ private <T, P> RegisteredCallback<T> safeInvoke(long callbackId, Node.Id from, P param, BiConsumer<RegisteredCallback<T>, P> invoker, boolean remove)
+ {
+ lock();
+ try
+ {
+ RegisteredCallback<T> registered = remove ? callbacks.remove(callbackId) : callbacks.get(callbackId);
+ if (registered == null)
+ return null;
+
+ if (remove) registered.cancelUnsafe();
+ Invariants.checkState(registered.to.equals(from));
+ registered.safeInvoke(invoker, param);
+ return registered;
+ }
+ finally
+ {
+ unlock();
+ }
+ }
+ }
+
+ public RequestCallbacks(TimeService time)
+ {
+ super(time, CallbackStripe[]::new, CallbackStripe::new);
+ }
+
+ public RequestCallbacks(TimeService time, int stripeCount)
+ {
+ super(time, stripeCount, CallbackStripe[]::new, CallbackStripe::new);
+ }
+
+ public <T> void register(long callbackId, AgentExecutor executor, Callback<T> callback, Node.Id to, long failDelay, TimeUnit units)
+ {
+ long now = time.elapsed(MICROSECONDS);
+ long failDeadline = now + units.toMicros(failDelay);
+ stripes[(int)callbackId & (stripes.length - 1)].register(callbackId, executor, callback, to, now, failDeadline);
+ }
+
+ public <T> void register(long callbackId, AgentExecutor executor, Callback<T> callback, Node.Id to, long slowDelay, long failDelay, TimeUnit units)
+ {
+ long now = time.elapsed(MICROSECONDS);
+ long reportFailAt = now + units.toMicros(failDelay);
+ long reportSlowAt = slowDelay >= failDelay ? Long.MAX_VALUE : now + units.toMicros(slowDelay);
+ stripes[(int)callbackId & (stripes.length - 1)].register(callbackId, executor, callback, to, now, reportSlowAt, reportFailAt);
+ }
+
+ public <T> void register(long callbackId, AgentExecutor executor, Callback<T> callback, Node.Id to, long now, long reportSlowAt, long reportFailAt, TimeUnit units)
+ {
+ if (units != MICROSECONDS)
+ {
+ now = units.toMicros(now);
+ reportSlowAt = reportSlowAt >= reportFailAt ? Long.MAX_VALUE : units.toMicros(reportSlowAt);
+ reportFailAt = units.toMicros(reportFailAt);
+ }
+ stripes[(int)callbackId & (stripes.length - 1)].register(callbackId, executor, callback, to, now, reportSlowAt, reportFailAt);
+ }
+
+ public <T> CallbackEntry onSuccess(long callbackId, Node.Id from, T reply, boolean remove)
+ {
+ return stripes[(int)callbackId & (stripes.length - 1)].onSuccess(callbackId, from, reply, remove);
+ }
+
+ public CallbackEntry onFailure(long callbackId, Node.Id from, Throwable reply)
+ {
+ return stripes[(int)callbackId & (stripes.length - 1)].onFailure(callbackId, from, reply);
+ }
+}
diff --git a/accord-core/src/main/java/accord/impl/SizeOfIntersectionSorter.java b/accord-core/src/main/java/accord/impl/SizeOfIntersectionSorter.java
index 90c6427a..1a9f8620 100644
--- a/accord-core/src/main/java/accord/impl/SizeOfIntersectionSorter.java
+++ b/accord-core/src/main/java/accord/impl/SizeOfIntersectionSorter.java
@@ -62,6 +62,12 @@ public class SizeOfIntersectionSorter implements TopologySorter
return count1 - count2;
}
+ @Override
+ public boolean isFaulty(Node.Id node)
+ {
+ return false;
+ }
+
private static int count(Node.Id node, ShardSelection shards, int offset, Topology topology)
{
return topology.foldlIntOn(node, (shard, v, index) -> shard.get(index) ? v + 1 : v, shards, offset, 0, 0);
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 9d1aea56..06b65370 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -22,7 +22,6 @@ import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
-import java.util.function.Function;
import javax.annotation.Nullable;
import org.slf4j.Logger;
@@ -34,7 +33,6 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommonAttributes;
import accord.local.Node;
-import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.primitives.SaveStatus;
@@ -68,8 +66,6 @@ import static accord.primitives.Status.PreCommitted;
import static accord.utils.ArrayBuffers.cachedAny;
import static accord.utils.btree.UpdateFunction.noOpReplace;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
public class DefaultProgressLog implements ProgressLog, Runnable
{
@@ -82,7 +78,6 @@ public class DefaultProgressLog implements ProgressLog, Runnable
private Object[] progressTokenMap = BTree.empty();
final LogGroupTimers<TxnState> timers = new LogGroupTimers<>(MICROSECONDS);
- final LogGroupTimers<TxnState>.Scheduling timerScheduling;
/**
* A collection of active callbacks (waiting remote replies) or submitted run invocations
@@ -109,13 +104,6 @@ public class DefaultProgressLog implements ProgressLog, Runnable
{
this.node = node;
this.commandStore = commandStore;
- Function<TxnState, Runnable> taskFactory = next -> {
- PreLoadContext context = next == null ? PreLoadContext.empty() : PreLoadContext.contextFor(next.txnId);
- return () -> commandStore.execute(context, safeStore -> run())
- .begin(commandStore.agent());
- };
- this.timerScheduling = timers.new Scheduling(node.scheduler(), taskFactory,
- MILLISECONDS.toMicros(10), 100, SECONDS.toMicros(1L));
}
Node node()
@@ -124,20 +112,13 @@ public class DefaultProgressLog implements ProgressLog, Runnable
}
void update(long deadline, TxnState timer)
- {
- update(node.elapsed(MICROSECONDS), deadline, timer);
- }
-
- void update(long nowMicros, long deadline, TxnState timer)
{
timers.update(deadline, timer);
- maybeReschedule(nowMicros, deadline);
}
- void add(long nowMicros, long deadline, TxnState timer)
+ void add(long deadline, TxnState timer)
{
timers.add(deadline, timer);
- maybeReschedule(nowMicros, deadline);
}
@Nullable TxnState get(TxnId txnId)
@@ -292,7 +273,6 @@ public class DefaultProgressLog implements ProgressLog, Runnable
public void clear()
{
timers.clear();
- timerScheduling.clear();
stateMap = BTree.empty();
progressTokenMap = BTree.empty();
@@ -371,16 +351,6 @@ public class DefaultProgressLog implements ProgressLog, Runnable
state.waiting().setBlockedUntil(safeStore, this, blockedUntil);
}
- void ensureScheduled(long nowMicros)
- {
- timerScheduling.ensureScheduled(nowMicros);
- }
-
- void maybeReschedule(long nowMicros, long newDeadlineMicros)
- {
- timerScheduling.maybeReschedule(nowMicros, newDeadlineMicros);
- }
-
@Override
public void run()
{
@@ -404,11 +374,6 @@ public class DefaultProgressLog implements ProgressLog, Runnable
{
node.agent().onUncaughtException(t);
}
- finally
- {
- if (!timers.isEmpty())
- ensureScheduled(nowMicros);
- }
}
private void addToRunBuffer(TxnState add)
@@ -573,4 +538,18 @@ public class DefaultProgressLog implements ProgressLog, Runnable
TxnState state = get(txnId);
return state != null && !state.isWaitingDone();
}
+
+ public void maybeNotify()
+ {
+ if (commandStore.inStore())
+ {
+ run();
+ }
+ else
+ {
+ long now = node.elapsed(MICROSECONDS);
+ if (timers.shouldWake(now))
+ commandStore.execute(this);
+ }
+ }
}
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 95ef3845..c189f6a2 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
@@ -115,20 +115,20 @@ final class TxnState extends HomeState
if (otherDeadline == 0)
{
setScheduledTimer(updated);
- if (previousDeadline > 0) instance.update(nowMicros, newDeadline, this);
- else instance.add(nowMicros, newDeadline, this);
+ if (previousDeadline > 0) instance.update(newDeadline, this);
+ else instance.add(newDeadline, this);
}
else if (newDeadline < otherDeadline)
{
setScheduledTimer(updated);
setPendingTimerDelay(Ints.saturatedCast(otherDeadline - newDeadline));
- instance.update(nowMicros, newDeadline, this);
+ instance.update(newDeadline, this);
}
else
{
setScheduledTimer(updated.other());
setPendingTimerDelay(Ints.saturatedCast(Math.max(1, newDeadline - otherDeadline)));
- instance.update(nowMicros, otherDeadline, this);
+ instance.update(otherDeadline, this);
}
}
}
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index afbafc71..5a4d17f9 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -341,14 +341,14 @@ public abstract class CommandStore implements AgentExecutor
*/
final Timestamp preaccept(TxnId txnId, Routables<?> keys, SafeCommandStore safeStore, boolean permitFastPath)
{
- NodeTimeService time = safeStore.time();
+ NodeCommandStoreService node = safeStore.node();
- boolean isExpired = time.now() - txnId.hlc() >= safeStore.preAcceptTimeout() && !txnId.isSyncPoint();
+ boolean isExpired = node.now() - txnId.hlc() >= safeStore.preAcceptTimeout() && !txnId.isSyncPoint();
if (rejectBefore != null && !isExpired)
isExpired = rejectBefore.rejects(txnId, keys);
if (isExpired)
- return time.uniqueNow(txnId).asRejected();
+ return node.uniqueNow(txnId).asRejected();
if (txnId.is(ExclusiveSyncPoint))
return txnId;
@@ -357,10 +357,10 @@ public abstract class CommandStore implements AgentExecutor
// this permits us to agree fast path decisions across epoch changes
// TODO (expected): we should (perhaps) separate conflicts for reads and writes
Timestamp min = TxnId.max(txnId, maxConflicts.get(keys));
- if (permitFastPath && txnId == min && txnId.epoch() >= time.epoch())
+ if (permitFastPath && txnId == min && txnId.epoch() >= node.epoch())
return txnId;
- return time.uniqueNow(min);
+ return node.uniqueNow(min);
}
public Timestamp maxConflict(Routables<?> keysOrRanges)
diff --git a/accord-core/src/main/java/accord/local/CommandStores.java b/accord-core/src/main/java/accord/local/CommandStores.java
index bc42d407..c32b9ece 100644
--- a/accord-core/src/main/java/accord/local/CommandStores.java
+++ b/accord-core/src/main/java/accord/local/CommandStores.java
@@ -538,6 +538,13 @@ public abstract class CommandStores
return AsyncChains.reduce(list, (a, b) -> null);
}
+ public void forEachCommandStore(Consumer<CommandStore> forEach)
+ {
+ Snapshot snapshot = current;
+ for (ShardHolder shard : snapshot.shards)
+ forEach.accept(shard.store);
+ }
+
public AsyncChain<Void> ifLocal(PreLoadContext context, RoutingKey key, long minEpoch, long maxEpoch, Consumer<SafeCommandStore> forEach)
{
return forEach(context, RoutingKeys.of(key), minEpoch, maxEpoch, forEach, false);
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 76c74a86..3b57f409 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -508,7 +508,7 @@ public class Commands
CommandStore unsafeStore = safeStore.commandStore();
// TODO (required, API): do we care about tracking the write persistence latency, when this is just a memtable write?
// the only reason it will be slow is because Memtable flushes are backed-up (which will be reported elsewhere)
- long t0 = safeStore.time().now();
+ long t0 = safeStore.node().now();
return command.writes().apply(safeStore, applyRanges(safeStore, command.executeAt()), command.partialTxn())
.flatMap(unused -> unsafeStore.submit(context, ss -> {
Command cmd = ss.get(txnId).current();
diff --git a/accord-core/src/main/java/accord/local/Node.java b/accord-core/src/main/java/accord/local/Node.java
index ef8394d3..ee48e239 100644
--- a/accord-core/src/main/java/accord/local/Node.java
+++ b/accord-core/src/main/java/accord/local/Node.java
@@ -32,9 +32,7 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
-import java.util.function.LongSupplier;
import java.util.function.Supplier;
-import java.util.function.ToLongFunction;
import javax.annotation.Nonnull;
import com.google.common.annotations.VisibleForTesting;
@@ -100,6 +98,8 @@ import net.nicoulaj.compilecommand.annotations.Inline;
import static accord.utils.Invariants.illegalState;
import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
public class Node implements ConfigurationService.Listener, NodeCommandStoreService
{
@@ -162,8 +162,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
private final CommandStores commandStores;
private final CoordinationAdapter.Factory coordinationAdapters;
- private final LongSupplier nowSupplier;
- private final ToLongFunction<TimeUnit> elapsed;
+ private final TimeService time;
private final AtomicReference<Timestamp> now;
private final Agent agent;
private final RandomSource random;
@@ -180,7 +179,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
private final PersistentField<DurableBefore, DurableBefore> persistDurableBefore;
public Node(Id id, MessageSink messageSink,
- ConfigurationService configService, LongSupplier nowSupplier, ToLongFunction<TimeUnit> elapsed,
+ ConfigurationService configService, TimeService time,
Supplier<DataStore> dataSupplier, ShardDistributor shardDistributor, Agent agent, RandomSource random, Scheduler scheduler, TopologySorter.Supplier topologySorter,
Function<Node, RemoteListeners> remoteListenersFactory, Function<Node, RequestTimeouts> requestTimeoutsFactory, Function<Node, ProgressLog.Factory> progressLogFactory,
Function<Node, LocalListeners.Factory> localListenersFactory, CommandStores.Factory factory, CoordinationAdapter.Factory coordinationAdapters,
@@ -193,18 +192,19 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
this.messageSink = messageSink;
this.configService = configService;
this.coordinationAdapters = coordinationAdapters;
- this.topology = new TopologyManager(topologySorter, agent, id, scheduler, elapsed, localConfig);
+ this.topology = new TopologyManager(topologySorter, agent, id, scheduler, time, localConfig);
topology.scheduleTopologyUpdateWatchdog();
this.listeners = remoteListenersFactory.apply(this);
this.timeouts = requestTimeoutsFactory.apply(this);
- this.nowSupplier = nowSupplier;
- this.elapsed = elapsed;
- this.now = new AtomicReference<>(Timestamp.fromValues(topology.epoch(), nowSupplier.getAsLong(), id));
+ this.time = time;
+ this.now = new AtomicReference<>(Timestamp.fromValues(topology.epoch(), time.now(), id));
this.agent = agent;
this.random = random;
this.persistDurableBefore = new PersistentField<>(() -> durableBefore, DurableBefore::merge, durableBeforePersister, this::setPersistedDurableBefore);
this.commandStores = factory.create(this, agent, dataSupplier.get(), random.fork(), shardDistributor, progressLogFactory.apply(this), localListenersFactory.apply(this));
- // TODO review these leak a reference to an object that hasn't finished construction, possibly to other threads
+ // TODO (desired): make frequency configurable
+ scheduler.recurring(() -> commandStores.forEachCommandStore(store -> store.progressLog.maybeNotify()), 1, SECONDS);
+ scheduler.recurring(timeouts::maybeNotify, 100, MILLISECONDS);
configService.registerListener(this);
}
@@ -432,7 +432,7 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
while (true)
{
Timestamp cur = now.get();
- Timestamp next = cur.withNextHlc(nowSupplier.getAsLong())
+ Timestamp next = cur.withNextHlc(time.now())
.withEpochAtLeast(topology.epoch());
if (now.compareAndSet(cur, next))
@@ -454,12 +454,6 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return uniqueNow();
}
- @Override
- public long elapsed(TimeUnit timeUnit)
- {
- return elapsed.applyAsLong(timeUnit);
- }
-
private static Timestamp nowAtLeast(Timestamp current, Timestamp proposed)
{
long currentEpoch = current.epoch(), proposedEpoch = proposed.epoch();
@@ -484,7 +478,13 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
@Override
public long now()
{
- return nowSupplier.getAsLong();
+ return time.now();
+ }
+
+ @Override
+ public long elapsed(TimeUnit timeUnit)
+ {
+ return time.elapsed(timeUnit);
}
public AsyncChain<Void> forEachLocal(PreLoadContext context, Unseekables<?> unseekables, long minEpoch, long maxEpoch, Consumer<SafeCommandStore> forEach)
@@ -832,8 +832,8 @@ public class Node implements ConfigurationService.Listener, NodeCommandStoreServ
return commandStores.current.shards[index].store;
}
- public LongSupplier unsafeGetNowSupplier()
+ public TimeService time()
{
- return nowSupplier;
+ return time;
}
}
diff --git a/accord-core/src/main/java/accord/local/NodeCommandStoreService.java b/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
index 5a07f81f..bbf6d80d 100644
--- a/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
+++ b/accord-core/src/main/java/accord/local/NodeCommandStoreService.java
@@ -18,8 +18,13 @@
package accord.local;
-public interface NodeCommandStoreService extends NodeTimeService
+import accord.primitives.Timestamp;
+
+public interface NodeCommandStoreService extends TimeService
{
+ long epoch();
Node.Id id();
DurableBefore durableBefore();
+ Timestamp uniqueNow();
+ Timestamp uniqueNow(Timestamp atLeast);
}
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 950aa76c..1deebf83 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -355,7 +355,7 @@ public abstract class SafeCommandStore
public abstract DataStore dataStore();
public abstract Agent agent();
public abstract ProgressLog progressLog();
- public abstract NodeTimeService time();
+ public abstract NodeCommandStoreService node();
public abstract CommandStores.RangesForEpoch ranges();
protected NavigableMap<TxnId, Ranges> bootstrapBeganAt()
diff --git a/accord-core/src/main/java/accord/local/NodeTimeService.java b/accord-core/src/main/java/accord/local/TimeService.java
similarity index 84%
rename from accord-core/src/main/java/accord/local/NodeTimeService.java
rename to accord-core/src/main/java/accord/local/TimeService.java
index acab3e2f..b9a918f2 100644
--- a/accord-core/src/main/java/accord/local/NodeTimeService.java
+++ b/accord-core/src/main/java/accord/local/TimeService.java
@@ -24,22 +24,42 @@ import java.util.function.ToLongFunction;
import com.google.common.annotations.VisibleForTesting;
-import accord.primitives.Timestamp;
-
-public interface NodeTimeService
+public interface TimeService
{
- long epoch();
/**
* Current time in some time unit that may be simulated and not match system time
*/
long now();
+
/**
* Return the current time since some arbitrary epoch in the specified time unit. May be simulated time and not
* real time. This clock should not go backwards, nor should it generally correct for clock skew - it should
* only track time elapsed between two points.
*/
long elapsed(TimeUnit unit);
- Timestamp uniqueNow(Timestamp atLeast);
+
+ static TimeService ofNonMonotonic(LongSupplier now, TimeUnit units)
+ {
+ return of(now, elapsedWrapperFromNonMonotonicSource(units, now));
+ }
+
+ static TimeService of(LongSupplier now, ToLongFunction<TimeUnit> elapsed)
+ {
+ return new TimeService()
+ {
+ @Override
+ public long now()
+ {
+ return now.getAsLong();
+ }
+
+ @Override
+ public long elapsed(TimeUnit unit)
+ {
+ return elapsed.applyAsLong(unit);
+ }
+ };
+ }
static ToLongFunction<TimeUnit> elapsedWrapperFromMonotonicSource(TimeUnit sourceUnit, LongSupplier monotonicNowSupplier)
{
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 906e016e..a5e5d056 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -342,7 +342,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
private TxnInfo(TxnId txnId, int encodedStatus, Timestamp executeAt)
{
super(txnId);
- Invariants.checkState(executeAt == txnId || executeAt.getClass() != TxnId.class);
+ Invariants.checkState(executeAt == txnId || !executeAt.equals(txnId));
this.encodedStatus = encodedStatus;
this.executeAt = executeAt == txnId ? this : executeAt;
}
@@ -351,7 +351,7 @@ public class CommandsForKey extends CommandsForKeyUpdate implements CommandsSumm
{
Timestamp executeAt = txnId;
if (status.hasExecuteAt()) executeAt = command.executeAt();
- if (executeAt.getClass() == TxnId.class) executeAt = txnId;
+ if (executeAt.equals(txnId)) executeAt = txnId;
Ballot ballot;
int encodedStatus = encode(txnId, status, mayExecute);
if (!status.hasBallot || (ballot = command.acceptedOrCommitted()).equals(Ballot.ZERO))
diff --git a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
index 08a7d686..354989cf 100644
--- a/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
+++ b/accord-core/src/main/java/accord/messages/GetEphemeralReadDeps.java
@@ -72,7 +72,7 @@ public class GetEphemeralReadDeps extends TxnRequest.WithUnsynced<GetEphemeralRe
StoreParticipants participants = StoreParticipants.read(safeStore, scope, txnId, minEpoch, Long.MAX_VALUE);
Deps deps = calculateDeps(safeStore, txnId, participants, constant(minEpoch), Timestamp.MAX);
- return new GetEphemeralReadDepsOk(deps, Math.max(safeStore.time().epoch(), node.epoch()));
+ return new GetEphemeralReadDepsOk(deps, Math.max(safeStore.node().epoch(), node.epoch()));
}
@Override
diff --git a/accord-core/src/main/java/accord/messages/GetMaxConflict.java b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
index 6ee10c40..d21ebdeb 100644
--- a/accord-core/src/main/java/accord/messages/GetMaxConflict.java
+++ b/accord-core/src/main/java/accord/messages/GetMaxConflict.java
@@ -65,7 +65,7 @@ public class GetMaxConflict extends TxnRequest.WithUnsynced<GetMaxConflict.GetMa
public GetMaxConflictOk apply(SafeCommandStore safeStore)
{
Timestamp maxConflict = safeStore.commandStore().maxConflict(scope);
- return new GetMaxConflictOk(maxConflict, Math.max(safeStore.time().epoch(), node.epoch()));
+ return new GetMaxConflictOk(maxConflict, Math.max(safeStore.node().epoch(), node.epoch()));
}
@Override
diff --git a/accord-core/src/main/java/accord/topology/Shard.java b/accord-core/src/main/java/accord/topology/Shard.java
index f95d3d40..35893dc6 100644
--- a/accord-core/src/main/java/accord/topology/Shard.java
+++ b/accord-core/src/main/java/accord/topology/Shard.java
@@ -43,6 +43,7 @@ public class Shard
public final int maxFailures;
public final int recoveryFastPathSize;
public final int fastPathQuorumSize;
+ public final int fastPathRejectSize;
public final int slowPathQuorumSize;
public final boolean pendingRemoval;
@@ -58,6 +59,7 @@ public class Shard
this.recoveryFastPathSize = (maxFailures+1)/2;
this.slowPathQuorumSize = slowPathQuorumSize(nodes.size());
this.fastPathQuorumSize = fastPathQuorumSize(nodes.size(), e, maxFailures);
+ this.fastPathRejectSize = 1 + fastPathElectorate.size() - fastPathQuorumSize;
this.pendingRemoval = pendingRemoval;
}
@@ -84,11 +86,6 @@ public class Shard
return (f + electorate)/2 + 1;
}
- public boolean rejectsFastPath(int rejectCount)
- {
- return rejectCount > fastPathElectorate.size() - fastPathQuorumSize;
- }
-
public static int slowPathQuorumSize(int replicas)
{
return replicas - maxToleratedFailures(replicas);
diff --git a/accord-core/src/main/java/accord/topology/Topologies.java b/accord-core/src/main/java/accord/topology/Topologies.java
index 48cedae6..c64f7e9d 100644
--- a/accord-core/src/main/java/accord/topology/Topologies.java
+++ b/accord-core/src/main/java/accord/topology/Topologies.java
@@ -30,7 +30,6 @@ import accord.utils.IndexedConsumer;
import accord.utils.Invariants;
import accord.utils.SortedArrays;
import accord.utils.SortedArrays.SortedArrayList;
-import accord.utils.SortedList;
import static accord.utils.Invariants.illegalState;
import static accord.utils.SortedArrays.isSortedUnique;
@@ -72,12 +71,9 @@ public interface Topologies extends TopologySorter
*/
SortedArrayList<Id> nodes();
- default SortedList<Node.Id> nonStaleNodes()
+ default SortedArrayList<Id> staleNodes()
{
- SortedArrayList<Node.Id> staleIds = current().staleIds;
- if (staleIds.isEmpty())
- return nodes();
- return nodes().without(staleIds);
+ return current().staleNodes;
}
Ranges computeRangesForNode(Id node);
@@ -253,6 +249,12 @@ public interface Topologies extends TopologySorter
{
return sorter.compare(node1, node2, shards);
}
+
+ @Override
+ public boolean isFaulty(Id node)
+ {
+ return sorter.isFaulty(node);
+ }
}
class Multi implements Topologies
@@ -426,6 +428,12 @@ public interface Topologies extends TopologySorter
{
return sorter.compare(node1, node2, shards);
}
+
+ @Override
+ public boolean isFaulty(Id node)
+ {
+ return sorter.isFaulty(node);
+ }
}
class Builder
diff --git a/accord-core/src/main/java/accord/topology/Topology.java b/accord-core/src/main/java/accord/topology/Topology.java
index d834b4db..a427901f 100644
--- a/accord-core/src/main/java/accord/topology/Topology.java
+++ b/accord-core/src/main/java/accord/topology/Topology.java
@@ -65,7 +65,7 @@ public class Topology
public static final Topology EMPTY = new Topology(null, EMPTY_EPOCH, new Shard[0], Ranges.EMPTY, NO_IDS, NO_IDS, new Int2ObjectHashMap<>(0, 0.9f), Ranges.EMPTY, EMPTY_SUBSET);
final long epoch;
- final SortedArrayList<Id> staleIds;
+ final SortedArrayList<Id> staleNodes;
final Shard[] shards;
final Ranges ranges;
@@ -126,11 +126,11 @@ public class Topology
this(epoch, NO_IDS, shards);
}
- public Topology(long epoch, SortedArrayList<Id> staleIds, Shard... shards)
+ public Topology(long epoch, SortedArrayList<Id> staleNodes, Shard... shards)
{
this.global = null;
this.epoch = epoch;
- this.staleIds = staleIds;
+ this.staleNodes = staleNodes;
this.ranges = Ranges.ofSortedAndDeoverlapped(Arrays.stream(shards).map(shard -> shard.range).toArray(Range[]::new));
this.shards = shards;
this.subsetOfRanges = ranges;
@@ -156,11 +156,11 @@ public class Topology
}
@VisibleForTesting
- Topology(@Nullable Topology global, long epoch, Shard[] shards, Ranges ranges, SortedArrayList<Id> staleIds, SortedArrayList<Id> nodeIds, Int2ObjectHashMap<NodeInfo> nodeById, Ranges subsetOfRanges, int[] supersetIndexes)
+ Topology(@Nullable Topology global, long epoch, Shard[] shards, Ranges ranges, SortedArrayList<Id> staleNodes, SortedArrayList<Id> nodeIds, Int2ObjectHashMap<NodeInfo> nodeById, Ranges subsetOfRanges, int[] supersetIndexes)
{
this.global = global;
this.epoch = epoch;
- this.staleIds = staleIds;
+ this.staleNodes = staleNodes;
this.shards = shards;
this.ranges = ranges;
this.nodeIds = nodeIds;
@@ -235,12 +235,12 @@ public class Topology
return Topology.EMPTY;
SortedArrayList<Id> nodeIds = new SortedArrayList<>(new Id[] { node });
- return new Topology(global(), epoch, shards, ranges, staleIds, nodeIds, nodeLookup, info.ranges, info.supersetIndexes);
+ return new Topology(global(), epoch, shards, ranges, staleNodes, nodeIds, nodeLookup, info.ranges, info.supersetIndexes);
}
public Topology trim()
{
- return select(epoch, staleIds, shards, this.supersetIndexes);
+ return select(epoch, staleNodes, shards, this.supersetIndexes);
}
public Ranges rangesForNode(Id node)
@@ -299,7 +299,7 @@ public class Topology
int count = 0;
for (int i = nodes.firstSetBit() ; i >= 0 ; i = nodes.nextSetBit(i + 1, -1))
nodeIds[count++] = this.nodeIds.get(i);
- return new Topology(global(), epoch, shards, ranges, staleIds, new SortedArrayList<>(nodeIds), nodeLookup, rangeSubset, newSubset);
+ return new Topology(global(), epoch, shards, ranges, staleNodes, new SortedArrayList<>(nodeIds), nodeLookup, rangeSubset, newSubset);
}
@VisibleForTesting
@@ -318,7 +318,7 @@ public class Topology
if (nodeLookup.size() != nodeIds.length)
nodeIds = Arrays.copyOf(nodeIds, nodeLookup.size());
Arrays.sort(nodeIds);
- return new Topology(global(), epoch, shards, ranges, staleIds, new SortedArrayList<>(nodeIds), nodeLookup, rangeSubset, newSubset);
+ return new Topology(global(), epoch, shards, ranges, staleNodes, new SortedArrayList<>(nodeIds), nodeLookup, rangeSubset, newSubset);
}
private int[] subsetFor(Routables<?> select, boolean permitMissing)
@@ -577,7 +577,7 @@ public class Topology
public Set<Id> staleIds()
{
- return staleIds;
+ return staleNodes;
}
public Shard[] unsafeGetShards()
diff --git a/accord-core/src/main/java/accord/topology/TopologyManager.java b/accord-core/src/main/java/accord/topology/TopologyManager.java
index c0bd2c90..af74ba09 100644
--- a/accord-core/src/main/java/accord/topology/TopologyManager.java
+++ b/accord-core/src/main/java/accord/topology/TopologyManager.java
@@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
-import java.util.function.ToLongFunction;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
@@ -46,6 +45,7 @@ import accord.coordinate.TopologyMismatch;
import accord.coordinate.tracking.QuorumTracker;
import accord.local.CommandStore;
import accord.local.Node.Id;
+import accord.local.TimeService;
import accord.primitives.EpochSupplier;
import accord.primitives.Ranges;
import accord.primitives.Routables;
@@ -251,12 +251,12 @@ public class TopologyManager
this(epochs, new ArrayList<>(), new ArrayList<>());
}
- private FutureEpoch awaitEpoch(long epoch, Agent agent, ToLongFunction<TimeUnit> nowTimeUnit, LocalConfig localConfig)
+ private FutureEpoch awaitEpoch(long epoch, Agent agent, TimeService time, LocalConfig localConfig)
{
if (epoch <= currentEpoch)
return SUCCESS;
- long now = nowTimeUnit.applyAsLong(TimeUnit.MILLISECONDS);
+ long now = time.elapsed(TimeUnit.MILLISECONDS);
long deadline = now + localConfig.epochFetchInitialTimeout().toMillis();
int diff = (int) (epoch - currentEpoch);
long epochIndex = epoch - diff + 1;
@@ -444,19 +444,19 @@ public class TopologyManager
private final Agent agent;
private final Id node;
private final Scheduler scheduler;
- private final ToLongFunction<TimeUnit> nowTimeUnit;
+ private final TimeService time;
private volatile Epochs epochs;
private Scheduler.Scheduled topologyUpdateWatchdog;
private final LocalConfig localConfig;
- public TopologyManager(TopologySorter.Supplier sorter, Agent agent, Id node, Scheduler scheduler, ToLongFunction<TimeUnit> nowTimeUnit, LocalConfig localConfig)
+ public TopologyManager(TopologySorter.Supplier sorter, Agent agent, Id node, Scheduler scheduler, TimeService time, LocalConfig localConfig)
{
this.sorter = sorter;
this.agent = agent;
this.node = node;
this.scheduler = scheduler;
- this.nowTimeUnit = nowTimeUnit;
+ this.time = time;
this.epochs = Epochs.EMPTY;
this.localConfig = localConfig;
}
@@ -475,7 +475,7 @@ public class TopologyManager
if (current.futureEpochs.isEmpty())
return;
- long now = nowTimeUnit.applyAsLong(TimeUnit.MILLISECONDS);
+ long now = time.elapsed(TimeUnit.MILLISECONDS);
if (now > current.futureEpochs.get(0).deadlineMillis)
{
for (int i = 0; i < current.futureEpochs.size(); i++)
@@ -523,7 +523,7 @@ public class TopologyManager
AsyncResult<Void> result = null;
synchronized (this)
{
- result = epochs.awaitEpoch(epoch, agent, nowTimeUnit, localConfig).future;
+ result = epochs.awaitEpoch(epoch, agent, time, localConfig).future;
}
CommandStore current = CommandStore.maybeCurrent();
return current == null || result.isDone() ? result : result.withExecutor(current);
diff --git a/accord-core/src/main/java/accord/utils/ArrayBuffers.java b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
index 1bca73d3..455db381 100644
--- a/accord-core/src/main/java/accord/utils/ArrayBuffers.java
+++ b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
@@ -91,6 +91,12 @@ public class ArrayBuffers
return OBJECTS.get();
}
+ public static <T> ObjectBuffers<T> uncached(T[] ofType)
+ {
+ Class<T> clazz = (Class<T>) ofType.getClass().getComponentType();
+ return uncached(size -> (T[])Array.newInstance(clazz, size));
+ }
+
public static <T> ObjectBuffers<T> uncached(IntFunction<T[]> allocator) { return new UncachedObjectBuffers<>(allocator); }
public static IntBuffers uncachedInts() { return UncachedIntBuffers.INSTANCE; }
@@ -797,6 +803,16 @@ public class ArrayBuffers
return (E) buffer[index];
}
+ @Override
+ public E set(int index, E value)
+ {
+ if (index >= size)
+ throw new IndexOutOfBoundsException();
+ E prev = (E) buffer[index];
+ buffer[index] = value;
+ return prev;
+ }
+
@Override
public int size()
{
diff --git a/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java b/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
index ff6229fa..63c867e0 100644
--- a/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
+++ b/accord-core/src/main/java/accord/utils/IntrusivePriorityHeap.java
@@ -110,23 +110,19 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
protected N peekNode()
{
- if (heapifiedSize < size)
- heapify();
-
if (size == 0)
return null;
+ Invariants.checkState(heapifiedSize == size);
return (N) heap[0];
}
protected N pollNode()
{
- if (heapifiedSize < size)
- heapify();
-
if (size == 0)
return null;
+ Invariants.checkState(isHeapified());
N result = (N) heap[0];
result.heapIndex = -1;
@@ -251,6 +247,11 @@ public abstract class IntrusivePriorityHeap<N extends IntrusivePriorityHeap.Node
heapify();
}
+ protected boolean isHeapified()
+ {
+ return heapifiedSize == size;
+ }
+
protected void clear()
{
Arrays.fill(heap, 0, size, null);
diff --git a/accord-core/src/main/java/accord/utils/LogGroupTimers.java b/accord-core/src/main/java/accord/utils/LogGroupTimers.java
index 8c59cbc4..15274140 100644
--- a/accord-core/src/main/java/accord/utils/LogGroupTimers.java
+++ b/accord-core/src/main/java/accord/utils/LogGroupTimers.java
@@ -21,9 +21,6 @@ package accord.utils;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
-import java.util.function.Function;
-
-import accord.api.Scheduler;
/**
* Basic idea is we collect timers in buckets that are logarithmically/exponentially spaced,
@@ -63,91 +60,12 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
public static class Timer extends IntrusivePriorityHeap.Node
{
private long deadline;
- protected long deadline()
+ protected final long deadline()
{
return deadline;
}
}
- public class Scheduling
- {
- final Scheduler scheduler;
- final Function<T, Runnable> taskFactory;
- final long schedulerImpreciseLateTolerance;
- final long schedulerPreciseLateTolerance;
- final long preciseDelayThreshold;
-
- Scheduler.Scheduled scheduled = Scheduler.CANCELLED;
- long lastNow; // now can go backwards in time since we calculate it before taking the lock
- long scheduledAt = Long.MAX_VALUE;
-
- /**
- * Note that the parameter to taskFactory may be null
- */
- public Scheduling(Scheduler scheduler, Function<T, Runnable> taskFactory, long schedulerImpreciseLateTolerance, long schedulerPreciseLateTolerance, long preciseDelayThreshold)
- {
- this.scheduler = scheduler;
- this.taskFactory = taskFactory;
- this.schedulerImpreciseLateTolerance = schedulerImpreciseLateTolerance;
- this.schedulerPreciseLateTolerance = schedulerPreciseLateTolerance;
- this.preciseDelayThreshold = preciseDelayThreshold;
- }
-
- public void ensureScheduled(long now)
- {
- lastNow = now = Math.max(lastNow, now);
- T next = peekIfSoonerThan(now + preciseDelayThreshold);
- long runAt;
- if (next == null)
- {
- runAt = nextDeadlineEpoch();
- if (runAt < 0)
- return;
-
- // when scheduling an imprecise run time, we don't mind if we're late by some amount
- // (often we will be early)
- runAt = Math.max(now, runAt + schedulerImpreciseLateTolerance);
- }
- else
- {
- runAt = Math.max(now, next.deadline());
- }
-
- if (!scheduled.isDone())
- {
- if (runAt > scheduledAt - schedulerPreciseLateTolerance)
- return;
-
- scheduled.cancel();
- }
-
- scheduledAt = runAt;
- lastNow = now;
- long delay = Math.max(0, runAt - now);
- scheduled = scheduler.once(taskFactory.apply(next), delay, TimeUnit.MICROSECONDS);
- }
-
- public void maybeReschedule(long now, long newDeadline)
- {
- if (scheduled.isDone() || newDeadline < scheduledAt - schedulerPreciseLateTolerance)
- ensureScheduled(now);
- }
-
- public void clear()
- {
- if (!scheduled.isDone())
- scheduled.cancel();
- scheduled = Scheduler.CANCELLED;
- lastNow = 0;
- scheduledAt = Long.MAX_VALUE;
- }
-
- public long scheduledAt()
- {
- return scheduledAt;
- }
- }
-
static class Bucket<T extends Timer> extends IntrusivePriorityHeap<T> implements Comparable<Bucket<T>>
{
final LogGroupTimers<T> owner;
@@ -242,7 +160,7 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
int bucketsStart, bucketsEnd;
int timerCount;
- long curEpoch;
+ long curEpoch, wakeAt;
public LogGroupTimers(TimeUnit units)
{
@@ -274,59 +192,35 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
}
}
- public T peekIfSoonerThan(long deadlineThreshold)
- {
- while (bucketsStart < bucketsEnd)
- {
- Bucket<T> head = buckets[bucketsStart];
- if (head.epoch >= deadlineThreshold)
- return null;
-
- head.ensureHeapified();
- if (!head.isEmpty())
- return head.peekNode();
-
- buckets[bucketsStart++] = null;
- if (head == addFinger) addFinger = null;
- }
-
- return null;
- }
-
- /**
- * Return the epoch of the next timer we have to schedule
- */
- public long nextDeadlineEpoch()
- {
- int i = bucketsStart;
- while (i < bucketsEnd)
- {
- Bucket<T> head = buckets[i++];
- if (!head.isEmpty())
- return head.epoch;
- }
-
- return -1L;
- }
-
// unsafe for reentry during advance
public T poll()
{
- while (bucketsStart < bucketsEnd)
+ if (bucketsStart == bucketsEnd)
+ return null;
+
+ Bucket<T> head = buckets[bucketsStart];
+ while (true)
{
- Bucket<T> head = buckets[bucketsStart];
- head.ensureHeapified();
+ head.ensureHeapified(); // must heapify before testing if empty, as may redistribute during heapify
if (!head.isEmpty())
{
--timerCount;
- return head.pollNode();
+ T result = head.pollNode();
+ T next = head.peekNode();
+ if (next == null) wakeAt = head.end();
+ else wakeAt = next.deadline();
+ return result;
}
buckets[bucketsStart++] = null;
if (head == addFinger) addFinger = null;
+ if (bucketsStart == bucketsEnd)
+ {
+ wakeAt = Long.MAX_VALUE;
+ return null;
+ }
+ head = buckets[bucketsStart];
}
-
- return null;
}
/**
@@ -343,26 +237,38 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
curEpoch = nextEpoch;
while (bucketsStart < bucketsEnd)
{
- // drain buckets that are wholly contained by our new time
Bucket<T> head = buckets[bucketsStart];
+ if (head.epoch > now)
+ {
+ wakeAt = Math.max(wakeAt, head.epoch);
+ return;
+ }
+
if (head.epoch + head.span <= now)
{
+ // drain buckets that are wholly contained by our new time, without sorting
timerCount -= head.size;
head.drain(param, expiredTimers);
}
else
{
+ head.ensureHeapified();
T timer;
while (null != (timer = head.peekNode()))
{
- if (timer.deadline() > now)
+ long deadline = timer.deadline();
+ if (deadline > now)
+ {
+ wakeAt = deadline;
return;
+ }
--timerCount;
Invariants.checkState(timer == head.pollNode());
Invariants.checkState(!timer.isInHeap());
expiredTimers.accept(param, timer);
}
+ wakeAt = head.end();
}
Invariants.checkState(head.isEmpty());
@@ -371,23 +277,23 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
if (head == addFinger)
addFinger = null;
}
-
+ wakeAt = Long.MAX_VALUE;
Invariants.checkState(addFinger == null || addFinger == findBucket(addFinger.epoch));
}
public void add(long deadline, T timer)
{
- Invariants.checkState(deadline >= curEpoch);
addInternal(deadline, timer);
++timerCount;
+ refreshWakeAt(Long.MAX_VALUE, deadline);
}
public void update(long deadline, T timer)
{
- Invariants.checkState(deadline >= curEpoch);
Timer t = timer; // cast to access private field
Bucket<T> bucket = findBucket(t.deadline);
Invariants.checkState(bucket != null);
+ long prevDeadline = t.deadline;
if (bucket.contains(deadline))
{
t.deadline = deadline;
@@ -398,6 +304,7 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
bucket.remove(timer);
addInternal(deadline, timer);
}
+ refreshWakeAt(prevDeadline, deadline);
}
private void addInternal(long deadline, T timer)
@@ -417,10 +324,78 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
public void remove(T timer)
{
Timer t = timer; // cast to access private field
+ long prevDeadline = t.deadline;
Bucket<T> bucket = findBucket(t.deadline);
Invariants.checkState(bucket != null);
bucket.remove(timer);
--timerCount;
+ refreshWakeAt(prevDeadline, Long.MAX_VALUE);
+ }
+
+ private void refreshWakeAt(long prevDeadline, long deadline)
+ {
+ if (deadline < wakeAt)
+ {
+ wakeAt = deadline;
+ }
+ else if (prevDeadline == wakeAt && bucketsStart != bucketsEnd)
+ {
+ Bucket<T> head = buckets[bucketsStart];
+ if (!head.isHeapified())
+ head.heapify();
+
+ Timer next = head.peekNode();
+ if (next == null)
+ {
+ if (head.end() >= curEpoch)
+ {
+ wakeAt = head.end();
+ }
+ else
+ {
+ while (true)
+ {
+ buckets[bucketsStart++] = null;
+ if (head == addFinger) addFinger = null;
+ if (bucketsStart == bucketsEnd)
+ {
+ wakeAt = Long.MAX_VALUE;
+ return;
+ }
+ head = buckets[bucketsStart];
+ if (head.end() >= curEpoch)
+ {
+ if (head.epoch >= curEpoch)
+ {
+ wakeAt = head.epoch;
+ return;
+ }
+ else
+ {
+ head.heapify();
+ if (!head.isEmpty())
+ {
+ wakeAt = head.peekNode().deadline();
+ return;
+ }
+ }
+ }
+ Invariants.checkState(head.isEmpty());
+ }
+ }
+ }
+ else wakeAt = next.deadline;
+ }
+ }
+
+ public boolean shouldWake(long now)
+ {
+ return now >= wakeAt;
+ }
+
+ public long wakeAt()
+ {
+ return wakeAt;
}
public int size()
@@ -628,6 +603,7 @@ public class LogGroupTimers<T extends LogGroupTimers.Timer>
}
bucketsStart = bucketsEnd = 0;
curEpoch = 0;
+ wakeAt = Long.MAX_VALUE;
addFinger = null;
}
}
diff --git a/accord-core/src/main/java/accord/utils/SortedArrays.java b/accord-core/src/main/java/accord/utils/SortedArrays.java
index 3e7f2480..1576bf48 100644
--- a/accord-core/src/main/java/accord/utils/SortedArrays.java
+++ b/accord-core/src/main/java/accord/utils/SortedArrays.java
@@ -18,12 +18,14 @@
package accord.utils;
+import java.lang.reflect.Array;
import java.util.AbstractList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.function.IntBinaryOperator;
import java.util.function.IntFunction;
+import java.util.function.Predicate;
import accord.utils.ArrayBuffers.ObjectBuffers;
import accord.utils.ArrayBuffers.IntBufferAllocator;
@@ -32,6 +34,7 @@ import net.nicoulaj.compilecommand.annotations.Inline;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import static accord.utils.ArrayBuffers.cachedAny;
import static accord.utils.ArrayBuffers.uncached;
import static accord.utils.Invariants.checkArgument;
import static accord.utils.Invariants.illegalState;
@@ -64,13 +67,13 @@ public class SortedArrays
return new SortedArrayList<>(items);
}
- public static <T extends Comparable<? super T>> SortedArrayList<T> copySorted(Collection<T> copy, IntFunction<T[]> allocator)
+ public static <T extends Comparable<? super T>> SortedArrayList<T> copySorted(Collection<? extends T> copy, IntFunction<T[]> allocator)
{
T[] array = copy.toArray(allocator);
return new SortedArrayList<>(array);
}
- public static <T extends Comparable<? super T>> SortedArrayList<T> copyUnsorted(Collection<T> copy, IntFunction<T[]> allocator)
+ public static <T extends Comparable<? super T>> SortedArrayList<T> copyUnsorted(Collection<? extends T> copy, IntFunction<T[]> allocator)
{
T[] array = copy.toArray(allocator);
Arrays.sort(array);
@@ -135,9 +138,48 @@ public class SortedArrays
public SortedArrayList<T> without(SortedArrayList<T> without)
{
- T[] array = (T[])SortedArrays.linearSubtract(Comparable::compareTo, this.array, without.array, ArrayBuffers.cachedAny());
+ // TODO (expected): wrap cachedAny with a complete() method that copies to an array of the source type.
+ T[] array = (T[])SortedArrays.linearSubtract(Comparable::compareTo, this.array, without.array, ArrayBuffers.uncached(this.array));
return array == this.array ? this : new SortedArrayList<>(array);
}
+
+ public SortedArrayList<T> with(SortedArrayList<T> with)
+ {
+ T[] array = (T[])SortedArrays.linearUnion(this.array, this.array.length, with.array, with.array.length, Comparable::compareTo, ArrayBuffers.uncached(this.array));
+ return array == this.array ? this : new SortedArrayList<>(array);
+ }
+
+ public SortedArrayList<T> without(Predicate<T> remove)
+ {
+ Object[] buffer = null;
+ int bufferCount = 0;
+ int previ = 0;
+ for (int i = 0 ; i < size() ; ++i)
+ {
+ T test = get(i);
+ if (remove.test(test))
+ {
+ if (buffer == null)
+ buffer = cachedAny().get(size());
+
+ int copyCount = i - previ;
+ System.arraycopy(array, previ, buffer, bufferCount, copyCount);
+ previ = i + 1;
+ bufferCount += copyCount;
+ }
+ }
+
+ if (buffer == null)
+ return this;
+
+ int copyCount = size() - previ;
+ System.arraycopy(array, previ, buffer, bufferCount, copyCount);
+ bufferCount += copyCount;
+
+ T[] newArray = (T[]) Array.newInstance(this.array.getClass().getComponentType(), bufferCount);
+ System.arraycopy(buffer, 0, newArray, 0, bufferCount);
+ return SortedArrayList.ofSorted(newArray);
+ }
}
/**
@@ -183,17 +225,17 @@ public class SortedArrays
return linearUnion(left, leftLength, right, rightLength, Comparable::compareTo, buffers);
}
- public static <T> T[] linearUnion(T[] left, int leftLength, T[] right, int rightLength, AsymmetricComparator<? super T, ? super T> comparator, ObjectBuffers<T> buffers)
+ public static <O, T extends O> O[] linearUnion(T[] left, int leftLength, T[] right, int rightLength, AsymmetricComparator<? super T, ? super T> comparator, ObjectBuffers<O> buffers)
{
return linearUnion(left, 0, leftLength, right, 0, rightLength, comparator, buffers);
}
- public static <T> T[] linearUnion(T[] left, int leftStart, int leftEnd, T[] right, int rightStart, int rightEnd, AsymmetricComparator<? super T, ? super T> comparator, ObjectBuffers<T> buffers)
+ public static <O, T extends O> O[] linearUnion(T[] left, int leftStart, int leftEnd, T[] right, int rightStart, int rightEnd, AsymmetricComparator<? super T, ? super T> comparator, ObjectBuffers<O> buffers)
{
int leftIdx = leftStart;
int rightIdx = rightStart;
- T[] result = null;
+ O[] result = null;
int resultSize = 0;
// first, pick the superset candidate and merge the two until we find the first missing item
diff --git a/accord-core/src/test/java/accord/Utils.java b/accord-core/src/test/java/accord/Utils.java
index a1135e50..69df53a4 100644
--- a/accord-core/src/test/java/accord/Utils.java
+++ b/accord-core/src/test/java/accord/Utils.java
@@ -70,7 +70,6 @@ import accord.utils.ThreadPoolScheduler;
import org.awaitility.Awaitility;
import org.awaitility.core.ThrowingRunnable;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
import static accord.utils.async.AsyncChains.awaitUninterruptibly;
public class Utils
@@ -185,7 +184,6 @@ public class Utils
messageSink,
new MockConfigurationService(messageSink, EpochFunction.noop(), topology),
clock,
- elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, clock),
() -> store,
new ShardDistributor.EvenSplit(8, ignore -> new IntKey.Splitter()),
agent,
@@ -221,6 +219,6 @@ public class Utils
public static TopologyManager testTopologyManager(TopologySorter.Supplier sorter, Id node)
{
- return new TopologyManager(sorter, new TestAgent.RethrowAgent(), node, Scheduler.NEVER_RUN_SCHEDULED, elapsedWrapperFromNonMonotonicSource(TimeUnit.MILLISECONDS, () -> 0), LocalConfig.DEFAULT);
+ return new TopologyManager(sorter, new TestAgent.RethrowAgent(), node, Scheduler.NEVER_RUN_SCHEDULED, new MockCluster.Clock(0), LocalConfig.DEFAULT);
}
}
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 740e06e7..95517ae0 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/FastPathTrackerReconciler.java
@@ -18,7 +18,7 @@
package accord.coordinate.tracking;
-import accord.coordinate.tracking.FastPathTracker.FastPathQuorumShardTracker;
+import accord.coordinate.tracking.FastPathTracker.FastPathShardTracker;
import accord.utils.RandomSource;
import accord.local.Node;
import accord.topology.Topologies;
@@ -26,7 +26,7 @@ import org.junit.jupiter.api.Assertions;
import java.util.ArrayList;
-public class FastPathTrackerReconciler extends TrackerReconciler<FastPathQuorumShardTracker, FastPathTracker, FastPathTrackerReconciler.Rsp>
+public class FastPathTrackerReconciler extends TrackerReconciler<FastPathShardTracker, FastPathTracker, FastPathTrackerReconciler.Rsp>
{
enum Rsp { FAST, SLOW, FAIL }
@@ -58,19 +58,19 @@ public class FastPathTrackerReconciler extends TrackerReconciler<FastPathQuorumS
switch (status)
{
case Failed:
- Assertions.assertTrue(tracker.any(FastPathQuorumShardTracker::hasFailed));
- Assertions.assertFalse(tracker.all(FastPathQuorumShardTracker::hasReachedQuorum));
+ Assertions.assertTrue(tracker.any(FastPathShardTracker::hasFailed));
+ Assertions.assertFalse(tracker.all(FastPathShardTracker::hasReachedQuorum));
break;
case Success:
- Assertions.assertTrue(tracker.all(FastPathQuorumShardTracker::hasReachedQuorum));
+ Assertions.assertTrue(tracker.all(FastPathShardTracker::hasReachedQuorum));
Assertions.assertTrue(tracker.all(shard -> shard.hasRejectedFastPath() || shard.hasMetFastPathCriteria()));
- Assertions.assertFalse(tracker.any(FastPathQuorumShardTracker::hasFailed));
+ Assertions.assertFalse(tracker.any(FastPathShardTracker::hasFailed));
break;
case NoChange:
- Assertions.assertFalse(tracker.all(shard -> shard.hasRejectedFastPath() || shard.hasMetFastPathCriteria()) && tracker.all(FastPathQuorumShardTracker::hasReachedQuorum));
- Assertions.assertFalse(tracker.any(FastPathQuorumShardTracker::hasFailed));
+ Assertions.assertFalse(tracker.all(shard -> shard.hasRejectedFastPath() || shard.hasMetFastPathCriteria()) && tracker.all(FastPathShardTracker::hasReachedQuorum));
+ Assertions.assertFalse(tracker.any(FastPathShardTracker::hasFailed));
}
}
}
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 c912c95b..0f53bd48 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerReconciler.java
@@ -71,7 +71,7 @@ public class ReadTrackerReconciler extends TrackerReconciler<ReadShardTracker, R
default: throw new AssertionError();
case DATA: inflight.remove(from); return tracker.recordReadSuccess(from);
case QUORUM: inflight.remove(from); return tracker.recordQuorumReadSuccess(from);
- case FAIL: inflight.remove(from); return tracker.recordReadFailure(from);
+ case FAIL: inflight.remove(from); return tracker.recordFailure(from);
case SLOW: return tracker.recordSlowResponse(from);
}
}
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 70e0420d..04eecf14 100644
--- a/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java
+++ b/accord-core/src/test/java/accord/coordinate/tracking/ReadTrackerTest.java
@@ -53,7 +53,7 @@ public class ReadTrackerTest
@Override
public RequestStatus trySendMore()
{
- return super.trySendMore(Set::add, inflight);
+ return super.trySendMore((s, i) -> s.add(i.id), inflight);
}
}
@@ -101,7 +101,7 @@ public class ReadTrackerTest
tracker.recordInFlightRead(ids[0]);
assertResponseState(tracker, false, false);
- tracker.recordReadFailure(ids[0]);
+ tracker.recordFailure(ids[0]);
assertResponseState(tracker, false, false);
tracker.recordInFlightRead(ids[1]);
@@ -118,15 +118,15 @@ public class ReadTrackerTest
ReadTracker tracker = new TestReadTracker(topologies(subTopology));
tracker.recordInFlightRead(ids[0]);
- tracker.recordReadFailure(ids[0]);
+ tracker.recordFailure(ids[0]);
assertResponseState(tracker, false, false);
tracker.recordInFlightRead(ids[1]);
- tracker.recordReadFailure(ids[1]);
+ tracker.recordFailure(ids[1]);
assertResponseState(tracker, false, false);
tracker.recordInFlightRead(ids[2]);
- tracker.recordReadFailure(ids[2]);
+ tracker.recordFailure(ids[2]);
assertResponseState(tracker, false, true);
}
@@ -159,13 +159,13 @@ public class ReadTrackerTest
assertResponseState(responses, false, false);
- responses.recordReadFailure(ids[2]);
+ responses.recordFailure(ids[2]);
assertResponseState(responses, false, false);
assertContacts(Sets.newHashSet(ids[1], ids[3]), responses);
assertResponseState(responses, false, false);
- responses.recordReadFailure(ids[1]);
+ responses.recordFailure(ids[1]);
assertContacts(Sets.newHashSet(ids[0]), responses);
responses.recordReadSuccess(ids[3]);
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index 9e039be0..ebad71ac 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -46,7 +46,7 @@ import accord.impl.LocalListenersTest.TestSafeCommand;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Node;
-import accord.local.NodeTimeService;
+import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
@@ -431,7 +431,7 @@ public class RemoteListenersTest
@Override public DataStore dataStore() { return null; }
@Override public Agent agent() { return null; }
@Override public ProgressLog progressLog() { return null; }
- @Override public NodeTimeService time() { return null; }
+ @Override public NodeCommandStoreService node() { return null; }
@Override public CommandStores.RangesForEpoch ranges() { return null; }
@Override public void registerHistoricalTransactions(Deps deps) { }
}
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 1944a67c..d98b4443 100644
--- a/accord-core/src/test/java/accord/impl/basic/Cluster.java
+++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java
@@ -82,6 +82,7 @@ import accord.local.CommandStore;
import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
+import accord.local.TimeService;
import accord.primitives.SaveStatus;
import accord.local.ShardDistributor;
import accord.primitives.Seekables;
@@ -111,7 +112,6 @@ import static accord.impl.basic.Cluster.OverrideLinksKind.NONE;
import static accord.impl.basic.Cluster.OverrideLinksKind.RANDOM_BIDIRECTIONAL;
import static accord.impl.basic.NodeSink.Action.DELIVER;
import static accord.impl.basic.NodeSink.Action.DROP;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
import static accord.utils.AccordGens.keysInsideRanges;
import static accord.utils.AccordGens.rangeInsideRange;
import static accord.utils.Gens.mixedDistribution;
@@ -469,7 +469,7 @@ public class Cluster implements Scheduler
journalMap.put(id, journal);
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, nowSupplier, elapsedWrapperFromNonMonotonicSource(TimeUnit.MILLISECONDS, nowSupplier),
+ Node node = new Node(id, messageSink, configService, TimeService.ofNonMonotonic(nowSupplier, MILLISECONDS),
() -> new ListStore(sinks, random, id), new ShardDistributor.EvenSplit<>(8, ignore -> new PrefixedIntHashKey.Splitter()),
nodeExecutor.agent(),
randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new,
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 509bb957..93743b21 100644
--- a/accord-core/src/test/java/accord/impl/mock/MockCluster.java
+++ b/accord-core/src/test/java/accord/impl/mock/MockCluster.java
@@ -27,7 +27,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
-import java.util.function.LongSupplier;
+import java.util.function.ToLongFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +49,7 @@ import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.ShardDistributor;
+import accord.local.TimeService;
import accord.messages.Callback;
import accord.messages.Reply;
import accord.messages.Request;
@@ -65,7 +66,7 @@ import accord.utils.ThreadPoolScheduler;
import static accord.Utils.id;
import static accord.Utils.idList;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
+import static accord.local.TimeService.elapsedWrapperFromNonMonotonicSource;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.Write;
import static accord.utils.async.AsyncChains.awaitUninterruptibly;
@@ -76,7 +77,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
private final RandomSource random;
private final Config config;
- private final LongSupplier nowSupplier;
+ private final TimeService time;
private final Map<Id, Node> nodes = new ConcurrentHashMap<>();
private final BiFunction<Id, Network, MessageSink> messageSinkFactory;
private int nextNodeId = 1;
@@ -90,7 +91,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
{
this.config = new Config(builder);
this.random = new DefaultRandom(config.seed);
- this.nowSupplier = builder.nowSupplier;
+ this.time = builder.time;
this.messageSinkFactory = builder.messageSinkFactory;
this.onFetchTopology = builder.onFetchTopology;
@@ -128,8 +129,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
Node node = new Node(id,
messageSink,
configurationService,
- nowSupplier,
- elapsedWrapperFromNonMonotonicSource(TimeUnit.MILLISECONDS, nowSupplier),
+ time,
() -> store,
new ShardDistributor.EvenSplit(8, ignore -> new IntKey.Splitter()),
new TestAgent(),
@@ -321,7 +321,7 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
private int replication = 3;
private int maxKey = 10000;
private Topology topology = null;
- private LongSupplier nowSupplier = System::currentTimeMillis;
+ private TimeService time = TimeService.ofNonMonotonic(System::currentTimeMillis, TimeUnit.MILLISECONDS);
private BiFunction<Id, Network, MessageSink> messageSinkFactory = SimpleMessageSink::new;
private EpochFunction<MockConfigurationService> onFetchTopology = EpochFunction.noop();
@@ -349,9 +349,9 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
return this;
}
- public Builder nowSupplier(LongSupplier supplier)
+ public Builder time(TimeService time)
{
- nowSupplier = supplier;
+ this.time = time;
return this;
}
@@ -387,9 +387,10 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
return new Builder();
}
- public static class Clock implements LongSupplier
+ public static class Clock implements TimeService
{
private final AtomicLong now;
+ private final ToLongFunction<TimeUnit> elapsed = elapsedWrapperFromNonMonotonicSource(TimeUnit.MILLISECONDS, this::now);
public Clock(long now)
{
@@ -412,11 +413,12 @@ public class MockCluster implements Network, AutoCloseable, Iterable<Node>
}
@Override
- public long getAsLong()
+ public long elapsed(TimeUnit unit)
{
- return now();
+ return elapsed.applyAsLong(unit);
}
+
public TxnId idForNode(long epoch, Id id)
{
return new TxnId(epoch, now.get(), Write, Key, id);
diff --git a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
index 6dd8811d..37c2e243 100644
--- a/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
+++ b/accord-core/src/test/java/accord/local/ImmutableCommandTest.java
@@ -20,7 +20,6 @@ package accord.local;
import java.util.List;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -64,7 +63,6 @@ import accord.utils.DefaultRandom;
import static accord.Utils.id;
import static accord.Utils.writeTxn;
import static accord.impl.InMemoryCommandStore.inMemory;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.Write;
import static accord.utils.async.AsyncChains.awaitUninterruptibly;
@@ -104,7 +102,7 @@ public class ImmutableCommandTest
MockCluster.Clock clock = new MockCluster.Clock(100);
LocalConfig localConfig = LocalConfig.DEFAULT;
Node node = new Node(id, null, new MockConfigurationService(null, (epoch, service) -> { }, storeSupport.local.get()),
- clock, elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, clock),
+ clock,
() -> storeSupport.data, new ShardDistributor.EvenSplit(8, ignore -> new IntKey.Splitter()), new TestAgent(), new DefaultRandom(), Scheduler.NEVER_RUN_SCHEDULED,
SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new, ignore -> ignore2 -> new NoOpProgressLog(), DefaultLocalListeners.Factory::new,
InMemoryCommandStores.Synchronized::new,
@@ -147,7 +145,7 @@ public class ImmutableCommandTest
Node node = createNode(ID1, support);
CommandStore commands = node.unsafeByIndex(0);
TxnId txnId = node.nextTxnId(Write, Key);
- ((MockCluster.Clock)node.unsafeGetNowSupplier()).increment(10);
+ ((MockCluster.Clock)node.time()).increment(10);
Keys keys = Keys.of(KEY);
Txn txn = writeTxn(keys);
diff --git a/accord-core/src/test/java/accord/local/NodeTest.java b/accord-core/src/test/java/accord/local/NodeTest.java
index 5019bb8a..46ac7586 100644
--- a/accord-core/src/test/java/accord/local/NodeTest.java
+++ b/accord-core/src/test/java/accord/local/NodeTest.java
@@ -37,7 +37,7 @@ public class NodeTest
void uniqueNowTest()
{
MockCluster.Clock clock = new MockCluster.Clock(100);
- try (MockCluster cluster = MockCluster.builder().nowSupplier(clock).build())
+ try (MockCluster cluster = MockCluster.builder().time(clock).build())
{
Node node = cluster.get(1);
@@ -60,7 +60,7 @@ public class NodeTest
void uniqueNowEpochUpdate()
{
MockCluster.Clock clock = new MockCluster.Clock(100);
- try (MockCluster cluster = MockCluster.builder().nowSupplier(clock).build())
+ try (MockCluster cluster = MockCluster.builder().time(clock).build())
{
Node node = cluster.get(1);
MockConfigurationService configService = (MockConfigurationService) node.configService();
@@ -79,7 +79,7 @@ public class NodeTest
void uniqueNowAtLeastTest()
{
MockCluster.Clock clock = new MockCluster.Clock(100);
- try (MockCluster cluster = MockCluster.builder().nowSupplier(clock).build())
+ try (MockCluster cluster = MockCluster.builder().time(clock).build())
{
Node node = cluster.get(1);
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 182fd98d..d8bb7bfa 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -56,7 +56,7 @@ import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.CommonAttributes;
import accord.local.Node;
-import accord.local.NodeTimeService;
+import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
@@ -824,7 +824,7 @@ public class CommandsForKeyTest
}
@Override
- public NodeTimeService time()
+ public NodeCommandStoreService node()
{
throw new UnsupportedOperationException();
}
diff --git a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
index 508a4a6d..b5def35e 100644
--- a/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
+++ b/accord-core/src/test/java/accord/utils/LogGroupTimersTest.java
@@ -27,8 +27,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import accord.api.Scheduler;
-
import static java.util.concurrent.TimeUnit.DAYS;
import static java.util.concurrent.TimeUnit.HOURS;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
@@ -40,14 +38,14 @@ public class LogGroupTimersTest
public void test()
{
Random random = new Random();
- for (int i = 0 ; i < 100 ; ++i)
+ for (int i = 0 ; i < 10000 ; ++i)
testOne(random.nextLong(), 1000, 100);
}
@Test
public void testOne()
{
- testOne(5843168000636021001L, 1000, 100);
+ testOne(-9121426382346619484L, 1000, 100);
}
static class Timer extends LogGroupTimers.Timer
@@ -106,7 +104,6 @@ public class LogGroupTimersTest
LogGroupTimers<Timer> test = new LogGroupTimers<>(MICROSECONDS);
TreeSet<Timer> canonical = new TreeSet<>(Timer::compareDeadline);
List<Timer> readTest = new ArrayList<>(), readCanonical = new ArrayList<>();
- TestScheduler testScheduler = new TestScheduler(rnd, test, globalMaxDeadline, canonical);
TreeSet<Timer> randomOrder = new TreeSet<>(Timer::compareRandom);
int timerId = 0;
for (int round = 0 ; round < rounds ; ++round)
@@ -118,9 +115,6 @@ public class LogGroupTimersTest
long minDeadline = test.curEpoch + rnd.nextLong(globalMaxDeadline - deadlineClustering);
long maxDeadline = minDeadline + deadlineClustering;
- if (rnd.nextBoolean())
- testScheduler.updateNow(test.curEpoch);
-
while (adds > 0 || ((removes + updates) > 0 && !canonical.isEmpty()))
{
boolean shouldAdd = adds > 0 && ((removes + updates == 0) || canonical.isEmpty() || rnd.nextBoolean());
@@ -129,7 +123,6 @@ public class LogGroupTimersTest
--adds;
Timer add = new Timer(rnd.nextLong(minDeadline, maxDeadline), rnd.nextInt(), ++timerId);
test.add(add.deadline, add);
- testScheduler.add(add.deadline);
canonical.add(add);
randomOrder.add(add);
}
@@ -144,7 +137,6 @@ public class LogGroupTimersTest
int newWeight = rnd.nextInt();
timer.update(newDeadline, newWeight);
test.update(newDeadline, timer);
- testScheduler.add(newDeadline);
randomOrder.add(timer);
canonical.add(timer);
}
@@ -156,8 +148,22 @@ public class LogGroupTimersTest
}
}
+ if (!canonical.isEmpty())
+ {
+ Timer first = canonical.first();
+ Assertions.assertTrue(test.wakeAt() <= first.deadline);
+ Assertions.assertTrue(test.wakeAt() >= test.curEpoch);
+ }
+
if (rnd.decide(readChance))
testAdvance(test, canonical, randomOrder, readTest, readCanonical, maxBatchSize, rnd);
+
+ if (!canonical.isEmpty())
+ {
+ Timer first = canonical.first();
+ Assertions.assertTrue(test.wakeAt() <= first.deadline);
+ Assertions.assertTrue(test.wakeAt() >= test.curEpoch);
+ }
}
while (!canonical.isEmpty())
@@ -201,12 +207,6 @@ public class LogGroupTimersTest
nextCanonical.add(canonical.pollFirst());
long advanceTo = nextCanonical.get(nextCanonical.size() - 1).deadline;
- if (rnd.nextBoolean())
- {
- Timer first = nextCanonical.get(0);
- Assertions.assertTrue(test.nextDeadlineEpoch() <= first.deadline);
- Assertions.assertEquals(first, test.peekIfSoonerThan(first.deadline));
- }
test.advance(advanceTo, nextTest, List::add);
}
@@ -217,87 +217,4 @@ public class LogGroupTimersTest
nextTest.clear();
++i;
}
-
- static class TestScheduler implements Scheduler
- {
- final RandomSource rnd;
- final long schedulerImpreciseLateTolerance;
- final long schedulerPreciseLateTolerance;
- final long schedulerPreciseDelayThreshold;
- final LogGroupTimers<Timer>.Scheduling scheduling;
- final TreeSet<Timer> canonical;
-
- long now;
- long maxNow;
- long at;
-
- TestScheduler(RandomSource rnd, LogGroupTimers<Timer> test, long globalMaxDeadline, TreeSet<Timer> canonical)
- {
- this.rnd = rnd;
- this.canonical = canonical;
- schedulerImpreciseLateTolerance = rnd.nextLong(globalMaxDeadline / 128);
- schedulerPreciseLateTolerance = rnd.nextLong(globalMaxDeadline / 128);
- schedulerPreciseDelayThreshold = rnd.nextLong(globalMaxDeadline / 128);
- scheduling = test.new Scheduling(this, timer -> null, schedulerImpreciseLateTolerance, schedulerPreciseLateTolerance, schedulerPreciseDelayThreshold);
- }
-
- void updateNow(long epoch)
- {
- long prevAt = at;
- now = Math.max(0, epoch + rnd.nextLong(schedulerImpreciseLateTolerance) - schedulerPreciseLateTolerance/2);
- maxNow = Math.max(now, maxNow);
- scheduling.ensureScheduled(now);
- check(prevAt);
- }
-
- @Override
- public Scheduled recurring(Runnable run, long delay, TimeUnit units)
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Scheduled once(Runnable run, long delay, TimeUnit units)
- {
- at = maxNow + delay;
- return new Scheduled()
- {
- @Override
- public void cancel()
- {
- at = 0;
- }
-
- @Override
- public boolean isDone()
- {
- return false;
- }
- };
- }
-
- @Override
- public void now(Runnable run)
- {
- throw new UnsupportedOperationException();
- }
-
- public void add(long deadline)
- {
- long prev = at;
- scheduling.maybeReschedule(now, deadline);
- check(prev);
- }
-
- private void check(long prevAt)
- {
- if (canonical.isEmpty())
- return;
-
- Timer first = canonical.first();
- Assertions.assertTrue(at <= prevAt || prevAt < first.deadline);
- Assertions.assertEquals(at, scheduling.scheduledAt());
- }
- }
-
}
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java b/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
index 902607b3..fa8a2048 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java
@@ -54,6 +54,7 @@ import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.ShardDistributor;
+import accord.local.TimeService;
import accord.messages.Callback;
import accord.messages.Reply;
import accord.messages.Reply.FailureReply;
@@ -65,7 +66,7 @@ import accord.utils.RandomSource;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
+import static accord.local.TimeService.elapsedWrapperFromNonMonotonicSource;
import static java.util.stream.Collectors.toList;
// TODO (low priority, testing): merge with accord.impl.basic.Cluster
@@ -330,7 +331,7 @@ public class Cluster implements Scheduler
LongSupplier nowSupplier = nowSupplierSupplier.get();
LocalConfig localConfig = LocalConfig.DEFAULT;
lookup.put(node, new Node(node, messageSink, new SimpleConfigService(topology),
- nowSupplier, elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, nowSupplier),
+ TimeService.of(nowSupplier, elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, nowSupplier)),
MaelstromStore::new, new ShardDistributor.EvenSplit(8, ignore -> new MaelstromKey.Splitter()),
MaelstromAgent.INSTANCE,
randomSupplier.get(), sinks, SizeOfIntersectionSorter.SUPPLIER, DefaultRemoteListeners::new, DefaultRequestTimeouts::new,
diff --git a/accord-maelstrom/src/main/java/accord/maelstrom/Main.java b/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
index 6418a1b0..9b3cbe39 100644
--- a/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
+++ b/accord-maelstrom/src/main/java/accord/maelstrom/Main.java
@@ -46,6 +46,7 @@ import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.ShardDistributor;
+import accord.local.TimeService;
import accord.maelstrom.Packet.Type;
import accord.messages.Callback;
import accord.messages.Reply;
@@ -56,7 +57,6 @@ import accord.topology.Topology;
import accord.utils.DefaultRandom;
import accord.utils.ThreadPoolScheduler;
-import static accord.local.NodeTimeService.elapsedWrapperFromNonMonotonicSource;
import static accord.utils.async.AsyncChains.awaitUninterruptibly;
public class Main
@@ -180,7 +180,7 @@ public class Main
sink = new StdoutSink(System::currentTimeMillis, scheduler, start, init.self, out, err);
LocalConfig localConfig = LocalConfig.DEFAULT;
on = new Node(init.self, sink, new SimpleConfigService(topology),
- System::currentTimeMillis, elapsedWrapperFromNonMonotonicSource(TimeUnit.MILLISECONDS, System::currentTimeMillis),
+ TimeService.ofNonMonotonic(System::currentTimeMillis, TimeUnit.MILLISECONDS),
MaelstromStore::new, new ShardDistributor.EvenSplit(8, ignore -> new MaelstromKey.Splitter()),
MaelstromAgent.INSTANCE, new DefaultRandom(), scheduler, SizeOfIntersectionSorter.SUPPLIER,
DefaultRemoteListeners::new, DefaultRequestTimeouts::new, DefaultProgressLogs::new, DefaultLocalListeners.Factory::new,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org