Executor QoS

accord-core/src/main/java/accord/api/AsyncExecutor.java

@@ -35,6 +35,11 @@ default Cancellable execute(RunOrFail run)
         return AsyncCallbacks.execute(this, run);
     }
 
+    default Cancellable executeContinuation(RunOrFail run)
+    {
+        return AsyncCallbacks.execute(this, run);
+    }
+
     default boolean tryExecuteImmediately(Runnable run) { return false; }
 
     // Depending on this implementation this method may queue-jump, i.e. task submission order is not guaranteed.
@@ -50,6 +55,11 @@ default boolean executeMaybeImmediately(Runnable run)
     }
 
     AsyncChain<Void> chain(Runnable run);
+    /**
+     * As {@link #chain(Runnable)}, but if the submitting task fails while running this should be cancelled,
+     * failing the chain. See {@link #executeContinuation}.
+     */
+    AsyncChain<Void> continuationChain(Runnable run);
     <V> AsyncChain<V> chain(Callable<V> call);
     <V> AsyncChain<V> flatChain(Callable<? extends AsyncChain<V>> call);
 }

accord-core/src/main/java/accord/api/AsyncExecutorFactory.java

@@ -18,10 +18,8 @@
 
 package accord.api;
 
-import accord.local.SequentialAsyncExecutor;
-
 public interface AsyncExecutorFactory
 {
     AsyncExecutor someExecutor();
-    SequentialAsyncExecutor someSequentialExecutor();
+    ExclusiveAsyncExecutor someExclusiveExecutor();
 }

accord-core/src/main/java/accord/impl/SafeState.java → accord-core/src/main/java/accord/api/ExclusiveAsyncExecutor.java

@@ -16,17 +16,11 @@
  * limitations under the License.
  */
 
-package accord.impl;
+package accord.api;
 
 /**
- * State scoped to a single request that references global state
+ * A single-threaded AsyncExecutor
  */
-public interface SafeState<T>
+public interface ExclusiveAsyncExecutor extends AsyncExecutor
 {
-    T current();
-
-    default boolean isUnset()
-    {
-        return current() == null;
-    }
 }

accord-core/src/main/java/accord/api/ProtocolModifiers.java

@@ -26,7 +26,7 @@
 
 import accord.primitives.Ballot;
 import accord.primitives.Deps;
-import accord.primitives.Routable;
+import accord.primitives.Routable.Domain;
 import accord.primitives.SaveStatus;
 import accord.primitives.Timestamp;
 import accord.primitives.Txn;
@@ -211,7 +211,7 @@ public static synchronized void setTransitiveDependenciesAreVisible(Txn.Kind ...
 
         public static void validate()
         {
-            Invariants.require(dataStoreDetectsFutureReads || fastWriteExecution != MAY_BYPASS_SAFESTORE && fastReadExecution != MAY_BYPASS_SAFESTORE, "MAY_BYPASS_SAFESTORE is only permitted when dataStoreDetectsFutureReads");
+            Invariants.require(dataStoreDetectsFutureReads || (fastWriteExecution != MAY_BYPASS_SAFESTORE && fastReadExecution != MAY_BYPASS_SAFESTORE), "MAY_BYPASS_SAFESTORE is only permitted when dataStoreDetectsFutureReads");
             Invariants.require(permitCoordinatorLocalExecution || (!permittedFastPaths.contains(PrivilegedCoordinatorWithDeps) && !permittedFastPaths.contains(PrivilegedCoordinatorWithoutDeps)), "Privileged coordinator optimisations require coordinator local execution");
         }
     }
@@ -311,8 +311,8 @@ public static InformOfDurability informOfDurability(TxnId txnId, @Nullable Deps
     }
 
     private static FastExecution fastReadExecution = Configure.fastReadExecution;
