Split AsyncChain and AsyncResult; normalise AsyncResult with C* Future

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

Author: belliottsmith

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/belliottsmith/cassandra-accord.git
-	branch = 20905-ssverifier
+	url = https://github.com/apache/cassandra-accord.git
+	branch = trunk

modules/accord

@@ -229,7 +229,7 @@ public void finished()
             long deadlineNanos = startedAtNanos + timeoutNanos;
             // TODO (expected): use the source bounds for the streams to avoid waiting unnecessarily long
             AccordService.getBlocking(accordService.maxConflict(accordRanges)
-                                                   .flatMap(min -> accordService.sync("[Stream #" + session.planId() + ']', min, accordRanges, null, Self, NoRemote, timeoutNanos, NANOSECONDS))
+                                                   .flatMap(min -> accordService.sync("[Stream #" + session.planId() + ']', min, accordRanges, null, Self, NoRemote, timeoutNanos, NANOSECONDS).chain())
                                       , accordRanges, new LatencyRequestBookkeeping(cfs.metric.accordPostStreamRepair), startedAtNanos, deadlineNanos);
         }
 

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -146,7 +146,6 @@
 import static accord.local.RedundantStatus.Property.QUORUM_APPLIED;
 import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
 import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
-import static accord.utils.async.AsyncChains.getBlockingAndRethrow;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -368,14 +367,14 @@ public Partition getPartition(DecoratedKey partitionKey)
             List<Entry> cfks = new CopyOnWriteArrayList<>();
             PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key table query");
             CommandStores commandStores = AccordService.instance().node().commandStores();
-            getBlockingAndRethrow(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
+            AccordService.getBlocking(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
                 SafeCommandsForKey safeCfk = safeStore.get(key);
                 CommandsForKey cfk = safeCfk.current();
                 if (cfk == null)
                     return;
 
                 cfks.add(new Entry(safeStore.commandStore().id(), cfk));
-            }).beginAsResult());
+            }));
 
             if (cfks.isEmpty())
                 return null;
@@ -478,7 +477,7 @@ public Partition getPartition(DecoratedKey partitionKey)
             List<Entry> cfks = new CopyOnWriteArrayList<>();
             PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key_unmanaged table query");
             CommandStores commandStores = AccordService.instance().node().commandStores();
-            getBlockingAndRethrow(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
+            AccordService.getBlocking(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> {
                 SafeCommandsForKey safeCfk = safeStore.get(key);
                 CommandsForKey cfk = safeCfk.current();
                 if (cfk == null)
@@ -1469,7 +1468,7 @@ private void runWithRoute(TxnId txnId, int commandStoreId, Function<Command, BiC
                 {
                     consumer.accept(command.route(), result);
                 }
-                return result;
+                return result.chain();
             });
         }
 
@@ -1504,12 +1503,10 @@ private void recover(TxnId txnId, @Nullable Route<?> route, AsyncResult.Settable
         private void run(TxnId txnId, int commandStoreId, Function<SafeCommandStore, AsyncChain<Void>> apply)
         {
             AccordService accord = (AccordService) AccordService.instance();
-            AsyncChains.awaitUninterruptibly(accord.node()
-                                                   .commandStores()
-                                                   .forId(commandStoreId)
-                                                   .submit(PreLoadContext.contextFor(txnId, TXN_OPS), apply)
-                                                   .flatMap(i -> i)
-                                                   .beginAsResult());
+            AccordService.getBlocking(accord.node()
+                                            .commandStores()
+                                            .forId(commandStoreId)
+                                            .chain(PreLoadContext.contextFor(txnId, TXN_OPS), apply));
         }
 
         private void cleanup(TxnId txnId, int commandStoreId, Cleanup cleanup)

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -22,7 +22,6 @@
 import java.util.NavigableMap;
 import java.util.Objects;
 import java.util.concurrent.Callable;
-import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
@@ -38,6 +37,7 @@
 import org.slf4j.LoggerFactory;
 
 import accord.api.Agent;
+import accord.api.AsyncExecutor;
 import accord.api.DataStore;
 import accord.api.Journal;
 import accord.api.LocalListeners;
@@ -247,7 +247,7 @@ public AccordExecutor executor()
     // TODO (desired): we use this for executing callbacks with mutual exclusivity,
     //  but we don't need to block the actual CommandStore - could quite easily
     //  inflate a separate queue dynamically in AccordExecutor
-    public Executor taskExecutor()
+    public AsyncExecutor taskExecutor()
     {
         return exclusiveExecutor;
     }
@@ -336,25 +336,30 @@ public long nextSystemTimestampMicros()
         return lastSystemTimestampMicros;
     }
     @Override
-    public <T> AsyncChain<T> build(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
+    public <T> AsyncChain<T> chain(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
     {
         return AccordTask.create(this, loadCtx, function).chain();
     }
 
     @Override
-    public <T> AsyncChain<T> build(Callable<T> task)
+    public AsyncChain<Void> chain(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
     {
-        return AsyncChains.ofCallable(taskExecutor(), task);
+        return AccordTask.create(this, preLoadContext, consumer).chain();
     }
 
     @Override
-    public AsyncChain<Void> build(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
+    public <T> AsyncChain<T> chain(Callable<T> call)
     {
-        return AccordTask.create(this, preLoadContext, consumer).chain();
+        return taskExecutor().chain(call);
+    }
+
+    @Override
+    public void execute(Runnable run)
+    {
+        taskExecutor().execute(run);
     }
 
-    public AccordSafeCommandStore begin(AccordTask<?> operation,
-                                        @Nullable CommandsForRanges commandsForRanges)
+    public AccordSafeCommandStore begin(AccordTask<?> operation, @Nullable CommandsForRanges commandsForRanges)
     {
         require(current == null);
         current = AccordSafeCommandStore.create(operation, commandsForRanges, this);
@@ -495,7 +500,7 @@ class Ready extends CountingResult implements Runnable
                 }
             }
 
-            ready.begin((success, fail) -> {
+            ready.invoke((success, fail) -> {
                 if (fail != null)
                 {
                     logger.error("{}: failed to ensure durability of {} ({})", this, ranges, reportId, fail);
@@ -541,7 +546,7 @@ public AsyncChain<Route> replay(TxnId txnId)
             if (onlyNonDurable && !maybeShouldReplay(txnId))
                 return AsyncChains.success(null);
 
-            return store.submit(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
+            return store.chain(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
                 if (onlyNonDurable && !shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants()))
                     return null;
 

src/java/org/apache/cassandra/service/accord/AccordCommandStore.java

@@ -652,7 +652,7 @@ public long maxEpoch()
     public Future<Void> unsafeLocalSyncNotified(long epoch)
     {
         AsyncPromise<Void> promise = new AsyncPromise<>();
-        getOrCreateEpochState(epoch).localSyncNotified().begin((result, failure) -> {
+        getOrCreateEpochState(epoch).localSyncNotified().invoke((result, failure) -> {
             if (failure != null) promise.tryFailure(failure);
             else promise.trySuccess(result);
         });