Executor Fairness

src/java/org/apache/cassandra/concurrent/CassandraThread.java

@@ -18,14 +18,21 @@
 
 package org.apache.cassandra.concurrent;
 
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
 import org.apache.cassandra.metrics.ThreadLocalMetrics;
+import org.apache.cassandra.service.accord.AccordExecutor;
 
 import io.netty.util.concurrent.FastThreadLocalThread;
 
-public class CassandraThread extends FastThreadLocalThread
+public class CassandraThread extends FastThreadLocalThread implements AccordExecutor.AccordTaskRunner
 {
     private ThreadLocalMetrics threadLocalMetrics;
     private ExecutorLocals executorLocals;
+    private AccordExecutor accordActiveExecutor;
+    private AccordExecutor accordLockedExecutor;
+    private volatile AccordExecutor.Task accordActiveTask;
+    private static final AtomicReferenceFieldUpdater<CassandraThread, AccordExecutor.Task> accordActiveTaskUpdater = AtomicReferenceFieldUpdater.newUpdater(CassandraThread.class, AccordExecutor.Task.class, "accordActiveTask");
 
     private final ImmediateTaskHolder immediateTaskHolder;
 
@@ -89,6 +96,54 @@ public class CassandraThread extends FastThreadLocalThread
         return current != null ? current : ExecutorLocals.none();
     }
 
+    public final AccordExecutor accordActiveExecutor()
+    {
+        return accordActiveExecutor;
+    }
+
+    public final void setAccordActiveExecutor(AccordExecutor newExecutor)
+    {
+        accordActiveExecutor = newExecutor;
+    }
+
+    @Override
+    public final AccordExecutor accordLockedExecutor()
+    {
+        return accordLockedExecutor;
+    }
+
+    @Override
+    public final boolean trySetAccordLockedExecutor(AccordExecutor newLockedExecutor)
+    {
+        if (accordLockedExecutor != null)
+            return false;
+        accordLockedExecutor = newLockedExecutor;
+        return true;
+    }
+
+    @Override
+    public final void clearAccordLockedExecutor()
+    {
+        accordLockedExecutor = null;
+    }
+
+    public final AccordExecutor.Task accordActiveTask()
+    {
+        return accordActiveTask;
+    }
+
+    // to be called only by the thread itself, so can (eventually) avoid any memory barriers
+    public final AccordExecutor.Task accordActiveSelfTask()
+    {
+        // TODO (expected): with newer JDK use accordActiveTaskUpdater.getPlain
+        return accordActiveTask;
+    }
+
+    public final void setAccordActiveTask(AccordExecutor.Task newActiveTask)
+    {
+        accordActiveTaskUpdater.lazySet(this, newActiveTask);
+    }
+
     // final to avoid skipping of the cleanup logic in child classes
     final public void run()
     {

src/java/org/apache/cassandra/config/AccordConfig.java

@@ -35,7 +35,6 @@ import org.apache.cassandra.service.accord.serializers.Version;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 
 import static org.apache.cassandra.config.AccordConfig.CatchupMode.NORMAL;
-import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.HLC_FIFO;
 import static org.apache.cassandra.config.AccordConfig.QueueShardModel.THREAD_POOL_PER_SHARD;
 import static org.apache.cassandra.config.AccordConfig.QueueSubmissionModel.SYNC;
 import static org.apache.cassandra.config.AccordConfig.RangeIndexMode.in_memory;
@@ -134,25 +133,65 @@ public class AccordConfig
         FIFO,
 
         /**
-         * If the work has an associated TxnId, prioritise by its HLC (and FIFO otherwise)
+         * If the work has an associated TxnId of Ballot, prioritise by the newest HLC (and FIFO otherwise)
          */
         HLC_FIFO,
 
         /**
-         * Prioritise Apply, Stable, Commit, Accept, and PreAccept messages in that order.
-         * Within a given message type, prioritise by HLC.
-         * Other messages will be mixed with PreAccept messages, but using the next counter rather than the HLC of the TxnId.
-         * Note: this can have some performance edge cases for contended keys, as we may process Stable messages for later commands before
-         *       we process earlier Accept/Commit, which may delay execution
+         * If the work has an associated TxnId, prioritise by its HLC (and FIFO otherwise)
          */
-        PHASE_HLC_FIFO,
+        ORIG_HLC_FIFO
+    }
+
+    public enum QueueBalancingModel
+    {
+        /**
+         * Always pick the task by priority
+         */
+        PRIORITY_ONLY,
 
         /**
-         * Prioritise Apply, Stable, Commit, Accept, and PreAccept messages from the original coordinator only, in that order.
-         * Within a given message type, prioritise by HLC.
-         * Other messages will be mixed with PreAccept messages, but using the next counter rather than the HLC of the TxnId.
+         * Pick by phase first, so the highest phase with work ALWAYS runs.
+         * Within a phase, pick by priority.
          */
-        ORIG_PHASE_HLC_FIFO
+        PHASE_ONLY,
+
+        /**
+         * Always pick the task by priority.
+         */
+        PRIORITY_BUDGET,
+
+        /**
+         * Pick by phase first, so the highest phase with budget ALWAYS runs.
+         * Within a phase, pick by priority.
+         */
+        PHASE_BUDGET,
+
+        /**
+         * Pick by phase first, selecting the phase that has processed the least work recently relative to arrivals.
+         * Within a phase, pick by priority.
+         */
+        PHASE_FAIR,
+
+        /**
+         * Pick by phase first, selecting the phase that has processed the least work recently relative to arrivals.
+         * However, each phase has a budget that is consumed on dispatch, and we pick only from those phases with budget.
+         * If there is no phase with budget, the budget resets.
+         * Within a phase, pick by priority.
+         */
+        PHASE_BUDGET_FAIR,
+
+        /**
+         * While phases are within a threshold of imbalance, pick tasks by priority.
+         * Once the threshold is crossed, over-processed phases have a small penalty applied
+         * and work is prioritised by phase with the phase with the least recent work processed picked first,
+         * until the imbalance is resolved.
+         *
+         * Within a phase, pick by priority.
+         */
+        BLENDED_PRIORITY_PHASE_FAIR,
+
+        BLENDED_PRIORITY_PHASE_BUDGET_FAIR,
     }
 
     public QueueShardModel queue_shard_model = THREAD_POOL_PER_SHARD;
@@ -168,7 +207,13 @@ public class AccordConfig
      */
     public volatile OptionaldPositiveInt queue_thread_count = OptionaldPositiveInt.UNDEFINED;
 
-    public QueuePriorityModel queue_priority_model = HLC_FIFO;
+    public QueuePriorityModel queue_priority_model;
+    public QueueBalancingModel queue_balancing_model;
+
+    public Integer queue_flow_imbalance_onset = null;
+    public Integer queue_flow_imbalance_width_shift = null;
+    public String queue_active_limits;
+    public String queue_budgets;
 
     /**
      * If set, the signal loop does not match park/unpark pairs, but instead consumers perform timed-park spin waits
@@ -181,15 +226,6 @@ public class AccordConfig
      */
     public DurationSpec.LongMicrosecondsBound queue_stop_check_interval;
 
-    /**
-     * If set, the signal loop reduces the number of threads it is using when the time spent parked exceeds real-time
-     * by this interval.
-     */
-    public DurationSpec.LongMicrosecondsBound queue_signal_stop_check_interval_credit;
-
-    // yield to other executor threads after executing this many tasks in a row, if there are waiting threads and tasks
-    public int queue_yield_interval = 100;
-
     /**
      * If the HLC is older than this, queue by FIFO instead
      */

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