-    public static boolean fastReadsMayBypassSafeStore(TxnId txnId) { return fastReadExecution == MAY_BYPASS_SAFESTORE && (dataStoreDetectsFutureReads() || txnId.is(EphemeralRead)) && txnId.is(Routable.Domain.Key); }
-    public static boolean fastReadsMayBypassCommandsForKey(TxnId txnId) { return fastReadExecution != FastExecution.DISABLED && !txnId.is(Txn.Kind.Write) && txnId.is(Routable.Domain.Key); }
+    public static boolean fastReadsMayBypassSafeStore(TxnId txnId) { return fastReadExecution == MAY_BYPASS_SAFESTORE && (dataStoreDetectsFutureReads() || txnId.is(EphemeralRead)) && txnId.is(Domain.Key); }
+    public static boolean fastReadsMayBypassCommandsForKey(TxnId txnId) { return fastReadExecution != FastExecution.DISABLED && !txnId.is(Txn.Kind.Write) && txnId.is(Domain.Key); }
 
     private static final boolean fastReadExecutionMayResendTxn = Configure.fastReadExecMayResendTxn;
     public static boolean fastReadExecutionMayResendTxn() { return fastReadExecutionMayResendTxn; }

accord-core/src/main/java/accord/coordinate/AbstractCoordinatePreAccept.java

@@ -22,7 +22,7 @@
 import javax.annotation.Nonnull;
 
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.Callback;
 import accord.primitives.FullRoute;
 import accord.primitives.TxnId;
