Facilitate queueing, execution and performance improvements

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20069

Author: belliottsmith

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

@@ -32,6 +32,8 @@
 import accord.primitives.TxnId;
 import accord.topology.Topologies;
 
+import static accord.utils.Invariants.illegalState;
+
 /**
  * Facility for augmenting node behaviour at specific points
  *
@@ -122,8 +124,6 @@ default EventsListener metricsEventsListener()
      */
     default Topologies selectPreferred(Node.Id from, Topologies to) { return to; }
 
-    long replyTimeout(ReplyContext replyContext, TimeUnit units);
-
     /**
      *  This method permits implementations to configure the time at which a local home shard will attempt
      *  to coordinate a transaction to completion.
@@ -148,4 +148,8 @@ default EventsListener metricsEventsListener()
      * and re-query the local state.
      */
     long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units);
+
+    long expiresAt(ReplyContext replyContext, TimeUnit unit);
+
+    default void onViolation(String message) { throw illegalState(message); }
 }

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

@@ -74,9 +74,9 @@ class EpochReady
         public final AsyncResult<Void> metadata;
 
         /**
-         * The node has retrieved enough remote information to become a coordinator for the new epoch.
+         * The node has retrieved enough remote information to become a fast path coordinator for the new epoch.
          */
-        public final AsyncResult<Void> coordination;
+        public final AsyncResult<Void> fastPath;
 
         /**
          * The node has successfully replicated the underlying DataStore information for the new epoch, but may need
@@ -90,11 +90,11 @@ class EpochReady
          */
         public final AsyncResult<Void> reads;
 
-        public EpochReady(long epoch, AsyncResult<Void> metadata, AsyncResult<Void> coordination, AsyncResult<Void> data, AsyncResult<Void> reads)
+        public EpochReady(long epoch, AsyncResult<Void> metadata, AsyncResult<Void> fastPath, AsyncResult<Void> data, AsyncResult<Void> reads)
         {
             this.epoch = epoch;
             this.metadata = metadata;
-            this.coordination = coordination;
+            this.fastPath = fastPath;
             this.data = data;
             this.reads = reads;
         }

…ain/java/accord/api/RequestTimeouts.java …e/src/main/java/accord/api/Timeouts.javaaccord-core/src/main/java/accord/api/RequestTimeouts.java renamed to accord-core/src/main/java/accord/api/Timeouts.java accord-core/src/main/java/accord/api/RequestTimeouts.java renamed to accord-core/src/main/java/accord/api/Timeouts.java

@@ -20,19 +20,22 @@
 
 import java.util.concurrent.TimeUnit;
 