@@ -1875,7 +1875,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
             Node node = AccordService.unsafeInstance().node();
             if (Route.isFullRoute(route))
             {
-                PrepareRecovery.recover(node, node.someSequentialExecutor(), txnId, NotKnownToBeInvalid, (FullRoute<?>) route, null, LatentStoreSelector.standard(), (success, fail) -> {
+                PrepareRecovery.recover(node, node.someExclusiveExecutor(), txnId, NotKnownToBeInvalid, (FullRoute<?>) route, null, LatentStoreSelector.standard(), (success, fail) -> {
                     if (fail != null) result.setFailure(fail);
                     else result.setSuccess(null);
                 });

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

@@ -345,7 +345,6 @@ public class AccordCache implements CacheSize
 
     <P1, P2, K, V> Collection<AccordTask<?>> load(LoadExecutor<P1, P2> loadExecutor, P1 p1, P2 p2, AccordCacheEntry<K, V> node)
     {
-        Type<K, V, ?> parent = node.owner.parent();
         return node.load(loadExecutor, p1, p2).waiters();
     }
 

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

@@ -27,7 +27,6 @@ import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
-import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.IntFunction;
@@ -59,12 +58,12 @@ import accord.local.CommandStore;
 import accord.local.CommandStores.RangesForEpoch;
 import accord.local.CommandSummaries;
 import accord.local.ExecutionContext;
+import accord.local.ExecutionContext.Empty;
 import accord.local.MaxConflicts;
 import accord.local.MaxDecidedRX;
 import accord.local.MinimalCommand;
 import accord.local.MinimalCommand.MinimalWithDeps;
 import accord.local.NodeCommandStoreService;
-import accord.local.ExecutionContext.Empty;
 import accord.local.RedundantBefore;
 import accord.local.RedundantBefore.Bounds;
 import accord.local.RedundantStatus.Property;
@@ -100,6 +99,7 @@ import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TableMetadataRef;
 import org.apache.cassandra.service.accord.AccordDurableOnFlush.ReportDurable;
+import org.apache.cassandra.service.accord.AccordExecutor.AccordTaskRunner;
 import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
 import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo;
 import org.apache.cassandra.service.accord.api.TokenKey;
@@ -167,12 +167,12 @@ public class AccordCommandStore extends CommandStore
 
     public static final class ExclusiveCaches extends Caches implements CommandStoreCaches<AccordSafeCommand, AccordSafeCommandsForKey>
     {
-        private final Lock lock;
+        private final AccordExecutor owner;
 
-        public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys)
+        public ExclusiveCaches(AccordExecutor owner, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys)
         {
             super(global, commands, commandsForKeys);
-            this.lock = lock;
+            this.owner = owner;
         }
 
         @Override
@@ -190,8 +190,8 @@ public class AccordCommandStore extends CommandStore
         @Override
         public void close()
         {
-            global().tryShrinkOrEvict(lock);
-            lock.unlock();
+            global().tryShrinkOrEvict(owner.unsafeLock());
+            owner.unlock(AccordTaskRunner.get());
         }
     }
 
@@ -261,7 +261,7 @@ public class AccordCommandStore extends CommandStore
         {
             commands = exclusive.commands.newInstance(this);
             commandsForKey = exclusive.commandsForKey.newInstance(this);
-            this.caches = new ExclusiveCaches(sharedExecutor.unsafeLock(), exclusive.global, commands, commandsForKey);
+            this.caches = new ExclusiveCaches(sharedExecutor, exclusive.global, commands, commandsForKey);
         }
         this.exclusiveExecutor = sharedExecutor.executor(id);
 
@@ -323,13 +323,13 @@ public class AccordCommandStore extends CommandStore
     public ExclusiveCaches lockCaches()
     {
         //noinspection LockAcquiredButNotSafelyReleased
-        caches.lock.lock();
+        caches.owner.lock(AccordTaskRunner.get());
         return caches;
     }
 
     public ExclusiveCaches tryLockCaches()
     {
-        if (caches.lock.tryLock())
+        if (caches.owner.tryLock(AccordTaskRunner.get()))
             return caches;
         return null;
     }
@@ -403,27 +403,15 @@ public class AccordCommandStore extends CommandStore
         return lastSystemTimestampMicros.accumulateAndGet(node.now(), (a, b) -> Math.max(a + 1, b));
     }
     @Override
-    public <T> AsyncChain<T> chain(ExecutionContext loadCtx, Function<? super SafeCommandStore, T> function)
+    public <T> AsyncChain<T> chain(ExecutionContext context, Function<? super SafeCommandStore, T> function)
     {
-        return AccordTask.create(this, loadCtx, function).chain();
+        return AccordTask.create(this, context, function).chain();
     }
 
     @Override
-    public AsyncChain<Void> chain(ExecutionContext executionContext, Consumer<? super SafeCommandStore> consumer)
+    public AsyncChain<Void> chain(ExecutionContext context, Consumer<? super SafeCommandStore> consumer)
     {
-        return AccordTask.create(this, executionContext, consumer).chain();
-    }
-
-    @Override
-    public AsyncChain<Void> priorityChain(ExecutionContext executionContext, Consumer<? super SafeCommandStore> consumer)
-    {
-        return AccordTask.create(this, executionContext, consumer).priorityChain();
-    }
-
-    @Override
-    public <T> AsyncChain<T> priorityChain(ExecutionContext executionContext, Function<? super SafeCommandStore, T> function)
-    {
-        return AccordTask.create(this, executionContext, function).priorityChain();
+        return AccordTask.create(this, context, consumer).chain();
     }
 
     @Override

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

@@ -31,12 +31,12 @@ import org.slf4j.LoggerFactory;
 
 import accord.api.Agent;
 import accord.api.DataStore;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Journal;
 import accord.api.LocalListeners;
 import accord.api.ProgressLog;
 import accord.local.CommandStores;
 import accord.local.NodeCommandStoreService;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.local.ShardDistributor;
 import accord.utils.RandomSource;
 import accord.utils.Reduce;
@@ -148,10 +148,10 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
     }
 
     @Override
-    public ExclusiveAsyncExecutor someSequentialExecutor()
+    public ExclusiveAsyncExecutor someExclusiveExecutor()
     {
         int idx = ((int) Thread.currentThread().getId()) & mask;
-        return executors[idx].newSequentialExecutor();
+        return executors[idx].newExclusiveExecutor();
     }
 
     public synchronized void setCapacity(long bytes)

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

@@ -24,7 +24,7 @@ import accord.api.Agent;
 import accord.utils.QuadFunction;
 import accord.utils.QuintConsumer;
 
-import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.concurrent.CassandraThread;
 import org.apache.cassandra.service.accord.AccordExecutorLoops.LoopTask;
 import org.apache.cassandra.service.accord.debug.DebugExecution.DebugExecutorLoop;
 
@@ -33,7 +33,6 @@ import static org.apache.cassandra.service.accord.debug.DebugExecution.DEBUG_EXE
 
 abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
 {
-    private static final int YIELD_INTERVAL = DatabaseDescriptor.getAccord().queue_yield_interval;
     int runningThreads;
     boolean shutdown;
 
@@ -44,11 +43,10 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
 
     abstract void notifyWork();
     abstract void notifyWorkExclusive();
-    void loopYieldExclusive() throws InterruptedException {}
     abstract void awaitExclusive() throws InterruptedException;
     abstract <P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Task> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4);
 
-    <P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Task> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
+    final <P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Task> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
     {
         // if we're a loop thread, we will poll the waitingToRun queue when we come around
         // NOTE: this assumes no synchronous blocking tasks are submitted to this executor
@@ -56,7 +54,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
         else submitExternal(sync, async, p1s, p1a, p2, p3, p4);
     }
 