@@ -38,7 +38,7 @@ abstract class AbstractCoordinatePreAccept<Result, Reply extends accord.messages
 {
     final Topologies topologies;
 
-    AbstractCoordinatePreAccept(Node node, SequentialAsyncExecutor executor, Topologies topologies, FullRoute<?> route, @Nonnull TxnId txnId, BiConsumer<? super Result, Throwable> callback)
+    AbstractCoordinatePreAccept(Node node, ExclusiveAsyncExecutor executor, Topologies topologies, FullRoute<?> route, @Nonnull TxnId txnId, BiConsumer<? super Result, Throwable> callback)
     {
         super(node, executor, txnId, route, topologies.nodes(), callback);
         this.topologies = topologies;

accord-core/src/main/java/accord/coordinate/AbstractCoordination.java

@@ -32,7 +32,7 @@
 import accord.coordinate.tracking.RequestStatus;
 import accord.local.MapReduceConsumeCommandStores;
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.Request;
 import accord.messages.Callback;
 import accord.primitives.Participants;
@@ -65,9 +65,9 @@ public abstract class AbstractCoordination<P extends Participants<?>, Result, Re
     private BiConsumer<? super Result, Throwable> callback;
     private Object[] replyState;
     private int replyCount;
-    private boolean unsafeToReplyImmediately;
+    private boolean unsafeToReply;
 
-    protected AbstractCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, P scope, SortedArrayList<Node.Id> nodes, BiConsumer<? super Result, Throwable> callback)
+    protected AbstractCoordination(Node node, ExclusiveAsyncExecutor executor, TxnId txnId, P scope, SortedArrayList<Node.Id> nodes, BiConsumer<? super Result, Throwable> callback)
     {
         super(node, executor, txnId, scope);
         this.nodes = nodes;
@@ -198,38 +198,45 @@ void contact(Function<Node.Id, Request> request)
     void contact(Function<Node.Id, Request> request, @Nullable Predicate<Node.Id> include)
     {
         executor.executeMaybeImmediately(() -> {
-            unsafeToReplyImmediately = true;
-            AbstractTracker<?> tracker = tracker();
-            Topologies topologies = tracker.topologies();
-            if (tracing != null)
-                tracing.trace(null, "contacting %s", nodes);
-
-            for (int i = 0; i < nodes.size() ; ++i)
+            unsafeToReply = true;
+            try
             {
-                Node.Id to = nodes.get(i);
-                if (include == null || include.test(to))
+
+                AbstractTracker<?> tracker = tracker();
+                Topologies topologies = tracker.topologies();
+                if (tracing != null)
+                    tracing.trace(null, "contacting %s", nodes);
+
+                for (int i = 0; i < nodes.size() ; ++i)
                 {
-                    if (topologies.isFaulty(to))
+                    Node.Id to = nodes.get(i);
+                    if (include == null || include.test(to))
                     {
-                        if (tracing != null)
-                            tracing.trace(null, "%s is considered faulty; recording failure instead", to);
-                        if (RequestStatus.Failed == tracker.prerecordFailure(to))
+                        if (topologies.isFaulty(to))
                         {
-                            finishOnExaustion();
-                            return;
+                            if (tracing != null)
+                                tracing.trace(null, "%s is considered faulty; recording failure instead", to);
+                            if (RequestStatus.Failed == tracker.prerecordFailure(to))
+                            {
+                                finishOnExaustion();
+                                return;
+                            }
+                        }
+                        else
+                        {
+                            Invariants.require(replyState[i] == null);
+                            expectingReply.set(i);
+                            // TODO (expected): do not cancel PreAccept, Accept, Commit, Stable or Apply to self on done
+                            replyState[i] = node.send(to, request.apply(to), executor, this, tracing);
+                            Invariants.require(expectingReply.get(i) || replyState[i] == null);
                         }
-                    }
-                    else
-                    {
-                        Invariants.require(replyState[i] == null);
-                        expectingReply.set(i);
-                        // TODO (expected): do not cancel PreAccept, Accept, Commit, Stable or Apply to self on done
-                        replyState[i] = node.send(to, request.apply(to), executor, this, tracing);
-                        Invariants.require(expectingReply.get(i) || replyState[i] == null);
                     }
                 }
             }
-            unsafeToReplyImmediately = false;
+            finally
+            {
+                unsafeToReply = false;
+            }
         });
     }
 
@@ -245,19 +252,19 @@ void recontact(Node.Id to, Request send)
     @Override
     public final void onSuccess(Node.Id from, Reply reply)
     {
-        CallbackExclusive.onSuccess(executor, unsafeToReplyImmediately, this, from, reply);
+        CallbackExclusive.onSuccess(executor, unsafeToReply, this, from, reply);
     }
 
     @Override
     public final void onSlow(Node.Id from)
     {
-        CallbackExclusive.onSlow(executor, unsafeToReplyImmediately, this, from);
+        CallbackExclusive.onSlow(executor, unsafeToReply, this, from);
     }
 
     @Override
     public final void onFailure(Node.Id from, Throwable failure)
     {
-        CallbackExclusive.onFailure(executor, unsafeToReplyImmediately, this, from, failure);
+        CallbackExclusive.onFailure(executor, unsafeToReply, this, from, failure);
     }
 
     @Override
@@ -308,6 +315,7 @@ public void onFailureExclusive(Node.Id from, @Nullable Throwable failure)
 
     private int onReply(Node.Id from, Object reply, boolean isFinal)
     {
+        Invariants.require(!unsafeToReply);
         int fromIndex = nodes.find(from);
         if (isDoneWithReplies())
         {

accord-core/src/main/java/accord/coordinate/AbstractSimpleCoordination.java

@@ -22,7 +22,7 @@
 
 import accord.api.Tracing;
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.primitives.Participants;
 import accord.primitives.TxnId;
 import accord.utils.Invariants;
@@ -33,14 +33,14 @@ public abstract class AbstractSimpleCoordination<P extends Participants<?>> impl
 {
     final long coordinationId;
     protected final Node node;
-    protected final SequentialAsyncExecutor executor;
+    protected final ExclusiveAsyncExecutor executor;
     protected final TxnId txnId;
     protected final P scope;
     protected final @Nullable Tracing tracing;
     private Throwable failure;
     private boolean isDoneWithReplies, isFinishing, isDone;
 
-    protected AbstractSimpleCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, P scope)
+    protected AbstractSimpleCoordination(Node node, ExclusiveAsyncExecutor executor, TxnId txnId, P scope)
     {
         this.coordinationId = node.nextCoordinationId();
         this.node = node;
@@ -65,7 +65,7 @@ public final TxnId txnId()
     public final P scope() { return scope; }
 
     @Override
-    public final SequentialAsyncExecutor executor()
+    public final ExclusiveAsyncExecutor executor()
     {
         return executor;
     }

accord-core/src/main/java/accord/coordinate/AsynchronousAwait.java

@@ -26,7 +26,7 @@
 import accord.coordinate.tracking.RequestStatus;
 import accord.local.Commands;
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.Await;
 import accord.messages.Await.AwaitOk;
 import accord.messages.Callback;
@@ -74,7 +74,7 @@ public SynchronousResult(Unseekables<?> ready, @Nullable Unseekables<?> notReady
     final int asynchronousCallbackId;
     final boolean notifyProgressLog;
 
-    public AsynchronousAwait(Node node, SequentialAsyncExecutor executor, Participants<?> contact, TxnId txnId, AwaitTracker tracker, Await.Until until, boolean notifyProgressLog, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
+    public AsynchronousAwait(Node node, ExclusiveAsyncExecutor executor, Participants<?> contact, TxnId txnId, AwaitTracker tracker, Await.Until until, boolean notifyProgressLog, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
     {
         super(node, executor, txnId, contact, tracker.nodes(), synchronousCallback);
         this.tracker = tracker;
@@ -85,14 +85,14 @@ public AsynchronousAwait(Node node, SequentialAsyncExecutor executor, Participan
 
     public static AsynchronousAwait awaitAny(Node node, Topologies topologies, TxnId txnId, Route<?> contact, Await.Until until, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
     {
-        return awaitAny(node, node.someSequentialExecutor(), topologies, txnId, contact, until, true, asynchronousCallbackId, synchronousCallback);
+        return awaitAny(node, node.someExclusiveExecutor(), topologies, txnId, contact, until, true, asynchronousCallbackId, synchronousCallback);
     }
 
     /**
      * we require a Route to contact so we can be sure a home shard recipient invokes {@link Commands#supplementParticipants},
      * notifying the progress log of a Route to determine it is the home shard.
      */
-    public static AsynchronousAwait awaitAny(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Route<?> contact, Await.Until until, boolean notifyProgressLog, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
+    public static AsynchronousAwait awaitAny(Node node, ExclusiveAsyncExecutor executor, Topologies topologies, TxnId txnId, Route<?> contact, Await.Until until, boolean notifyProgressLog, int asynchronousCallbackId, BiConsumer<SynchronousResult, Throwable> synchronousCallback)
     {
         Invariants.requireArgument(topologies.size() == 1);
         AwaitTracker tracker = new AwaitTracker(topologies);

accord-core/src/main/java/accord/coordinate/CheckShards.java

@@ -23,7 +23,7 @@
 
 import accord.local.Node;
 import accord.local.Node.Id;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.CheckStatus;
 import accord.messages.CheckStatus.CheckStatusOk;
 import accord.messages.CheckStatus.CheckStatusReply;
@@ -57,12 +57,12 @@ public abstract class CheckShards<R, U extends Participants<?>> extends ReadCoor
     protected boolean truncated;
 
     // srcEpoch is either txnId.epoch() or executeAt.epoch()
-    protected CheckShards(Node node, SequentialAsyncExecutor executor, TxnId txnId, U query, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback) throws TopologyException
+    protected CheckShards(Node node, ExclusiveAsyncExecutor executor, TxnId txnId, U query, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback) throws TopologyException
     {
         this(node, executor, txnId, query, txnId.epoch(), includeInfo, bumpBallot, previouslyKnownToBeInvalidIf, callback);
     }
 
-    protected CheckShards(Node node, SequentialAsyncExecutor executor, TxnId txnId, U query, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback) throws TopologyException
+    protected CheckShards(Node node, ExclusiveAsyncExecutor executor, TxnId txnId, U query, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf, BiConsumer<? super R, Throwable> callback) throws TopologyException
     {
         super(node, executor, topologyFor(node, txnId, query, srcEpoch), txnId, query, callback);
         this.sourceEpoch = srcEpoch;

accord-core/src/main/java/accord/coordinate/CollectLatestDeps.java

@@ -55,7 +55,7 @@ public class CollectLatestDeps extends AbstractCoordination<Route<?>, List<Lates
 
     CollectLatestDeps(Node node, Topologies topologies, TxnId txnId, Route<?> route, @Nullable Ballot ballot, Timestamp executeAt, BiConsumer<List<LatestDeps>, Throwable> callback)
     {
-        super(node, node.someSequentialExecutor(), txnId, route, topologies.nodes(), callback);
+        super(node, node.someExclusiveExecutor(), txnId, route, topologies.nodes(), callback);
         this.executeAt = executeAt;
         this.ballot = ballot;
         this.tracker = new QuorumTracker(topologies);

accord-core/src/main/java/accord/coordinate/CoordinateEphemeralRead.java

@@ -29,7 +29,7 @@
 import accord.coordinate.tracking.AbstractTracker;
 import accord.coordinate.tracking.QuorumTracker;
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.GetEphemeralReadDeps;
 import accord.messages.GetEphemeralReadDeps.GetEphemeralReadDepsOk;
 import accord.primitives.Deps;
@@ -86,7 +86,7 @@ public static void coordinate(Node node, TxnId txnId, Txn txn, BiConsumer<? supe
         {
             FullRoute<?> route = node.computeRoute(txnId, txn.keys());
             Topologies topologies = node.topology().active().withUnsyncedEpochs(route, txnId, txnId);
-            coordinate = new CoordinateEphemeralRead(node, node.someSequentialExecutor(), topologies, route, txnId, txn, callback);
+            coordinate = new CoordinateEphemeralRead(node, node.someExclusiveExecutor(), topologies, route, txnId, txn, callback);
         }
         catch (Throwable t)
         {
@@ -102,7 +102,7 @@ public static void coordinate(Node node, TxnId txnId, Txn txn, BiConsumer<? supe
     private long executeAtEpoch;
     private long retryInEpoch;
 
-    CoordinateEphemeralRead(Node node, SequentialAsyncExecutor executor, Topologies topologies, FullRoute<?> route, TxnId txnId, Txn txn, BiConsumer<? super Result, Throwable> callback)
+    CoordinateEphemeralRead(Node node, ExclusiveAsyncExecutor executor, Topologies topologies, FullRoute<?> route, TxnId txnId, Txn txn, BiConsumer<? super Result, Throwable> callback)
     {
         super(node, executor, topologies, route, txnId, callback);
         this.txn = txn;

accord-core/src/main/java/accord/coordinate/CoordinateMaxConflict.java

@@ -27,7 +27,7 @@
 import accord.coordinate.tracking.AbstractTracker;
 import accord.coordinate.tracking.QuorumTracker;
 import accord.local.Node;
-import accord.local.SequentialAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.messages.GetMaxConflict;
 import accord.messages.GetMaxConflict.GetMaxConflictOk;
 import accord.primitives.FullRoute;
@@ -56,7 +56,7 @@ public class CoordinateMaxConflict extends AbstractCoordinatePreAccept<Timestamp
     Timestamp maxConflict;
     long executionEpoch;
 
-    private CoordinateMaxConflict(Node node, SequentialAsyncExecutor executor, Topologies topologies, FullRoute<?> route, long executionEpoch, BiConsumer<? super Timestamp, Throwable> callback)
+    private CoordinateMaxConflict(Node node, ExclusiveAsyncExecutor executor, Topologies topologies, FullRoute<?> route, long executionEpoch, BiConsumer<? super Timestamp, Throwable> callback)
     {
         super(node, executor, topologies, route, TxnId.NONE, callback);
         this.maxConflict = Timestamp.NONE;
@@ -86,7 +86,7 @@ public static void maxConflict(Node node, Routables<?> keysOrRanges, BiConsumer<
             long epoch = active.maxEpoch(Long.MIN_VALUE, ActiveEpoch::all, keysOrRanges);
             FullRoute<?> route = node.computeRoute(epoch, keysOrRanges, active);
             Topologies topologies = active.withUnsyncedEpochs(route, epoch, epoch, ALL);
-            coordinate = new CoordinateMaxConflict(node, node.someSequentialExecutor(), topologies, route, epoch, callback);
+            coordinate = new CoordinateMaxConflict(node, node.someExclusiveExecutor(), topologies, route, epoch, callback);
         }
         catch (Throwable t)
         {