-public interface RequestTimeouts
-{
-    interface RegisteredTimeout
-    {
-        void cancel();
-    }
+import accord.utils.async.Cancellable;
 
+public interface Timeouts
+{
     interface Timeout
     {
         void timeout();
         int stripe();
     }
 
-    RegisteredTimeout register(Timeout timeout, long delay, TimeUnit units);
+    interface RegisteredTimeout extends Cancellable
+    {
+        void cancel();
+    }
+
+    RegisteredTimeout registerWithDelay(Timeout timeout, long delay, TimeUnit units);
+    RegisteredTimeout registerAt(Timeout timeout, long deadline, TimeUnit units);
     void maybeNotify();
 }

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

@@ -208,7 +208,7 @@ final void onPreAcceptedOrNewEpoch()
         {
             long latestEpoch = executeAtEpoch();
             if (latestEpoch > topologies.currentEpoch()) node.withEpoch(latestEpoch, node.agent(), () -> onPreAcceptedInNewEpoch(topologies, latestEpoch));
-            else onPreAcceptedInNewEpoch(topologies, latestEpoch);
+            else onPreAccepted(topologies);
             return;
         }
 

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

@@ -233,7 +233,7 @@ private ExistingTransactionCheck checkForExistingTransaction()
         ExistingTransactionCheck check = new ExistingTransactionCheck();
         RoutingKey k = route.get(0).asRoutingKey();
         node.commandStores().mapReduceConsume(
-                contextFor(k, KeyHistory.COMMANDS),
+                contextFor(k, KeyHistory.SYNC),
                 k.toUnseekable(),
                 minEpoch,
                 Long.MAX_VALUE,

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

@@ -52,7 +52,6 @@
  */
 abstract class CoordinatePreAccept<T> extends AbstractCoordinatePreAccept<T, PreAcceptReply>
 {
-    // TODO (required): do not use FastPathTracker for coordinating exclusive sync points - these always take the slow path
     final PreAcceptTracker<?> tracker;
     // TODO (expected): this can be cleared after preaccept
     // TODO (expected): back by SortedListMap; must handle additional preaccepts (but this is no longer ordinarily enabled)

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

@@ -98,7 +98,7 @@ protected Action process(Id from, ReadReply reply)
             if (ok.futureEpoch > allTopologies.currentEpoch())
             {
                 // TODO (expected): only submit new requests for the keys that execute in a later epoch
-                new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId, txn, ok.futureEpoch, deps, callback).start();
+                new ExecuteEphemeralRead(node, node.topology().preciseEpochs(route, ok.futureEpoch, ok.futureEpoch), route, txnId.withEpoch(ok.futureEpoch), txn, ok.futureEpoch, deps, callback).start();
                 return Aborted;
             }
 

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

@@ -54,6 +54,8 @@ public class ExecuteTxn extends ReadCoordinator<ReadReply>
 {
     @SuppressWarnings("unused")
     private static final Logger logger = LoggerFactory.getLogger(ExecuteTxn.class);
+    private static boolean SEND_MINIMUM_STABLE_MESSAGES = true;
+    public static void setSendMinimumStableMessages(boolean sendMin) {SEND_MINIMUM_STABLE_MESSAGES = sendMin; }
 
     final ExecutePath path;
     final Txn txn;
@@ -85,22 +87,25 @@ protected void start(Iterable<Id> to)
     {
         IntHashSet readSet = new IntHashSet();
         to.forEach(i -> readSet.add(i.id));
-        Commit.Kind kind;
+        Commit.stableAndRead(node, allTopologies, commitKind(), txnId, txn, route, readScope, executeAt, stableDeps, readSet, this, SEND_MINIMUM_STABLE_MESSAGES);
+    }
+
+    private Commit.Kind commitKind()
+    {
         switch (path)
         {
             default: throw new AssertionError("Unhandled path: " + path);
-            case FAST:    kind = StableFastPath; break;
-            case SLOW:    kind = StableSlowPath; break;
-            case RECOVER: kind = StableWithTxnAndDeps; break;
+            case FAST:    return StableFastPath;
+            case SLOW:    return StableSlowPath;
+            case RECOVER: return StableWithTxnAndDeps;
         }
-        // we want to send to all topologies, but we only want to track responses from the readScope
-        Commit.stableAndRead(node, allTopologies, kind, txnId, txn, route, readScope, executeAt, stableDeps, readSet, this);
     }
 
     @Override
     public void contact(Id to)
     {
-        node.send(to, new ReadTxnData(to, topologies(), txnId, readScope, executeAt.epoch()), this);
+        if (SEND_MINIMUM_STABLE_MESSAGES) Commit.stableAndRead(to, node, allTopologies, commitKind(), txnId, txn, route, readScope, executeAt, stableDeps, this, SEND_MINIMUM_STABLE_MESSAGES);
+        else node.send(to, new ReadTxnData(to, topologies(), txnId, readScope, executeAt.epoch()), this);
     }
 
     @Override

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

@@ -47,10 +47,11 @@ public class MaybeRecover extends CheckShards<Route<?>>
         this.callback = callback;
     }
 
-    public static void maybeRecover(Node node, TxnId txnId, Route<?> someRoute, ProgressToken prevProgress, BiConsumer<Outcome, Throwable> callback)
+    public static Object maybeRecover(Node node, TxnId txnId, Route<?> someRoute, ProgressToken prevProgress, BiConsumer<Outcome, Throwable> callback)
     {
         MaybeRecover maybeRecover = new MaybeRecover(node, txnId, someRoute, prevProgress, callback);
         maybeRecover.start();
+        return maybeRecover;
     }
 
     @Override

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

@@ -140,7 +140,7 @@ else if (failure instanceof Redundant)
                 node.agent().metricsEventsListener().onPreempted(txnId);
             else if (failure instanceof Timeout)
                 node.agent().metricsEventsListener().onTimeout(txnId);
-            else if (failure instanceof Invalidated)
+            else if (failure instanceof Invalidated) // TODO (expected): should we tick this counter? we haven't invalidated anything
                 node.agent().metricsEventsListener().onInvalidated(txnId);
         }