-    <P1s, P2, P3, P4> void submitExternalExclusive(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, P1s p1s, P2 p2, P3 p3, P4 p4)
+    final <P1s, P2, P3, P4> void submitExternalExclusive(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, P1s p1s, P2 p2, P3 p3, P4 p4)
     {
         try
         {
@@ -84,6 +82,73 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
             notifyWorkExclusive();
     }
 
+    final boolean hasWaitingToRun()
+    {
+        updateWaitingToRunExclusive();
+        return hasAlreadyWaitingToRun();
+    }
+
+    final Task pollWaitingToRunExclusive()
+    {
+        updateWaitingToRunExclusive();
+        return pollAlreadyWaitingToRunExclusive();
+    }
+
+    final void updateWaitingToRunExclusive()
+    {
+        drainUnqueuedExclusive();
+        super.updateWaitingToRunExclusive();
+    }
+
+    final void drainUnqueuedExclusive()
+    {
+        Task cur = Task.reverse(acquireUnqueuedExclusive());
+        while (cur != null)
+            cur = enqueueOneExclusive(cur);
+    }
+
+    final void drainUnqueuedNewWorkExclusive()
+    {
+        Task cur = acquireUnqueuedExclusive();
+        Task prev = null, requeue = null, requeueLast = null;
+        while (cur != null)
+        {
+            Task next = cur.next;
+            if (cur.isNewWork())
+            {
+                cur.next = prev;
+                prev = cur;
+            }
+            else
+            {
+                if (requeue == null) requeue = cur;
+                else requeueLast.next = cur;
+                requeueLast = cur;
+            }
+            cur = next;
+        }
+
+        if (requeue != null)
+        {
+            while (true)
+            {
+                Task next = unqueued;
+                requeueLast.next = next;
+                if (unqueuedUpdater.compareAndSet(this, next, requeue))
+                    break;
+            }
+        }
+
+        cur = prev;
+        while (cur != null)
+        {
+            Task next = cur.next;
+            cur.submitExclusive(this);
+            cur.next = null;
+            cur = next;
+        }
+    }
+
     @Override
     final void beforeUnlockExternal()
     {
@@ -130,11 +195,15 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
             @Override
             public void run()
             {
-                Thread self = Thread.currentThread();
+                Thread thread = Thread.currentThread();
+                AccordTaskRunner self = AccordTaskRunner.get(thread);
+                self.setAccordActiveExecutor(AccordExecutorAbstractLockLoop.this);
+                setWrapped(self);
+
                 Task task;
                 while (true)
                 {
-                    lock();
+                    lock(self);
                     try
                     {
                         enterLockLoop();
@@ -144,7 +213,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
 
                             if (task != null)
                             {
-                                setRunning(task);
+                                self.setAccordActiveTask(task);
                                 try
                                 {
                                     task.preRunExclusive();
@@ -157,7 +226,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
                                 finally
                                 {
                                     completeTaskExclusive(task);
-                                    clearRunning();
+                                    self.setAccordActiveTask(null);
                                 }
                             }
                             else
@@ -185,7 +254,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
                     }
                     finally
                     {
-                        unlock();
+                        unlock(self);
                     }
                 }
             }
@@ -200,12 +269,15 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
             @Override
             public void run()
             {
-                int count = 0;
+                CassandraThread self = (CassandraThread) Thread.currentThread();
+                self.setAccordActiveExecutor(AccordExecutorAbstractLockLoop.this);
+                setWrapped(self);
+
                 Task task = null;
                 while (true)
                 {
                     if (DEBUG_EXECUTION) debug.onLock();
-                    lock();
+                    lock(self);
                     try
                     {
                         if (DEBUG_EXECUTION) debug.onEnterLock();
@@ -215,13 +287,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
                             Task tmp = task;
                             task = null;
                             completeTaskExclusive(tmp);
-                            clearRunning();
-                        }
-
-                        if (count >= YIELD_INTERVAL)
-                        {
-                            loopYieldExclusive();
-                            count = 0;
+                            self.setAccordActiveTask(null);
                         }
 
                         while (true)
@@ -230,7 +296,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
 
                             if (task != null)
                             {
-                                setRunning(task);
+                                self.setAccordActiveTask(task);
                                 task.preRunExclusive();
                                 if (DEBUG_EXECUTION) debug.onExitLock();
                                 exitLockLoop();
@@ -250,7 +316,6 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
                             awaitExclusive();
                             if (DEBUG_EXECUTION) debug.onEnterLock();
                             resumeLoop();
-                            count = 0;
                         }
                     }
                     catch (Throwable t)
@@ -276,12 +341,11 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
                     finally
                     {
                         if (DEBUG_EXECUTION) debug.onExitLock();
-                        unlock();
+                        unlock(self);
                     }
 
                     try
                     {
-                        ++count;
                         task.run();
                     }
                     catch (Throwable t)

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

@@ -29,8 +29,8 @@ import org.apache.cassandra.concurrent.DebuggableTask.DebuggableTaskRunner;
 
 abstract class AccordExecutorAbstractLoop extends AccordExecutor
 {
-    private volatile Task unqueued;
-    private static final AtomicReferenceFieldUpdater<AccordExecutorAbstractLoop, Task> unqueuedUpdater = AtomicReferenceFieldUpdater.newUpdater(AccordExecutorAbstractLoop.class, Task.class, "unqueued");
+    volatile Task unqueued;
+    static final AtomicReferenceFieldUpdater<AccordExecutorAbstractLoop, Task> unqueuedUpdater = AtomicReferenceFieldUpdater.newUpdater(AccordExecutorAbstractLoop.class, Task.class, "unqueued");
 
     AccordExecutorAbstractLoop(Lock lock, int executorId, Agent agent)
     {
@@ -67,30 +67,18 @@ abstract class AccordExecutorAbstractLoop extends AccordExecutor
         if (hasUnqueued() || tasks > 0)
             return true;
 
-        lock();
+        AccordTaskRunner self = AccordTaskRunner.get();
+        lock(self);
         try
         {
             return hasUnqueued() || tasks > 0;
         }
         finally
         {
-            unlock();
+            unlock(self);
         }
     }
 
-    final void updateWaitingToRunExclusive()
-    {
-        drainUnqueuedExclusive();
-        super.updateWaitingToRunExclusive();
-    }
-
-    final void drainUnqueuedExclusive()
-    {
-        Task cur = Task.reverse(acquireUnqueuedExclusive());
-        while (cur != null)
-            cur = enqueueOneExclusive(cur);
-    }
-
     final Task acquireUnqueuedExclusive()
     {
         return unqueuedUpdater.getAndSet(this, null);

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

@@ -46,18 +46,6 @@ public class AccordExecutorSemiSyncSubmit extends AccordExecutorAbstractSemiSync
         this.loops = new AccordExecutorLoops(mode, threads, name, this::task);
     }
 
-    @Override
-    void loopYieldExclusive() throws InterruptedException
-    {
-        if (waiting > 0 && hasWaitingToRun())
-        {
-            pauseLoop();
-            hasWork.signal();
-            awaitWork();
-            resumeLoop();
-        }
-    }
-
     @Override
     void awaitExclusive() throws InterruptedException
     {

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

@@ -100,6 +100,217 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
         return new LoopTask(index, id);
     }
 
+    @Override
+    final void drainUnqueuedNewWorkExclusive()
+    {
+        updatePendingUnqueued();
+        Task requeue = null, requeueTail = null;
+        Task cur = pendingNewHead;
+        while (cur != null)
+        {
+            Task next = cur.next;
+            if (cur.isNewWork())
+            {
+                cur.next = null;
+                cur.submitExclusive(this);
+                --pendingCount;
+            }
+            else
+            {
+                if (requeue == null) requeue = cur;
+                else requeueTail.next = cur;
+                requeueTail = cur;
+            }
+            cur = next;
+        }
+
+        pendingNewHead = requeue;
+        pendingNewTail = requeueTail;
+    }
+
+    private boolean enqueueOnePending()
+    {
+        if (pendingCount == 0)
+            return false;
+
+        if (pendingSequentialHead != null)
+        {
+            pendingSequentialHead = enqueueOneCleanup(pendingSequentialHead);
+            if (pendingSequentialHead == null)
+                pendingSequentialTail = null;
+        }
+        else if (pendingCleanupHead != null)
+        {
+            pendingCleanupHead = enqueueOneCleanup(pendingCleanupHead);
+            if (pendingCleanupHead == null)
+                pendingCleanupTail = null;
+        }
+        else
+        {
+            pendingNewHead = enqueueOneSubmit(pendingNewHead);
+            if (pendingNewHead == null)
+                pendingNewTail = null;
+        }
+        --pendingCount;
+        return true;
+    }
+
+    private void fetchWorkExclusive()
+    {
+        boolean hasReadyToRun = hasReadyToRun();
+        boolean hadWaitingToRun = hasAlreadyWaitingToRun();
+        {
+            int prevPendingUnqueued = pendingCount;
+            updateAndEnqueuePendingUntilHasWaitingToRun(prevPendingUnqueued / 2);
+        }
+
+        if (hadWaitingToRun && hasReadyToRun)
+        {
+            lock.addAndGetEnabledThreadCount(1);
+        }
+        else if (hadWaitingToRun)
+        {
+            readyToRunTarget = Math.min(readyToRunLimit, readyToRunTarget + (1+readyToRunTarget)/2);
+        }
+        else if (hasReadyToRun && readyToRunTarget > 1)
+        {
+            --readyToRunTarget;
+        }
+
+        boolean hasDrainedSignal = false;
+        while (true)
+        {
+            long state = lock.state();
+            int signals = SignalLock.asyncSignalCount(state);
+            int waiters = SignalLock.waitingEnabledThreadCount(state);
+            if (signals >= readyToRunTarget)
+            {
+                if (enqueueOnePending() || (updatePendingUnqueued() && enqueueOnePending())) continue;
+                else if (hasDrainedSignal)
+                    lock.signalLockWorkExclusive();
+                return;
+            }
+            else if (waiters > 0 && signals > 1 && SignalLock.activeEnabledThreadCount(state) == 1)
+            {
+                // ensure at least one other thread is running if there's enough work for it; it will spin up other threads if necessary
+                lock.propagateAsyncWorkSignals(1);
+            }
+
+            Task task = pollAlreadyWaitingToRunExclusive();
+            if (task == null)
+            {
+                if (updateAndEnqueuePendingUntilHasWaitingToRun(0))
+                    continue;
+
+                lock.clearLockWork();
+                hasDrainedSignal = true;
+                if (!updateAndEnqueuePendingUntilHasWaitingToRun(0))
+                {
+                    if (tasks == 0)
+                        notifyQuiescentExclusive();
+                    return;
+                }
+            }
+            else
+            {
+                try { task.preRunExclusive(); }
+                catch (Throwable t)
+                {
+                    try { task.fail(t); }
+                    catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
+                    try { completeTaskExclusive(task); }
+                    catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
+                    continue;
+                }
+                if (DEBUG_EXECUTION) DebugTask.get(task).onPreRun();
+
+                addReadyToRun(task);
+                boolean incremented = lock.incrementAsyncWork(false);
+                Invariants.require(incremented);
+            }
+        }
+    }
+
+    private boolean updatePendingUnqueued()
+    {
+        if (!hasUnqueued())
+            return false;
+
+        int count = 0;
+        Task addSequentialHead = null, addSequentialTail = null;
+        Task addCleanupHead = null, addCleanupTail = null;
+        Task addNewHead = null, addNewTail = null;
+        {
+            Task cur = Task.reverse(acquireUnqueuedExclusive());
+            while (cur != null)
+            {
+                Task next = cur.next;
+                if (!cur.isReadyToCleanup())
+                {
+                    if (addNewHead == null) addNewHead = addNewTail = setNextNull(cur);
+                    else addNewHead = reverseOne(addNewHead, cur);
+                }
+                else if (cur instanceof ExclusiveExecutorTask)
+                {
+                    if (addSequentialHead == null) addSequentialHead = addSequentialTail = setNextNull(cur);
+                    else addSequentialHead = reverseOne(addSequentialHead, cur);
+                }
+                else
+                {
+                    if (addCleanupHead == null) addCleanupHead = addCleanupTail = setNextNull(cur);
+                    else addCleanupHead = reverseOne(addCleanupHead, cur);
+                }
+                ++count;
+                cur = next;
+            }
+        }
+
+        pendingCount += count;
+        if (addSequentialHead != null)
+        {
+            if (pendingSequentialHead == null) pendingSequentialHead = addSequentialHead;
+            else pendingSequentialTail.next = addSequentialHead;
+            pendingSequentialTail = addSequentialTail;
+        }
+        if (addCleanupHead != null)
+        {
+            if (pendingCleanupHead == null) pendingCleanupHead = addCleanupHead;
+            else pendingCleanupTail.next = addCleanupHead;
+            pendingCleanupTail = addCleanupTail;
+        }
+        if (addNewHead != null)
+        {
+            if (pendingNewHead == null) pendingNewHead = addNewHead;
+            else pendingNewTail.next = addNewHead;
+            pendingNewTail = addNewTail;
+        }
+        return true;
+    }
+
+    private Task reverseOne(Task prev, Task cur)
+    {
+        cur.next = prev;
+        return cur;
+    }
+
+    private Task setNextNull(Task cur)
+    {
+        cur.next = null;
+        return cur;
+    }
+
+    private boolean updateAndEnqueuePendingUntilHasWaitingToRun(int processAtLeast)
+    {
+        updatePendingUnqueued();
+        int count = 0;
+        while (enqueueOnePending())
+        {
+            if (++count >= processAtLeast && hasAlreadyWaitingToRun())
+                return true;
+        }
+        return hasAlreadyWaitingToRun();
+    }
+
     class LoopTask extends AccordExecutorLoops.LoopTask
     {
         final int index;
@@ -110,7 +321,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
             this.index = index;
         }
 
-        private Task awaitWork()
+        private Task awaitWork(AccordTaskRunner self)
         {
             while (true)
             {
@@ -123,11 +334,11 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
                     }
                     catch (Throwable t)
                     {
-                        unlock();
+                        unlock(self);
                         throw t;
                     }
 
-                    if (!unlockAndAcquire())
+                    if (!unlockAndAcquire(self))
                         continue;
                 }
 
@@ -138,11 +349,8 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
             }
         }
 
-        private Task cleanupAndMaybeGetWork(@Nullable Task cleanup)
+        private Task cleanupAndMaybeGetWork(AccordTaskRunner self, @Nullable Task cleanup)
         {
-            if (cleanup == null)
-                return null;
-
             if (lock.tryAcquireAsyncWork())
             {
                 if (shutdown)
@@ -151,22 +359,21 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
                 return pushCleanupAndReturn(cleanup, nonNull(pollReadyToRun()));
             }
 
-            if (!tryLock())
+            if (!tryLock(self))
                 return pushCleanupAndReturn(cleanup, null);
 
             try
             {
                 completeTaskExclusive(cleanup);
-                clearRunning();
                 fetchWorkExclusive();
             }
             catch (Throwable t)
             {
-                unlock();
+                unlock(self);
                 throw t;
             }
 
-            if (unlockAndAcquire())
+            if (unlockAndAcquire(self))
             {
                 if (shutdown)
                     throw new ShutdownException();
@@ -178,225 +385,52 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
 
         private Task pushCleanupAndReturn(Task cleanup, Task result)
         {
-            clearRunning();
             cleanup.setReadyToCleanup();
             if (push(cleanup) == null)
                 lock.signalLockWork();
             return result;
         }
 
-        final boolean tryLock()
+        final boolean tryLock(AccordTaskRunner self)
         {
-            return onTryLock(lock.tryLock(index));
+            if (self.accordLockedExecutor() != null)
+                return false;
+
+            return onTryLock(self, lock.tryLock(index));
         }
 
-        final boolean unlockAndAcquire()
+        final boolean unlockAndAcquire(AccordTaskRunner self)
         {
-            if (Invariants.isParanoid()) paranoidUnlockExclusive();
+            self.clearAccordLockedExecutor();
             if (DEBUG_EXECUTION) debug.onExitLock();
             return lock.unlockAndAcquireAsyncWork();
         }
 
-        private boolean enqueueOnePending()
-        {
-            if (pendingCount == 0)
-                return false;
-
-            if (pendingSequentialHead != null)
-            {
-                pendingSequentialHead = enqueueOneCleanup(pendingSequentialHead);
-                if (pendingSequentialHead == null)
-                    pendingSequentialTail = null;
-            }
-            else if (pendingCleanupHead != null)
-            {
-                pendingCleanupHead = enqueueOneCleanup(pendingCleanupHead);
-                if (pendingCleanupHead == null)
-                    pendingCleanupTail = null;
-            }
-            else
-            {
-                pendingNewHead = enqueueOneSubmit(pendingNewHead);
-                if (pendingNewHead == null)
-                    pendingNewTail = null;
-            }
-            --pendingCount;
-            return true;
-        }
-
-        private void fetchWorkExclusive()
-        {
-            boolean hasReadyToRun = hasReadyToRun();
-            boolean hadWaitingToRun = hasAlreadyWaitingToRun();
-            {
-                int prevPendingUnqueued = pendingCount;
-                updateAndEnqueuePendingUntilHasWaitingToRun(prevPendingUnqueued / 2);
-            }
-
-            if (hadWaitingToRun && hasReadyToRun)
-            {
-                lock.addAndGetEnabledThreadCount(1);
-            }
-            else if (hadWaitingToRun)
-            {
-                readyToRunTarget = Math.min(readyToRunLimit, readyToRunTarget + (1+readyToRunTarget)/2);
-            }
-            else if (hasReadyToRun && readyToRunTarget > 1)
-            {
-                --readyToRunTarget;
-            }
-
-            boolean hasDrainedSignal = false;
-            while (true)
-            {
-                long state = lock.state();
-                int signals = SignalLock.asyncSignalCount(state);
-                int waiters = SignalLock.waitingEnabledThreadCount(state);
-                if (signals >= readyToRunTarget)
-                {
-                    if (enqueueOnePending() || (updatePendingUnqueued() && enqueueOnePending())) continue;
-                    else if (hasDrainedSignal)
-                        lock.signalLockWorkExclusive();
-                    return;
-                }
-                else if (waiters > 0 && signals > 1 && SignalLock.activeEnabledThreadCount(state) == 1)
-                {
-                    // ensure at least one other thread is running if there's enough work for it; it will spin up other threads if necessary
-                    lock.propagateAsyncWorkSignals(1);
-                }
-
-                Task task = pollAlreadyWaitingToRunExclusive();
-                if (task == null)
-                {
-                    if (updateAndEnqueuePendingUntilHasWaitingToRun(0))
-                        continue;
-
-                    lock.clearLockWork();
-                    hasDrainedSignal = true;
-                    if (!updateAndEnqueuePendingUntilHasWaitingToRun(0))
-                    {
-                        if (tasks == 0)
-                            notifyQuiescentExclusive();
-                        return;
-                    }
-                }
-                else
-                {
-                    try { task.preRunExclusive(); }
-                    catch (Throwable t)
-                    {
-                        try { task.fail(t); }
-                        catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
-                        try { completeTaskExclusive(task); }
-                        catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
-                        continue;
-                    }
-                    if (DEBUG_EXECUTION) DebugTask.get(task).onPreRun();
-
-                    addReadyToRun(task);
-                    boolean incremented = lock.incrementAsyncWork(false);
-                    Invariants.require(incremented);
-                }
-            }
-        }
-
-        private boolean updatePendingUnqueued()
-        {
-            if (!hasUnqueued())
-                return false;
-
-            int count = 0;
-            Task addSequentialHead = null, addSequentialTail = null;
-            Task addCleanupHead = null, addCleanupTail = null;
-            Task addNewHead = null, addNewTail = null;
-            {
-                Task cur = Task.reverse(acquireUnqueuedExclusive());
-                while (cur != null)
-                {
-                    Task next = cur.next;
-                    if (!cur.isReadyToCleanup())
-                    {
-                        if (addNewHead == null) addNewHead = addNewTail = setNextNull(cur);
-                        else addNewHead = reverseOne(addNewHead, cur);
-                    }
-                    else if (cur instanceof SequentialQueueTask)
-                    {
-                        if (addSequentialHead == null) addSequentialHead = addSequentialTail = setNextNull(cur);
-                        else addSequentialHead = reverseOne(addSequentialHead, cur);
-                    }
-                    else
-                    {
-                        if (addCleanupHead == null) addCleanupHead = addCleanupTail = setNextNull(cur);
-                        else addCleanupHead = reverseOne(addCleanupHead, cur);
-                    }
-                    ++count;
-                    cur = next;
-                }
-            }
-
-            pendingCount += count;
-            if (addSequentialHead != null)
-            {
-                if (pendingSequentialHead == null) pendingSequentialHead = addSequentialHead;
-                else pendingSequentialTail.next = addSequentialHead;
-                pendingSequentialTail = addSequentialTail;
-            }
-            if (addCleanupHead != null)
-            {
-                if (pendingCleanupHead == null) pendingCleanupHead = addCleanupHead;
-                else pendingCleanupTail.next = addCleanupHead;
-                pendingCleanupTail = addCleanupTail;
-            }
-            if (addNewHead != null)
-            {
-                if (pendingNewHead == null) pendingNewHead = addNewHead;
-                else pendingNewTail.next = addNewHead;
-                pendingNewTail = addNewTail;
-            }
-            return true;
-        }
-
-        private Task reverseOne(Task prev, Task cur)
-        {
-            cur.next = prev;
-            return cur;
-        }
-
-        private Task setNextNull(Task cur)
-        {
-            cur.next = null;
-            return cur;
-        }
-
-        private boolean updateAndEnqueuePendingUntilHasWaitingToRun(int processAtLeast)
-        {
-            updatePendingUnqueued();
-            int count = 0;
-            while (enqueueOnePending())
-            {
-                if (++count >= processAtLeast && hasAlreadyWaitingToRun())
-                    return true;
-            }
-            return hasAlreadyWaitingToRun();
-        }
-
         @Override
         public void run()
         {
-            lock.register(index, Thread.currentThread());
+            Thread thread = Thread.currentThread();
+            AccordTaskRunner self = AccordTaskRunner.get(thread);
+            self.setAccordActiveExecutor(AccordExecutorSignalLoop.this);
+            setWrapped(self);
+
+            lock.register(index, thread);
             Task task = null;
             while (true)
             {
                 try
                 {
-                    try { task = cleanupAndMaybeGetWork(task); }
-                    catch (Throwable t) { task = null; throw t; }
+                    if (task != null)
+                    {
+                        try { task = cleanupAndMaybeGetWork(self, task); }
+                        catch (Throwable t) { task = null; throw t; }
+                    }
                     if (task == null)
-                        task = awaitWork();
+                        task = awaitWork(self);
 
                     try
                     {
-                        setRunning(task);
+                        self.setAccordActiveTask(task);
                         task.run();
                     }
                     catch (Throwable t)
@@ -421,6 +455,10 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
                 {
                     agent.onException(t);
                 }
+                finally
+                {
+                    self.setAccordActiveTask(null);
+                }
             }
         }
     }

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

@@ -128,6 +128,18 @@ class AccordExecutorSimple extends AccordExecutor
         }
     }
 
+    final boolean hasWaitingToRun()
+    {
+        updateWaitingToRunExclusive();
+        return hasAlreadyWaitingToRun();
+    }
+
+    final Task pollWaitingToRunExclusive()
+    {
+        updateWaitingToRunExclusive();
+        return pollAlreadyWaitingToRunExclusive();
+    }
+
     @Override
     boolean isOwningThread()
     {

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

@@ -91,14 +91,15 @@ public class AccordExecutorSyncSubmit extends AccordExecutorAbstractLockLoop
 
     <P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Task> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
     {
-        lock();
+        AccordTaskRunner self = AccordTaskRunner.get();
+        lock(self);
         try
         {
             submitExternalExclusive(sync, p1s, p2, p3, p4);
         }
         finally
         {
-            unlock();
+            unlock(self);
         }
     }
 

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

@@ -397,7 +397,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
 
     public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore) throws TopologyException
     {
-        super(node, node.someSequentialExecutor(), ranges, syncPoint, fetchRanges, commandStore);
+        super(node, node.someExclusiveExecutor(), ranges, syncPoint, fetchRanges, commandStore);
     }
 
     @Override

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

@@ -64,7 +64,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
                                    @Nullable CommandSummaries commandsForRanges,
                                    AccordCommandStore commandStore)
     {
-        super(task.preLoadContext(), commandStore);
+        super(task.executionContext(), commandStore);
         this.task = task;
         this.commandsForRanges = commandsForRanges;
         this.commandStore = commandStore;

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

@@ -964,7 +964,7 @@ public class AccordService implements IAccordService, Shutdownable
             return syncInternal(minBound, keys, syncLocal, syncRemote);
 
         return KeyBarriers.find(node, minBound, keys.get(0).toUnseekable(), syncLocal, syncRemote).chain()
-                          .flatMap(found -> KeyBarriers.await(node, node.someSequentialExecutor(), found, syncLocal, syncRemote))
+                          .flatMap(found -> KeyBarriers.await(node, node.someExclusiveExecutor(), found, syncLocal, syncRemote))
                           .flatMap(success -> {
                               if (success)
                                   return null;

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

@@ -71,7 +71,6 @@ import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
 import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
 import org.apache.cassandra.service.accord.AccordCommandStore.Caches;
 import org.apache.cassandra.service.accord.AccordExecutor.Task;
-import org.apache.cassandra.service.accord.AccordExecutor.TaskQueue;
 import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
 import org.apache.cassandra.service.accord.api.TokenKey;
 import org.apache.cassandra.service.accord.debug.DebugExecution.DebugTask;
@@ -87,18 +86,17 @@ import static accord.local.LoadKeysFor.WRITE;
 import static accord.primitives.Routable.Domain.Key;
 import static accord.primitives.Txn.Kind.EphemeralRead;
 import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
-import static org.apache.cassandra.service.accord.AccordTask.State.ASSIGNED;
-import static org.apache.cassandra.service.accord.AccordTask.State.CANCELLED;
-import static org.apache.cassandra.service.accord.AccordTask.State.FAILED;
-import static org.apache.cassandra.service.accord.AccordTask.State.FINISHED;
-import static org.apache.cassandra.service.accord.AccordTask.State.INITIALIZED;
-import static org.apache.cassandra.service.accord.AccordTask.State.LOADING;
-import static org.apache.cassandra.service.accord.AccordTask.State.PERSISTING;
-import static org.apache.cassandra.service.accord.AccordTask.State.RUNNING;
-import static org.apache.cassandra.service.accord.AccordTask.State.SCANNING_RANGES;
-import static org.apache.cassandra.service.accord.AccordTask.State.WAITING_TO_LOAD;
-import static org.apache.cassandra.service.accord.AccordTask.State.WAITING_TO_RUN;
-import static org.apache.cassandra.service.accord.AccordTask.State.WAITING_TO_SCAN_RANGES;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.ASSIGNED;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.CANCELLED;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.FAILED;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.FINISHED;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.LOADING;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.PERSISTING;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.RUNNING;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.SCANNING_RANGES;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.WAITING_TO_LOAD;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.WAITING_TO_RUN;
+import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.WAITING_TO_SCAN_RANGES;
 import static org.apache.cassandra.service.accord.debug.DebugExecution.DEBUG_EXECUTION;
 import static org.apache.cassandra.service.accord.debug.DebugExecution.DebugTask.SANITY_CHECK;
 
@@ -111,9 +109,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     {
         private final Function<? super SafeCommandStore, R> function;
 
-        public ForFunction(AccordCommandStore commandStore, ExecutionContext loadCtx, Function<? super SafeCommandStore, R> function)
+        public ForFunction(AccordCommandStore commandStore, ExecutionContext context, Function<? super SafeCommandStore, R> function)
         {
-            super(commandStore, loadCtx);
+            super(commandStore, context);
             this.function = function;
         }
 
@@ -129,9 +127,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     {
         private final Consumer<? super SafeCommandStore> consumer;
 
-        private ForConsumer(AccordCommandStore commandStore, ExecutionContext loadCtx, Consumer<? super SafeCommandStore> consumer)
+        private ForConsumer(AccordCommandStore commandStore, ExecutionContext context, Consumer<? super SafeCommandStore> consumer)
         {
-            super(commandStore, loadCtx);
+            super(commandStore, context);
             this.consumer = consumer;
         }
 
@@ -143,63 +141,16 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         }
     }
 
-    public static <T> AccordTask<T> create(CommandStore commandStore, ExecutionContext ctx, Function<? super SafeCommandStore, T> function)
+    public static <T> AccordTask<T> create(CommandStore commandStore, ExecutionContext context, Function<? super SafeCommandStore, T> function)
     {
-        return new ForFunction<>((AccordCommandStore) commandStore, ctx, function);
+        return new ForFunction<>((AccordCommandStore) commandStore, context, function);
     }
 
-    public static AccordTask<Void> create(CommandStore commandStore, ExecutionContext ctx, Consumer<? super SafeCommandStore> consumer)
+    public static AccordTask<Void> create(CommandStore commandStore, ExecutionContext context, Consumer<? super SafeCommandStore> consumer)
     {
-        return new ForConsumer((AccordCommandStore) commandStore, ctx, consumer);
+        return new ForConsumer((AccordCommandStore) commandStore, context, consumer);
     }
 
-    public enum State
-    {
-        INITIALIZED(),
-        WAITING_TO_SCAN_RANGES(INITIALIZED),
-        SCANNING_RANGES(WAITING_TO_SCAN_RANGES),
-        WAITING_TO_LOAD(INITIALIZED, SCANNING_RANGES),
-        LOADING(INITIALIZED, SCANNING_RANGES, WAITING_TO_LOAD),
-        WAITING_TO_RUN(INITIALIZED, SCANNING_RANGES, WAITING_TO_LOAD, LOADING),
-        ASSIGNED(WAITING_TO_RUN),
-        RUNNING(ASSIGNED),
-        PERSISTING(RUNNING),
-        FINISHED(RUNNING, PERSISTING),
-        CANCELLED(WAITING_TO_SCAN_RANGES, SCANNING_RANGES, WAITING_TO_LOAD, LOADING, WAITING_TO_RUN, ASSIGNED),
-        FAILED(WAITING_TO_SCAN_RANGES, SCANNING_RANGES, WAITING_TO_LOAD, LOADING, WAITING_TO_RUN, ASSIGNED, RUNNING, PERSISTING);
-
-        private final int permittedFrom;
-
-        State()
-        {
-            this.permittedFrom = 0;
-        }
-
-        State(State ... permittedFroms)
-        {
-            int permittedFrom = 0;
-            for (State state : permittedFroms)
-                permittedFrom |= 1 << state.ordinal();
-            this.permittedFrom = permittedFrom;
-        }
-
-        boolean isPermittedFrom(State prev)
-        {
-            return (permittedFrom & (1 << prev.ordinal())) != 0;
-        }
-
-        boolean isExecuted()
-        {
-            return this.compareTo(PERSISTING) >= 0;
-        }
-
-        boolean isComplete()
-        {
-            return this.compareTo(FINISHED) >= 0;
-        }
-    }
-
-    private State state = INITIALIZED;
     final AccordCommandStore commandStore;
     private final ExecutionContext executionContext;
     private volatile String loggingId;
@@ -214,14 +165,13 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     // TODO (desired): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
     @Nullable ArrayDeque<AccordCacheEntry<?, ?>> waitingToLoad;
     @Nullable RangeTxnScanner rangeScanner;
-    boolean hasRanges;
     @Nullable CommandSummaries commandsForRanges;
-    @Nullable private TaskQueue queued;
 
     private BiConsumer<? super R, Throwable> callback;
 
     public AccordTask(@Nonnull AccordCommandStore commandStore, ExecutionContext executionContext)
     {
+        super(executionContext);
         this.commandStore = commandStore;
         this.executionContext = executionContext;
         this.loggingId = "0x" + Long.toHexString(nextLoggingId.incrementAndGet());
@@ -250,13 +200,13 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
 
     public String toBriefString()
     {
-        return '{' + loggingId() + ',' + state + '}';
+        return '{' + loggingId() + ',' + state() + '}';
     }
 
     public String toDescription()
     {
         return toBriefString() + ": "
-               + (queued == null ? "unqueued" : queued.kind)
+               + (queued() == null ? "unqueued" : state())
                + ", primaryTxnId: " + executionContext.primaryTxnId()
                + ", waitingToLoad: " + summarise(waitingToLoad)
                + ", loading:" + summarise(loading, AccordSafeState::global)
@@ -302,17 +252,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         return out.toString();
     }
 
-    private void state(State state)
-    {
-        Invariants.require(state.isPermittedFrom(this.state), "%s forbidden from %s", state, this, AccordTask::toDescription);
-        this.state = state;
-        if (state == WAITING_TO_RUN)
-        {
-            Invariants.require(rangeScanner == null || rangeScanner.scanned);
-            Invariants.require(loading == null && waitingToLoad == null, "WAITING_TO_RUN => no loading or waiting; found %s", this, AccordTask::toDescription);
-        }
-    }
-
     Unseekables<?> keys()
     {
         return executionContext.keys();
@@ -334,20 +273,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         };
     }
 
-    public AsyncChain<R> priorityChain()
-    {
-        return new AsyncChains.Head<>()
-        {
-            @Override
-            protected Cancellable start(BiConsumer<? super R, Throwable> callback)
-            {
-                preSetup(callback);
-                commandStore.executor().submitPriority(AccordTask.this);
-                return AccordTask.this;
-            }
-        };
-    }
-
     private void preSetup(BiConsumer<? super R, Throwable> callback)
     {
         Invariants.require(this.callback == null);
@@ -358,7 +283,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     // to be invoked only by the CommandStore owning thread, to take references to objects already in use by the current execution
     public void presetup(AccordTask<?> parent)
     {
-        this.queuePosition = parent.queuePosition;
+        this.position = parent.position;
         // note we use the caches "unsafely" here deliberately, as we only reference commands we already have references to
         // so we do not mutate anything, except the atomic counter of references
         if (parent.commands != null)
@@ -393,8 +318,8 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     public void setupExclusive()
     {
         setupInternal(commandStore.cachesExclusive());
-        state(rangeScanner != null ? WAITING_TO_SCAN_RANGES
-                                   : waitingToLoad != null ? State.WAITING_TO_LOAD
+        setState(rangeScanner != null ? WAITING_TO_SCAN_RANGES
+                                      : waitingToLoad != null ? WAITING_TO_LOAD
                                     : loading != null ? LOADING : WAITING_TO_RUN);
     }
 
@@ -455,7 +380,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
                 throw new AssertionError("Incremental mode should only be used with an explicit list of keys");
 
             case SYNC:
-                hasRanges = true;
                 rangeScanner = new RangeTxnAndKeyScanner(caches.commandsForKeys());
         }
     }
@@ -535,11 +459,11 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
             return false;
 
         loading = null;
-        if (this.state.compareTo(State.WAITING_TO_LOAD) < 0)
+        if (compareTo(WAITING_TO_LOAD) < 0)
             return false;
 
         Invariants.require(waitingToLoad == null, "Invalid state: %s", this, AccordTask::toDescription);
-        state(WAITING_TO_RUN);
+        setState(WAITING_TO_RUN);
         return true;
     }
 
@@ -557,14 +481,14 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     private boolean onEmptyWaitingToLoad()
     {
         waitingToLoad = null;
-        if (this.state.compareTo(State.WAITING_TO_LOAD) < 0)
+        if (compareTo(WAITING_TO_LOAD) < 0)
             return false;
 
-        state(loading == null ? WAITING_TO_RUN : LOADING);
+        setState(loading == null ? WAITING_TO_RUN : LOADING);
         return true;
     }
 
-    public ExecutionContext preLoadContext()
+    public ExecutionContext executionContext()
     {
         return executionContext;
     }
@@ -602,7 +526,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
 
     private ArrayDeque<AccordCacheEntry<?, ?>> ensureWaitingToLoad()
     {
-        Invariants.require(state.compareTo(WAITING_TO_LOAD) <= 0, "Expected status to be on or before WAITING_TO_LOAD; found %s", this, AccordTask::toDescription);
+        Invariants.require(compareTo(WAITING_TO_LOAD) <= 0, "Expected status to be on or before WAITING_TO_LOAD; found %s", this, AccordTask::toDescription);
         if (waitingToLoad == null)
             waitingToLoad = new ArrayDeque<>();
         return waitingToLoad;
@@ -610,7 +534,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
 
     public AccordCacheEntry<?, ?> pollWaitingToLoad()
     {
-        Invariants.require(state == State.WAITING_TO_LOAD, "Expected status to be WAITING_TO_LOAD; found %s", this, AccordTask::toDescription);
+        Invariants.require(is(WAITING_TO_LOAD), "Expected status to be WAITING_TO_LOAD; found %s", this, AccordTask::toDescription);
         if (waitingToLoad == null)
             return null;
 
@@ -658,8 +582,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     @Override
     protected void preRunExclusive()
     {
-        state(ASSIGNED);
-        queued = null;
+        setState(ASSIGNED);
         if (rangeScanner != null)
         {
             commandsForRanges = rangeScanner.finish(commandStore.cachesExclusive());
@@ -675,14 +598,13 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     public void run()
     {
         onRunning();
-        logger.trace("Running {} with state {}", this, state);
         AccordSafeCommandStore safeStore = null;
         try (Closeable close = resources.get())
         {
             if (Tracing.isTracing())
                 Tracing.trace(executionContext.describe());
 
-            state(RUNNING);
+            setState(RUNNING);
 
             safeStore = commandStore.begin(this, commandsForRanges);
             R result = apply(safeStore);
@@ -710,7 +632,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
             boolean flush = changes != null || safeStore.fieldUpdates() != null;
             if (flush)
             {
-                state(PERSISTING);
+                setState(PERSISTING);
                 Runnable onFlush = () -> finish(result, null);
                 safeStore.persistFieldUpdatesInternal(changes == null ? onFlush : null);
                 if (changes != null) save(changes, onFlush);
@@ -735,12 +657,12 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
 
     public void fail(Throwable throwable)
     {
-        if (state.isComplete())
+        if (state().isComplete())
             return;
 
         try
         {
-            state(FAILED);
+            setState(FAILED);
             commandStore.agent().onException(throwable);
         }
         finally
@@ -750,6 +672,12 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         }
     }
 
+    @Override
+    protected boolean isNewWork()
+    {
+        return true;
+    }
+
     public void failExclusive(Throwable throwable)
     {
         fail(throwable);
@@ -758,7 +686,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
     @Override
     protected void cleanupExclusive(AccordExecutor executor)
     {
-        Invariants.expect(state.isExecuted());
+        Invariants.expect(state().isExecuted());
         releaseResources(commandStore.cachesExclusive());
         super.cleanupExclusive(executor);
         executor.keys.increment(commandsForKey == null ? 0 : commandsForKey.size(), runningAt);
@@ -775,22 +703,17 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         return rangeScanner;
     }
 
-    public boolean hasRanges()
-    {
-        return hasRanges;
-    }
-
     @Override
     public void cancel()
     {
-        if (!state.isComplete())
+        if (!state().isComplete())
             commandStore.executor().cancel(this);
     }
 
     void cancelExclusive()
     {
         logger.info("Cancelling {}", executionContext);
-        state(CANCELLED);
+        setState(CANCELLED);
         if (rangeScanner != null)
             rangeScanner.cancelled = true;
         if (callback != null)
@@ -805,14 +728,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         owner.cancelExclusive(this);
     }
 
-    public State state()
-    {
-        return state;
-    }
-
     private void finish(R result, Throwable failure)
     {
-        state(failure == null ? FINISHED : FAILED);
+        setState(failure == null ? FINISHED : FAILED);
         if (callback != null)
             callback.accept(result, failure);
     }
@@ -921,38 +839,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
             commandsForKey.forEach((k, v) -> v.revert());
     }
 
-    protected void addToQueue(TaskQueue queue)
-    {
-        if (state == CANCELLED)
-            return;
-
-        Invariants.require(queue.kind == state || (queue.kind == State.WAITING_TO_LOAD && state == WAITING_TO_SCAN_RANGES), "Invalid queue type: %s vs %s", queue.kind, this, AccordTask::toDescription);
-        Invariants.require(this.queued == null, "Already queued with state: %s", this, AccordTask::toDescription);
-        queued = queue;
-        queue.append(this);
-    }
-
-    @Nullable
-    TaskQueue<?> queued()
-    {
-        return queued;
-    }
-
-    TaskQueue<?> unqueue()
-    {
-        TaskQueue<?> wasQueued = queued;
-        queued.remove(this);
-        queued = null;
-        return wasQueued;
-    }
-
-    TaskQueue<?> unqueueIfQueued()
-    {
-        if (queued == null)
-            return null;
-        return unqueue();
-    }
-
     public class RangeTxnAndKeyScanner extends RangeTxnScanner
     {
         class KeyWatcher implements AccordCache.Listener<RoutingKey, CommandsForKey>
@@ -965,7 +851,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
             }
         }
 
-        // TODO (expected): produce key summaries to avoid locking all in memory
         final Set<TokenKey> intersectingKeys = new ObjectHashSet<>();
         final KeyWatcher keyWatcher = new KeyWatcher();
         final Ranges ranges = ((AbstractRanges) executionContext.keys()).toRanges();
@@ -1102,9 +987,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
         public void start(AccordExecutor executor)
         {
             Caches caches = commandStore.cachesExclusive();
-            state(SCANNING_RANGES);
+            setState(SCANNING_RANGES);
             startInternal(caches);
-            executor.submitPlainExclusive(AccordTask.this, this);
+            executor.submitPlainExclusive(AccordTask.this, GlobalGroup.RANGE_SCAN, this);
         }
 
         void startInternal(Caches caches)
@@ -1115,12 +1000,12 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
 
         public void scannedExclusive()
         {
-            Invariants.require(state == SCANNING_RANGES, "Expected SCANNING_RANGES; found %s", AccordTask.this, AccordTask::toDescription);
+            Invariants.require(is(SCANNING_RANGES), "Expected SCANNING_RANGES; found %s", AccordTask.this, AccordTask::toDescription);
             scanned = true;
             scannedInternal();
-            if (loading == null) state(WAITING_TO_RUN);
-            else if (waitingToLoad == null) state(LOADING);
-            else state(State.WAITING_TO_LOAD);
+            if (loading == null) setState(WAITING_TO_RUN);
+            else if (waitingToLoad == null) setState(LOADING);
+            else setState(WAITING_TO_LOAD);
         }
 
         void scannedInternal()

src/java/org/apache/cassandra/service/accord/debug/DebugExecution.java

@@ -131,11 +131,11 @@ public class DebugExecution
         }
     }
 
-    public static class DebugSequentialExecutor
+    public static class DebugExclusiveExecutor
     {
-        public static DebugSequentialExecutor maybeDebug(DebugExecutor owner, int commandStoreId)
+        public static DebugExclusiveExecutor maybeDebug(DebugExecutor owner, int commandStoreId)
         {
-            return DEBUG_EXECUTION ? new DebugSequentialExecutor(owner, commandStoreId) : null;
+            return DEBUG_EXECUTION ? new DebugExclusiveExecutor(owner, commandStoreId) : null;
         }
 
         final DebugExecutor owner;
@@ -144,7 +144,7 @@ public class DebugExecution
         long setTaskAt, waitingAt;
         Task prev;
 
-        public DebugSequentialExecutor(DebugExecutor owner, int commandStoreId)
+        public DebugExclusiveExecutor(DebugExecutor owner, int commandStoreId)
         {
             this.owner = owner;
             this.commandStoreId = commandStoreId;

src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java

@@ -23,6 +23,7 @@ import java.util.function.BiConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Result;
 import accord.api.Update;
 import accord.coordinate.CoordinationAdapter;
@@ -30,7 +31,6 @@ import accord.coordinate.CoordinationAdapter.Adapters.TxnAdapter;
 import accord.coordinate.ExecuteFlag.CoordinationFlags;
 import accord.coordinate.ExecutePath;
 import accord.local.Node;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.messages.Apply;
 import accord.primitives.Ballot;
 import accord.primitives.Deps;

src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java

@@ -28,12 +28,12 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.function.BiConsumer;
 
 import accord.api.Data;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Result;
 import accord.coordinate.CoordinationAdapter;
 import accord.coordinate.ExecuteFlag.CoordinationFlags;
 import accord.local.Node;
 import accord.local.Node.Id;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.messages.Commit;
 import accord.messages.Commit.Kind;
 import accord.primitives.AbstractRanges;

src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java

@@ -20,13 +20,13 @@ package org.apache.cassandra.service.accord.interop;
 
 import java.util.function.BiConsumer;
 
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Result;
 import accord.coordinate.ExecuteFlag;
 import accord.coordinate.Persist;
 import accord.coordinate.tracking.AllTracker;
 import accord.coordinate.tracking.QuorumTracker;
 import accord.local.Node;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.messages.Apply;
 import accord.primitives.Ballot;
 import accord.primitives.Deps;

src/java/org/apache/cassandra/utils/concurrent/SignalLock.java

@@ -424,12 +424,6 @@ public final class SignalLock implements Lock
 
     public boolean incrementAsyncWork(boolean signalIfWaiting)
     {
-        return incrementAsyncWork(signalIfWaiting, 1);
-    }
-
-    public boolean incrementAsyncWork(boolean signalIfWaiting, int increment)
-    {
-        Invariants.require(increment >= 1 && increment <= MAX_SIGNAL_COUNT);
         while (true)
         {
             long cur = state;

test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java

@@ -401,10 +401,13 @@ public class ActionSchedule implements CloseableIterator<Object>, LongConsumer
                 advance(scheduled.poll());
         }
 
-        Action perform = runnable.poll();
+        Action perform = runnableByDeadline.peek();
         if (perform == null)
             throw new NoSuchElementException();
 
+        if (perform.deadline() < now - currentJitter) runnable.remove(perform);
+        else perform = runnable.poll();
+
         if (!runnableByDeadline.remove(perform) && perform.deadline() > 0)
             throw new IllegalStateException();
         time.tick(perform.deadline());

test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java

@@ -21,6 +21,8 @@ package org.apache.cassandra.simulator.test;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -28,14 +30,19 @@ import java.util.concurrent.Future;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
+
+import javax.annotation.Nullable;
 
 import org.junit.Test;
 
 import accord.api.AsyncExecutor;
-import accord.local.ExclusiveAsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
+import accord.utils.async.Cancellable;
 
 import org.apache.cassandra.concurrent.ExecutorPlus;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -44,37 +51,144 @@ import org.apache.cassandra.service.accord.AccordExecutor;
 import org.apache.cassandra.service.accord.AccordExecutorAsyncSubmit;
 import org.apache.cassandra.service.accord.AccordExecutorSignalLoop;
 import org.apache.cassandra.service.accord.api.AccordAgent;
+import org.apache.cassandra.utils.concurrent.AsyncPromise;
+import org.apache.cassandra.utils.concurrent.SignalLock;
 
 import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
 import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITHOUT_LOCK;
-import static org.apache.cassandra.service.accord.AccordService.toFuture;
 
 // TODO (required): have simulator intercept ReentantLock so we can test SyncSubmit and SemiSyncSubmit
 public class AccordExecutorTest extends SimulationTestBase
 {
-    static final int EXECUTOR_THREAD_COUNT = 44;
-
     @Test
     public void signalLoopTest()
     {
-        executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, -1, -1, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent()),
+        executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, SignalLock.MAX_THREADS, -1, -1, TimeUnit.MICROSECONDS, i -> "Loop" + i, new AccordAgent()),
                      16);
     }
 
     @Test
     public void signalSpinLoopTest()
     {
-        executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, 10, 100, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent()),
+        executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, SignalLock.MAX_THREADS, 10, 100, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent()),
                      16);
     }
 
     @Test
     public void ayncSubmitTest()
     {
-        executorTest(() -> new AccordExecutorAsyncSubmit(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, i -> "Loop" + i, new AccordAgent()),
+        int threads = 32;
+        executorTest(() -> new AccordExecutorAsyncSubmit(1, RUN_WITHOUT_LOCK, threads, i -> "Loop" + i, new AccordAgent()),
                      16);
     }
 
+    static class Submitted
+    {
+        final AtomicInteger nextId = new AtomicInteger();
+        final AtomicInteger doneBefore = new AtomicInteger();
+        final ConcurrentHashMap<Integer, Collection<Future<?>>> consequences = new ConcurrentHashMap<>();
+
+        boolean isDone()
+        {
+            return isDoneBetween(0, nextId.get());
+        }
+
+        boolean isDoneBefore(int before)
+        {
+            if (!isDoneBetween(doneBefore.get(), before))
+                return false;
+            doneBefore.accumulateAndGet(before, Integer::max);
+            return true;
+        }
+
+        boolean isDoneBetween(int from, int before)
+        {
+            for (int id = from ; id < before ; ++id)
+            {
+                for (Future<?> future : consequences.get(id))
+                {
+                    if (!future.isDone())
+                        return false;
+                }
+            }
+            return true;
+        }
+
+        Collection<Future<?>> start()
+        {
+            ConcurrentLinkedQueue<Future<?>> result = new ConcurrentLinkedQueue<>();
+
+            while (true)
+            {
+                int id = nextId.get();
+                Object prev = consequences.putIfAbsent(id, result);
+                nextId.compareAndSet(id, id + 1);
+                if (prev == null)
+                    return result;
+            }
+        }
+    }
+
+    static class Control extends ConcurrentLinkedQueue<Cancellable>
+    {
+        final Submitted submitted;
+        final AtomicInteger count = new AtomicInteger();
+        final float cancelChance;
+        float processChance;
+
+        Control(float cancelChance, Submitted submitted)
+        {
+            this(submitted, cancelChance, ThreadLocalRandom.current().nextFloat() * 0.5f);
+        }
+
+        Control(Submitted submitted, float cancelChance, float processChance)
+        {
+            this.submitted = submitted;
+            this.cancelChance = cancelChance;
+            this.processChance = processChance;
+        }
+
+        void submit(AsyncExecutor executor, Collection<Future<?>> consequences, Consumer<Collection<Future<?>>> run)
+        {
+            AsyncPromise<?> future = new AsyncPromise<>();
+            Cancellable cancel = executor.chain(() -> run.accept(consequences)).begin((success, fail) -> {
+                if (fail == null) future.trySuccess(null);
+                else
+                {
+                    future.tryFailure(fail);
+                    if (fail instanceof CancellationException)
+                        run.accept(submitted.start());
+                }
+            });
+            consequences.add(future);
+
+            if (cancel != null && ThreadLocalRandom.current().nextFloat() <= cancelChance)
+            {
+                add(cancel);
+                count.incrementAndGet();
+            }
+
+            if (count.get() > 0 && ThreadLocalRandom.current().nextFloat() <= processChance)
+            {
+                int cancelCount = 0;
+                do
+                {
+                    ++cancelCount;
+
+                    float delta = ThreadLocalRandom.current().nextFloat() - 0.5f;
+                    if (delta < 0) processChance /= delta;
+                    else processChance *= -delta;
+                    if (processChance < 0.001f || processChance > 0.999f)
+                        processChance = cancelChance;
+                } while (count.decrementAndGet() > 0 && ThreadLocalRandom.current().nextFloat() <= processChance);
+
+                // do outside of loop to avoid reentry
+                while (cancelCount-- > 0)
+                    remove().cancel();
+            }
+        }
+    }
+
     public void executorTest(SerializableSupplier<AccordExecutor> supplier, int submissionThreads)
     {
         simulate(arr(() -> {
@@ -84,30 +198,38 @@ public class AccordExecutorTest extends SimulationTestBase
                          ExecutorPlus submit = executorFactory().pooled("submit-test", submissionThreads);
                          AccordExecutor executor = supplier.get();
                          Lock lock = executor.unsafeLock();
-                         ExclusiveAsyncExecutor sequentialExecutor = executor.newSequentialExecutor();
+                         ExclusiveAsyncExecutor sequentialExecutor = executor.newExclusiveExecutor();
                          Executor lockExecutor = executorFactory().sequential("lock");
 
                          for (float sleepChance : new float[] { 0f, 0.01f, 0.1f })
                          {
                              for (float lockChance : new float[] { 0f, 0.01f, 0.1f })
                              {
-                                 List<Future<?>> done = new ArrayList<>();
-                                 for (int i = 0 ; i < submissionThreads ; ++i)
+                                 for (float cancelChance : new float[] { 0f, 0.01f, 0.1f })
                                  {
-                                     int id = i;
-                                     done.add(submit.submit(() -> {
-                                         try
-                                         {
-                                             submitLoop(id, lock, executor, sequentialExecutor, lockExecutor, 20, 10, sleepChance, lockChance);
-                                         }
-                                         catch (ExecutionException | InterruptedException e)
-                                         {
-                                             throw new RuntimeException(e);
-                                         }
-                                     }));
+                                     System.out.println(String.format("sleepChance %.2f, lockChance %.2f, cancelChance %.2f", sleepChance, lockChance, cancelChance));
+                                     List<Future<?>> done = new ArrayList<>();
+                                     Submitted submitted = new Submitted();
+                                     for (int i = 0; i < submissionThreads; ++i)
+                                     {
+                                         int id = i;
+                                         done.add(submit.submit(() -> {
+                                             try
+                                             {
+                                                 submitLoop(id, lock, executor, sequentialExecutor, lockExecutor, 20, 10, sleepChance, lockChance, new Control(cancelChance, submitted), submitted);
+                                             }
+                                             catch (ExecutionException | InterruptedException e)
+                                             {
+                                                 throw new RuntimeException(e);
+                                             }
+                                         }));
+                                     }
+                                     for (Future<?> f : done)
+                                         f.get();
+
+                                     if (!submitted.isDone())
+                                         throw new AssertionError();
                                  }
-                                 for (Future<?> f : done)
-                                     f.get();
                              }
                          }
                      }
@@ -119,27 +241,34 @@ public class AccordExecutorTest extends SimulationTestBase
                  () -> {}, 1L);
     }
 
-    private static void submitLoop(int id, Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, Executor lockExecutor, int outerLoop, int innerLoop, float sleepChance, float lockChance) throws ExecutionException, InterruptedException
+    private static void submitLoop(int id, Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, Executor lockExecutor, int outerLoop, int innerLoop, float sleepChance, float lockChance, Control control, Submitted submitted) throws ExecutionException, InterruptedException
     {
-        ConcurrentLinkedQueue<Future<?>> await = new ConcurrentLinkedQueue<>();
+        ConcurrentLinkedQueue<Future<?>> awaitConsequences = new ConcurrentLinkedQueue<>();
         while (outerLoop-- > 0)
         {
+            List<Collection<Future<?>>> allAwaitSubmitted = new ArrayList<>();
             for (int i = 0; i < innerLoop; ++i)
-                submitRecursive(lock, executor, sequentialExecutor, 1 + i, await, sleepChance, lockChance);
+            {
+                Collection<Future<?>> awaitSubmitted = submitted.start();
+                allAwaitSubmitted.add(awaitSubmitted);
+                submitRecursive(lock, executor, sequentialExecutor, 1 + i, awaitSubmitted, awaitConsequences, submitted, sleepChance, lockChance, control);
+            }
 
             AtomicBoolean done = new AtomicBoolean();
             submitUntil(lock, lockExecutor, sleepChance, done::get);
-            while (!await.isEmpty())
-                await.poll().get();
+            for (Collection<Future<?>> awaitSubmitted : allAwaitSubmitted)
+                await(awaitSubmitted, CancellationException.class);
+            await(awaitConsequences, null);
             done.set(true);
             System.out.println("Loop " + id + '.' + (1 + outerLoop));
         }
     }
 
-    private static void submitRecursive(Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, int count, Collection<Future<?>> await, float sleepChance, float lockChance)
+    private static void submitRecursive(Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, int count, Collection<Future<?>> consequences, Collection<Future<?>> awaitConsequences, Submitted submitted, float sleepChance, float lockChance, Control control)
     {
         AsyncExecutor submitTo = ThreadLocalRandom.current().nextBoolean() ? executor : sequentialExecutor;
-        await.add(toFuture(submitTo.chain(() -> {
+
+        control.submit(submitTo, consequences, nextConsequences -> {
             ThreadLocalRandom rnd = ThreadLocalRandom.current();
             boolean locked = false;
             if (rnd.nextFloat() < lockChance)
@@ -147,10 +276,21 @@ public class AccordExecutorTest extends SimulationTestBase
                 if (rnd.nextBoolean()) locked = lock.tryLock();
                 else { locked = true; lock.lock(); }
             }
+            if (ThreadLocalRandom.current().nextFloat() < 0.01f)
+            {
+                int expectDoneBefore = submitted.nextId.get();
+                AsyncPromise<Void> afterConsequences = new AsyncPromise<>();
+                executor.afterSubmittedAndConsequences(() -> {
+                    if (!submitted.isDoneBefore(expectDoneBefore))
+                        throw new AssertionError();
+                    afterConsequences.setSuccess(null);
+                });
+                awaitConsequences.add(afterConsequences);
+            }
             try
             {
                 if (count > 1)
-                    submitRecursive(lock, executor, sequentialExecutor, count -1, await, sleepChance, lockChance);
+                    submitRecursive(lock, executor, sequentialExecutor, count -1, nextConsequences, awaitConsequences, submitted, sleepChance, lockChance, control);
                 if (rnd.nextFloat() < sleepChance)
                     LockSupport.parkNanos(rnd.nextInt(10000, 100000));
             }
@@ -159,7 +299,7 @@ public class AccordExecutorTest extends SimulationTestBase
                 if (locked)
                     lock.unlock();
             }
-        }).beginAsResult()));
+        });
     }
 
     private static void submitUntil(Lock lock, Executor executor, float sleepChance, BooleanSupplier done)
@@ -188,4 +328,17 @@ public class AccordExecutorTest extends SimulationTestBase
             }
         });
     }
+
+    private static void await(Collection<Future<?>> await, @Nullable Class<? extends Throwable> ignore) throws InterruptedException, ExecutionException
+    {
+        for (Future<?> future : await)
+        {
+            try { future.get(); }
+            catch (ExecutionException e)
+            {
+                if (ignore == null || !(ignore.isInstance(e.getCause())))
+                    throw e;
+            }
+        }
+    }
 }

test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java

@@ -90,9 +90,9 @@ import org.apache.cassandra.service.accord.IAccordService;
 import org.apache.cassandra.service.accord.api.TokenKey;
 import org.apache.cassandra.utils.FBUtilities;
 
+import static accord.local.ExecutionContext.contextFor;
 import static accord.local.LoadKeys.SYNC;
 import static accord.local.LoadKeysFor.READ_WRITE;
-import static accord.local.ExecutionContext.contextFor;
 import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_DURABLE;
 import static accord.primitives.Routable.Domain.Range;
 import static accord.primitives.Timestamp.Flag.HLC_BOUND;

test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java

@@ -27,9 +27,9 @@ import org.junit.Test;
 import accord.api.Key;
 import accord.api.RoutingKey;
 import accord.local.Command;
+import accord.local.ExecutionContext;
 import accord.local.LoadKeys;
 import accord.local.Node;
-import accord.local.ExecutionContext;
 import accord.local.SafeCommand;
 import accord.local.SafeCommandStore;
 import accord.local.StoreParticipants;

test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java

@@ -85,9 +85,9 @@ import org.apache.cassandra.utils.AssertionUtils;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.Condition;
 
+import static accord.local.ExecutionContext.contextFor;
 import static accord.local.LoadKeys.SYNC;
 import static accord.local.LoadKeysFor.READ_WRITE;
-import static accord.local.ExecutionContext.contextFor;
 import static accord.utils.Property.qt;
 import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
 import static org.apache.cassandra.service.accord.AccordService.getBlocking;

test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java

@@ -37,6 +37,7 @@ import org.junit.Assert;
 
 import accord.api.AsyncExecutor;
 import accord.api.Data;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Journal;
 import accord.api.ProgressLog.NoOpProgressLog;
 import accord.api.RemoteListeners.NoOpRemoteListeners;
@@ -56,7 +57,6 @@ import accord.local.Node;
 import accord.local.Node.Id;
 import accord.local.NodeCommandStoreService;
 import accord.local.SafeCommandStore;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.local.StoreParticipants;
 import accord.local.TimeService;
 import accord.local.durability.DurabilityService;
@@ -132,7 +132,7 @@ public class AccordTestUtils
 
     public static class Commands
     {
-        public static Command notDefined(TxnId txnId, PartialTxn txn)
+        public static Command notDefined(TxnId txnId)
         {
             return Command.NotDefined.notDefined(txnId, NotDefined, NotDurable, StoreParticipants.empty(txnId), Ballot.ZERO);
         }
@@ -371,7 +371,7 @@ public class AccordTestUtils
             }
 
             @Override
-            public ExclusiveAsyncExecutor someSequentialExecutor()
+            public ExclusiveAsyncExecutor someExclusiveExecutor()
             {
                 return null;
             }

test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java

@@ -35,6 +35,7 @@ import javax.annotation.Nullable;
 import org.assertj.core.api.Assertions;
 
 import accord.api.AsyncExecutor;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Journal;
 import accord.api.LocalListeners;
 import accord.api.ProgressLog;
@@ -59,7 +60,6 @@ import accord.local.Node;
 import accord.local.NodeCommandStoreService;
 import accord.local.SafeCommand;
 import accord.local.SafeCommandStore;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.local.TimeService;
 import accord.local.durability.DurabilityService;
 import accord.messages.BeginRecovery;
@@ -182,7 +182,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
             }
 
             @Override
-            public ExclusiveAsyncExecutor someSequentialExecutor()
+            public ExclusiveAsyncExecutor someExclusiveExecutor()
             {
                 return null;
             }

test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java

@@ -49,6 +49,7 @@ import org.junit.Test;
 import accord.api.Agent;
 import accord.api.AsyncExecutor;
 import accord.api.DataStore;
+import accord.api.ExclusiveAsyncExecutor;
 import accord.api.Journal;
 import accord.api.Key;
 import accord.api.OwnershipEventListener;
@@ -72,7 +73,6 @@ import accord.local.NodeCommandStoreService;
 import accord.local.RedundantBefore;
 import accord.local.SafeCommand;
 import accord.local.SafeCommandStore;
-import accord.local.ExclusiveAsyncExecutor;
 import accord.local.StoreParticipants;
 import accord.local.TimeService;
 import accord.local.cfk.CommandsForKey;
@@ -718,7 +718,7 @@ public class CommandsForKeySerializerTest
         @Override public NodeCommandStoreService node() { return new NodeCommandStoreService()
         {
             @Override public AsyncExecutor someExecutor() { throw new UnsupportedOperationException(); }
-            @Override public ExclusiveAsyncExecutor someSequentialExecutor() { throw new UnsupportedOperationException(); }
+            @Override public ExclusiveAsyncExecutor someExclusiveExecutor() { throw new UnsupportedOperationException(); }
             @Override public long epoch() { return 0;}
             @Override public Node.Id id() { return Node.Id.NONE; }
             @Override public Timeouts timeouts() { return null; }

test/unit/org/apache/cassandra/utils/AccordGenerators.java

@@ -175,7 +175,7 @@ public class AccordGenerators
             switch (targetType)
             {
                 case notDefined:
-                    return AccordTestUtils.Commands.notDefined(id, txn);
+                    return AccordTestUtils.Commands.notDefined(id);
                 case preaccepted:
                     return AccordTestUtils.Commands.preaccepted(id, txn, executeAt);
                 case committed: