From 1cffe9a8a032ce37889e821fb46b165bcd942323 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 9 Jul 2026 11:42:47 +0100 Subject: [PATCH] Executor Fairness --- .../cassandra/concurrent/CassandraThread.java | 57 +- .../apache/cassandra/config/AccordConfig.java | 80 +- .../db/virtual/AccordDebugKeyspace.java | 2 +- .../cassandra/service/accord/AccordCache.java | 1 - .../service/accord/AccordCommandStore.java | 40 +- .../service/accord/AccordCommandStores.java | 6 +- .../service/accord/AccordExecutor.java | 2629 +++++++++++++---- .../AccordExecutorAbstractLockLoop.java | 110 +- .../accord/AccordExecutorAbstractLoop.java | 22 +- .../accord/AccordExecutorSemiSyncSubmit.java | 12 - .../accord/AccordExecutorSignalLoop.java | 446 +-- .../service/accord/AccordExecutorSimple.java | 12 + .../accord/AccordExecutorSyncSubmit.java | 5 +- .../accord/AccordFetchCoordinator.java | 2 +- .../accord/AccordSafeCommandStore.java | 2 +- .../service/accord/AccordService.java | 2 +- .../cassandra/service/accord/AccordTask.java | 221 +- .../service/accord/debug/DebugExecution.java | 8 +- .../accord/interop/AccordInteropAdapter.java | 2 +- .../interop/AccordInteropExecution.java | 2 +- .../accord/interop/AccordInteropPersist.java | 2 +- .../utils/concurrent/SignalLock.java | 6 - .../cassandra/simulator/ActionSchedule.java | 5 +- .../simulator/test/AccordExecutorTest.java | 217 +- .../CompactionAccordIteratorsTest.java | 2 +- .../service/accord/AccordCommandTest.java | 2 +- .../service/accord/AccordTaskTest.java | 2 +- .../service/accord/AccordTestUtils.java | 6 +- .../accord/SimulatedAccordCommandStore.java | 4 +- .../CommandsForKeySerializerTest.java | 4 +- .../cassandra/utils/AccordGenerators.java | 2 +- 31 files changed, 2782 insertions(+), 1131 deletions(-) diff --git a/src/java/org/apache/cassandra/concurrent/CassandraThread.java b/src/java/org/apache/cassandra/concurrent/CassandraThread.java index 62cbcdb6a4..d66adc2442 100644 --- a/src/java/org/apache/cassandra/concurrent/CassandraThread.java +++ b/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 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() { diff --git a/src/java/org/apache/cassandra/config/AccordConfig.java b/src/java/org/apache/cassandra/config/AccordConfig.java index b9fe93b9a4..fe226fe92f 100644 --- a/src/java/org/apache/cassandra/config/AccordConfig.java +++ b/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 */ diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 35b11418a6..86238688f5 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/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); }); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java index 6b6c48b2fc..0d679b2f51 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java @@ -345,7 +345,6 @@ public class AccordCache implements CacheSize Collection> load(LoadExecutor loadExecutor, P1 p1, P2 p2, AccordCacheEntry node) { - Type parent = node.owner.parent(); return node.load(loadExecutor, p1, p2).waiters(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 7bdb69efc0..60205bf890 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/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 { - private final Lock lock; + private final AccordExecutor owner; - public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type.Instance commands, AccordCache.Type.Instance commandsForKeys) + public ExclusiveCaches(AccordExecutor owner, AccordCache global, AccordCache.Type.Instance commands, AccordCache.Type.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 AsyncChain chain(ExecutionContext loadCtx, Function function) + public AsyncChain chain(ExecutionContext context, Function function) { - return AccordTask.create(this, loadCtx, function).chain(); + return AccordTask.create(this, context, function).chain(); } @Override - public AsyncChain chain(ExecutionContext executionContext, Consumer consumer) + public AsyncChain chain(ExecutionContext context, Consumer consumer) { - return AccordTask.create(this, executionContext, consumer).chain(); - } - - @Override - public AsyncChain priorityChain(ExecutionContext executionContext, Consumer consumer) - { - return AccordTask.create(this, executionContext, consumer).priorityChain(); - } - - @Override - public AsyncChain priorityChain(ExecutionContext executionContext, Function function) - { - return AccordTask.create(this, executionContext, function).priorityChain(); + return AccordTask.create(this, context, consumer).chain(); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index 929d3d58cb..f82215919e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/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) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index 4277ec55c8..c600178513 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.List; -import java.util.Queue; +import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.CancellationException; import java.util.concurrent.RejectedExecutionException; @@ -41,6 +41,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.Agent; +import accord.api.ExclusiveAsyncExecutor; import accord.api.RoutingKey; import accord.impl.AbstractAsyncExecutor; import accord.local.Command; @@ -54,28 +55,31 @@ import accord.messages.Request; import accord.primitives.Ballot; import accord.primitives.SaveStatus; import accord.primitives.TxnId; -import accord.utils.ArrayBuffers.BufferList; +import accord.utils.ArrayBuffers; +import accord.utils.IntrusiveHeapNode; import accord.utils.IntrusivePriorityHeap; import accord.utils.Invariants; import accord.utils.QuadConsumer; import accord.utils.QuadFunction; import accord.utils.QuintConsumer; +import accord.utils.TinyEnumSet; import accord.utils.TriConsumer; import accord.utils.TriFunction; import accord.utils.UnhandledEnum; import accord.utils.async.AsyncCallbacks.CallAndCallback; -import accord.utils.async.AsyncCallbacks.FlatCallAndCallback; -import accord.utils.async.AsyncCallbacks.RunAndCallback; import accord.utils.async.AsyncCallbacks.RunOrFail; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import accord.utils.async.Cancellable; import org.apache.cassandra.cache.CacheSize; +import org.apache.cassandra.concurrent.CassandraThread; import org.apache.cassandra.concurrent.DebuggableTask; import org.apache.cassandra.concurrent.DebuggableTask.DebuggableTaskRunner; import org.apache.cassandra.concurrent.ExecutorLocals; import org.apache.cassandra.concurrent.Shutdownable; +import org.apache.cassandra.config.AccordConfig; +import org.apache.cassandra.config.AccordConfig.QueueBalancingModel; import org.apache.cassandra.config.AccordConfig.QueuePriorityModel; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.metrics.AccordCacheMetrics; @@ -89,29 +93,52 @@ import org.apache.cassandra.metrics.ShardedDecayingHistograms.DecayingHistograms import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor; import org.apache.cassandra.service.accord.AccordCacheEntry.SaveExecutor; import org.apache.cassandra.service.accord.AccordCacheEntry.UniqueSave; +import org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup; +import org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup; +import org.apache.cassandra.service.accord.AccordExecutor.Task.GroupKind; +import org.apache.cassandra.service.accord.debug.DebugExecution.DebugExclusiveExecutor; import org.apache.cassandra.service.accord.debug.DebugExecution.DebugExecutor; -import org.apache.cassandra.service.accord.debug.DebugExecution.DebugSequentialExecutor; import org.apache.cassandra.service.accord.debug.DebugExecution.DebugTask; import org.apache.cassandra.utils.Closeable; -import org.apache.cassandra.utils.MonotonicClock; import org.apache.cassandra.utils.WithResources; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Condition; import org.apache.cassandra.utils.concurrent.Future; -import io.netty.util.concurrent.FastThreadLocal; - +import static accord.primitives.Routable.Domain.Range; import static accord.utils.Invariants.createIllegalState; -import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.PHASE_HLC_FIFO; +import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.ORIG_HLC_FIFO; import static org.apache.cassandra.service.accord.AccordCache.CommandAdapter.COMMAND_ADAPTER; import static org.apache.cassandra.service.accord.AccordCache.CommandsForKeyAdapter.CFK_ADAPTER; import static org.apache.cassandra.service.accord.AccordCache.registerJfrListener; import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED; -import static org.apache.cassandra.service.accord.AccordTask.State.LOADING; -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.AccordExecutor.MultiTaskQueue.COUNTER_LOWBITS; +import static org.apache.cassandra.service.accord.AccordExecutor.MultiTaskQueue.COUNTER_MASKS; +import static org.apache.cassandra.service.accord.AccordExecutor.MultiTaskQueue.minCounterValue; +import static org.apache.cassandra.service.accord.AccordExecutor.MultiTaskQueue.selectByOverflowBits; +import static org.apache.cassandra.service.accord.AccordExecutor.MultiTaskQueue.setOverflowWhenLessEqual; +import static org.apache.cassandra.service.accord.AccordExecutor.RunnableTaskQueue.RUNNABLE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.ACCEPT; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.APPLY; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.COMMIT; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.PREACCEPT; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.RANGE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.ExclusiveGroup.STABLE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.COMMAND_STORE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.LOAD; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.OTHER; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.RANGE_LOAD; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.RANGE_SCAN; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.GlobalGroup.SAVE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.MAX_TRANCHE; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.ASSIGNED; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.INITIALIZED; +import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.LOADING; +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.utils.Clock.Global.nanoTime; @@ -123,11 +150,82 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutorflow. + private static final int BLEND_SHIFT = 6, BLEND_TOTAL = 1 << BLEND_SHIFT; + private static final int FLOW_ONSET, FLOW_WIDTH_SHIFT; + private static final boolean BALANCE_BY_POSITION; + private static final long GLOBAL_QUEUE_LIMITS, GLOBAL_QUEUE_BUDGETS; + private static final long EXCLUSIVE_QUEUE_LIMITS, EXCLUSIVE_QUEUE_BUDGETS; + + static + { + AccordConfig config = DatabaseDescriptor.getAccord(); + AGE_TO_FIFO = config.queue_priority_age_to_fifo.to(TimeUnit.MICROSECONDS); + PRIORITY_MODEL = config.queue_priority_model != null ? config.queue_priority_model : QueuePriorityModel.HLC_FIFO; + BALANCING_MODEL = config.queue_balancing_model != null ? config.queue_balancing_model : QueueBalancingModel.BLENDED_PRIORITY_PHASE_BUDGET_FAIR; + FLOW_ONSET = config.queue_flow_imbalance_onset == null ? 4 : config.queue_flow_imbalance_onset; + FLOW_WIDTH_SHIFT = config.queue_flow_imbalance_width_shift == null ? 5 : config.queue_flow_imbalance_width_shift; + Invariants.require(FLOW_ONSET >= 0 && FLOW_WIDTH_SHIFT >= 0); + switch (BALANCING_MODEL) + { + default: throw new UnhandledEnum(BALANCING_MODEL); + case PRIORITY_ONLY: + case BLENDED_PRIORITY_PHASE_FAIR: + case BLENDED_PRIORITY_PHASE_BUDGET_FAIR: + case PRIORITY_BUDGET: + BALANCE_BY_POSITION = true; + break; + case PHASE_ONLY: + case PHASE_FAIR: + case PHASE_BUDGET: + case PHASE_BUDGET_FAIR: + BALANCE_BY_POSITION = false; + } + + { + long global = COUNTER_MASKS, exclusive = COUNTER_MASKS; + global ^= (0x7f ^ 1) << RANGE_SCAN.ordinal(); + if (config.queue_active_limits != null) + { + long[] limits = parseEnumParams(config.queue_active_limits, "queue_active_limits"); + global = selectByOverflowBits(setOverflowWhenLessEqual(limits[0], 0), global, limits[0]); + exclusive = selectByOverflowBits(setOverflowWhenLessEqual(limits[1], 0), global, limits[1]); + } + GLOBAL_QUEUE_LIMITS = global; + EXCLUSIVE_QUEUE_LIMITS = exclusive; + } + + { + long global = encodeCounters(Map.of(COMMAND_STORE, 16L, LOAD, 16L, SAVE, 16L, OTHER, 16L, RANGE_LOAD, 2L, RANGE_SCAN, 1L)); + long exclusive = encodeCounters(Map.of(APPLY, 8L, STABLE, 7L, COMMIT, 6L, ACCEPT, 5L, OTHER, 4L, PREACCEPT, 4L, RANGE, 1L)); + if (config.queue_budgets != null) + { + long[] limits = parseEnumParams(config.queue_budgets, "queue_budgets"); + // any that are unset, set to the lowest value of any other specified + if (limits[0] != 0) + { + long min = minCounterValue(limits[0], 0); + long mins = min * COUNTER_LOWBITS; + if (min > 1) + { + mins ^= (min ^ 1) << (RANGE_SCAN.ordinal() * 8); // set the default RANGE_SCAN budget to 1 + mins ^= (min ^ 2) << (RANGE_LOAD.ordinal() * 8); // set the default RANGE_LOAD budget to 2 + } + global = selectByOverflowBits(setOverflowWhenLessEqual(limits[0], 0), mins, limits[0]); + } + if (limits[1] != 0) + exclusive = selectByOverflowBits(setOverflowWhenLessEqual(limits[1], 0), minCounterValue(limits[1], 0) * COUNTER_LOWBITS, limits[1]); + } + GLOBAL_QUEUE_BUDGETS = global; + EXCLUSIVE_QUEUE_BUDGETS = exclusive; + } + } + public static final ShardedDecayingHistograms HISTOGRAMS = new ShardedDecayingHistograms(); - private static final FastThreadLocal paranoidPriorityInversionCheck = new FastThreadLocal<>(); public interface AccordExecutorFactory { @@ -136,6 +234,92 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor threadLocal = ThreadLocal.withInitial(ThreadLocalAccordTaskRunner::new); + + @Override + public AccordExecutor accordActiveExecutor() + { + return activeExecutor; + } + + @Override + public void setAccordActiveExecutor(AccordExecutor newExecutor) + { + activeExecutor = newExecutor; + } + + @Override + public AccordExecutor accordLockedExecutor() + { + return lockedExecutor; + } + + @Override + public boolean trySetAccordLockedExecutor(AccordExecutor newLockedExecutor) + { + if (lockedExecutor != null) + return false; + lockedExecutor = newLockedExecutor; + return true; + } + + @Override + public void clearAccordLockedExecutor() + { + lockedExecutor = null; + } + + @Override + public void setAccordActiveTask(Task newActiveTask) + { + activeTask = newActiveTask; + } + + @Override + public Task accordActiveTask() + { + return activeTask; + } + + @Override + public Task accordActiveSelfTask() + { + return activeTask; + } + } + // WARNING: this is a shared object, so close is NOT idempotent public static final class ExclusiveGlobalCaches extends GlobalCaches implements AutoCloseable { @@ -152,7 +336,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor deferred; + + WithDeferred(Runnable run, Runnable deferred) + { + this.run = run; + this.deferred = new ArrayDeque<>(1); + this.deferred.add(deferred); + } + + WithDeferred(Runnable run, ArrayDeque deferred) + { + this.run = run; + this.deferred = deferred; + } + + @Override + public void run() + { + Runnable register = deferred.poll(); + if (!deferred.isEmpty()) + { + Invariants.require(runs[firstIndex] != null); + Invariants.require(!(runs[firstIndex] instanceof WithDeferred)); + runs[firstIndex] = new WithDeferred(runs[firstIndex], deferred); + } + registerWait(register); + try { run.run(); } + catch (Throwable t) { owner.agent.onException(t); } + } } - public String toString() + final AccordExecutor owner; + + int firstTranche; + int firstIndex; + long[] mins = new long[8]; + int[] counts = new int[8]; + Runnable[] runs = new Runnable[8]; + + // the next tranche, that all new work is being collected against + // (and will move to the array accounting once a new wait is registered) + long nextMin; + int nextTranche; + int nextCount; + + private Tranches(AccordExecutor owner) { - return run.toString() + " @" + position; + this.owner = owner; + } + + private int trancheToIndex(int tranche) + { + return firstIndex + trancheToIndexOffset(tranche); + } + + private int trancheToIndexOffset(int tranche) + { + int offset = tranche - firstTranche; + if (offset < 0) + offset += MAX_TRANCHE + 1; + return offset; + } + + int size() + { + return trancheToIndexOffset(nextTranche); + } + + int capacity() + { + return counts.length; + } + + int addNew(long position) + { + Invariants.require(position >= nextMin); + ++nextCount; + return nextTranche; + } + + void addInherited(int tranche, long position) + { + if (tranche == nextTranche) + { + Invariants.require(position >= nextMin); + ++nextCount; + } + else + { + int index = trancheToIndex(tranche); + Invariants.require(counts[index] > 0); + Invariants.require(mins[index] <= position); + ++counts[index]; + } + } + + void complete(int tranche) + { + if (tranche == nextTranche) + { + --nextCount; + } + else + { + + if (counts[trancheToIndex(tranche)] == 1) + owner.drainUnqueuedNewWorkExclusive(); // make sure we don't have any pending + + if (--counts[trancheToIndex(tranche)] == 0 && tranche == firstTranche) + { + do advance(); + while (firstTranche != nextTranche && counts[firstIndex] == 0); + } + + if (firstIndex >= counts.length / 2) + compact(); + } + } + + public void finishAll(long nextPosition) + { + while (firstTranche != nextTranche) + { + logger.warn("{} processed all pending tasks (<{}) but found {} waiting for {}", this, + nextPosition, counts[firstIndex], size() == 1 ? nextMin : mins[firstIndex + 1]); + advance(); + } + } + + private void advance() + { + Runnable run = runs[firstIndex]; + runs[firstIndex] = null; + ++firstIndex; + if (firstTranche == MAX_TRANCHE) firstTranche = 0; + else ++firstTranche; + try { run.run(); } + catch (Throwable t) { owner.agent.onException(t); } + } + + public void registerWait(Runnable run) + { + int newNextTranche = (nextTranche + 1) % (MAX_TRANCHE + 1); + if (newNextTranche == firstTranche) + { + Runnable cur = runs[firstIndex]; + if (cur instanceof WithDeferred) ((WithDeferred) cur).deferred.add(run); + else runs[firstIndex] = new WithDeferred(runs[firstIndex], run); + return; + } + + if ((firstIndex + size()) == capacity()) + growOrCompact(); + + int index = firstIndex + size(); + mins[index] = nextMin; + counts[index] = nextCount; + runs[index] = run; + nextMin = owner.minPosition = owner.nextPosition; + nextCount = 0; + nextTranche = newNextTranche; + } + + private void compact() + { + if (size() <= capacity()/4 && capacity() > 8) resize(capacity()/2); + else compact(mins, counts, runs); + } + + private void growOrCompact() + { + if (size() > capacity()/2) resize(capacity() * 2); + else compact(); + } + + private void resize(int newSize) + { + Invariants.require(newSize > 0); + long[] newMins = new long[newSize]; + int[] newCounts = new int[newSize]; + Runnable[] newRuns = new Runnable[newSize]; + compact(newMins, newCounts, newRuns); + mins = newMins; + counts = newCounts; + runs = newRuns; + } + + private void compact(long[] newMins, int[] newCounts, Runnable[] newRuns) + { + int size = size(); + System.arraycopy(mins, firstIndex, newMins, 0, size); + System.arraycopy(counts, firstIndex, newCounts, 0, size); + System.arraycopy(runs, firstIndex, newRuns, 0, size); + firstIndex = 0; } } - private final Lock lock; - final Agent agent; - final int executorId; - private final AccordCache cache; - - private final TaskQueue> scanningRanges = new TaskQueue<>(SCANNING_RANGES); // never queried, just parked here while scanning - private final TaskQueue> loading = new TaskQueue<>(LOADING); // never queried, just parked here while loading - private final TaskQueue running = new TaskQueue<>(RUNNING); - - private final TaskQueue> waitingToLoadRangeTxns = new TaskQueue<>(WAITING_TO_LOAD); - private final TaskQueue> waitingToLoad = new TaskQueue<>(WAITING_TO_LOAD); - private final TaskQueue waitingToRun = new TaskQueue<>(WAITING_TO_RUN); - - private final ExclusiveGlobalCaches caches; - - private List waitingForQuiescence; - private Queue waitingForCompletion; - final LogLinearDecayingHistograms histograms; final LogLinearDecayingHistogram elapsedPreparingToRun; final LogLinearDecayingHistogram elapsedWaitingToRun; @@ -215,11 +571,20 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor> scanningRanges = new StandaloneTaskQueue<>(TinyEnumSet.encode(SCANNING_RANGES)); // never queried, just parked here while scanning + private final StandaloneTaskQueue> waitingToLoadRangeTxns = new StandaloneTaskQueue<>(TinyEnumSet.encode(WAITING_TO_LOAD)); + private final StandaloneTaskQueue> waitingToLoad = new StandaloneTaskQueue<>(TinyEnumSet.encode(WAITING_TO_SCAN_RANGES, SCANNING_RANGES, WAITING_TO_LOAD)); + private final StandaloneTaskQueue> loading = new StandaloneTaskQueue<>(LOADING); + private final RunnableTaskQueue runnable = new RunnableTaskQueue<>(); + + private final Tranches tranches = new Tranches(this); + + /** + * Newly submitted work must take a position >= minPosition, but this condition does not apply to consequences of + * previously submitted work; this inherits the originating operation's position and tranche. + * This is to ensure afterSubmittedAndConsequences functions correctly. + */ + private long minPosition = 1; + private long nextPosition = 1; + int tasks; + private int activeLoads, activeRangeLoads; private boolean hasPausedLoading; - int tasks; - final DebugExecutor debug = DebugExecutor.maybeDebug(); + + private List waitingForQuiescence; AccordExecutor(Lock lock, int executorId, Agent agent) { @@ -267,7 +649,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor w.run.run()); - waitingForCompletion = null; - } + tranches.finishAll(nextPosition); } public void afterSubmittedAndConsequences(Runnable run) { - lock(); + AccordTaskRunner self = AccordTaskRunner.get(); + + lock(self); try { + drainUnqueuedNewWorkExclusive(); if (tasks == 0) { run.run(); return; } - if (waitingForCompletion != null) // escape hatch for some bug that means we lose a notification for a given task's queue position - maybeNotifyWaitingForCompletion(); - if (waitingForCompletion == null) - waitingForCompletion = new ArrayDeque<>(); - - long position = nextPosition; - minPosition = position; - waitingForCompletion.add(new WaitForCompletion(position, run)); + tranches.registerWait(run); } finally { - unlock(); + unlock(self); } } - void maybeUnpauseLoading() + final void maybeUnpauseLoading() { if (!hasPausedLoading) return; - if (cache.weightedSize() < maxWorkingCapacityInBytes || (loading.isEmpty() && waitingToRun.isEmpty())) + if (cache.weightedSize() < maxWorkingCapacityInBytes || (loading.isEmpty() || runnable.hasWaiting())) { hasPausedLoading = false; enqueueLoadsExclusive(); @@ -466,7 +812,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor> queue = waitingToLoadRangeTxns.isEmpty() || activeRangeLoads >= maxQueuedRangeLoads ? waitingToLoad : waitingToLoadRangeTxns; + StandaloneTaskQueue> queue = waitingToLoadRangeTxns.isEmpty() || activeRangeLoads >= maxQueuedRangeLoads ? waitingToLoad : waitingToLoadRangeTxns; AccordTask next = queue.peek(); if (next == null) return; @@ -474,7 +820,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor= maxWorkingCapacityInBytes) { // we have too much in memory already, and we have work waiting to run, so let that complete before queueing more - if (!loading.isEmpty() || !waitingToRun.isEmpty()) + if (!loading.isEmpty() || runnable.hasWaiting()) { AccordSystemMetrics.metrics.pausedExecutorLoading.inc(); hasPausedLoading = true; @@ -542,43 +888,24 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor task, AccordCacheEntry load) { - boolean isForRangeTxn = task.hasRanges(); + boolean isForRangeTxn = task.isRange(); if (!isForRangeTxn) return false; for (AccordTask t : load.loadingOrWaiting().waiters()) { - if (!t.hasRanges()) + if (!t.isRange()) return false; } return true; } - @Override - public Cancellable execute(RunOrFail runOrFail) - { - PlainChain submit = new PlainChain(runOrFail); - return submit(submit); - } - - public AsyncChain buildDebuggable(Callable task, Object describe) - { - return new AsyncChains.Head<>() - { - @Override - protected Cancellable start(BiConsumer callback) - { - return submit(new DebuggableChain(new CallAndCallback<>(task, callback), null, 0, describe)); - } - }; - } - private void parkRangeLoad(AccordTask task) { if (task.queued() != waitingToLoadRangeTxns) { - task.unqueueIfQueued(); - task.addToQueue(waitingToLoadRangeTxns); + task.unqueue(); + waitingToLoadRangeTxns.enqueue(task); } } @@ -590,13 +917,13 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor task) { task.onWaitingToRun(); - task.addToQueue(task.commandStore.exclusiveExecutor); + task.commandStore.exclusiveExecutor.enqueue(task); } private void waitingToRun(Task task, @Nullable ExclusiveExecutor queue) { task.onWaitingToRun(); - task.addToQueue(queue == null ? waitingToRun : queue); + if (queue == null) runnable.enqueue(task); + else queue.enqueue(task); } public ExclusiveExecutor executor() @@ -626,7 +954,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor i, operation); } - void submitPriority(AccordTask operation) - { - submit(AccordExecutor::submitPriorityExclusive, i -> i, operation); - } - void submitExclusive(AccordTask task) { - assignQueuePosition(task); - submitInternalExclusive(task); - } - - void submitPriorityExclusive(AccordTask task) - { - assignMinQueuePosition(task); - submitInternalExclusive(task); - } - - private void submitInternalExclusive(AccordTask task) - { + registerExclusive(task); task.setupExclusive(); - ++tasks; updateQueue(task); enqueueLoadsExclusive(); } public void submitExclusive(Runnable runnable) { - submitPlainExclusive(new PlainRunnable(null, runnable)); + submitPlainExclusive(new PlainRunnable(null, runnable, OTHER)); + } + + private Cancellable submitPlain(Plain task) + { + submit(AccordExecutor::submitPlainExclusive, i -> i, task); + return task; } private void submitPlainExclusive(Plain task) { - ++tasks; - assignQueuePosition(task); + registerExclusive(task); waitingToRun(task, task.executor()); } - Cancellable submitPlainExclusive(Task parent, AbstractIOTask task) + Cancellable submitPlainExclusive(Task parent, GlobalGroup group, AbstractIOTask task) { - return submitPlainExclusive(parent, new WrappedIOTask(task)); + return submitPlainExclusive(parent, new WrappedIOTask(task, group, parent.position, parent.tranche())); } - T submitPlainExclusive(Task parent, T task) + final T submitPlainExclusive(Task parent, T task) { Invariants.require(isOwningThread()); - ++tasks; - if (parent != null) inheritQueuePosition(parent, task); - else assignFifoQueuePosition(task); + if (parent == null) registerExclusive(task); + else registerConsequenceExclusive(parent, task); task.onWaitingToRun(); - waitingToRun.append(task); + runnable.enqueue(task); return task; } - private void assignQueuePosition(Task task) + private void registerConsequenceExclusive(Task parent, Task task) { - if (task.queuePosition != 0) updateNextPosition(task); - else assignFifoQueuePosition(task); + ++tasks; + int tranche = parent.tranche(); + tranches.addInherited(tranche, parent.position); + task.position = parent.position; + task.setInheritedTranche(tranche); } - private void assignQueuePosition(AccordTask task) + private void registerExclusive(Task task) { - if (task.queuePosition != 0) updateNextPosition(task); + ++tasks; + if (task.hasInherited()) + { + tranches.addInherited(task.tranche(), task.position); + } else { - long priority_bits = PRIORITY_BITS; - TxnId txnId = null; - switch (PRIORITY_MODEL) + long position = task.position; + if (position == 0) { - case ORIG_PHASE_HLC_FIFO: - case PHASE_HLC_FIFO: - { - // TODO (expected): we should process messages for a TxnId together, to avoid processing delayed messages out of order - ExecutionContext context = task.preLoadContext(); - if (context instanceof Request) - { - MessageType type = ((Request) context).type(); - if (type instanceof StandardMessage) - { - TxnId txnId0 = context.primaryTxnId(); - switch ((StandardMessage)type) - { - case APPLY_REQ: - { - priority_bits = 0L; - txnId = txnId0; - break; - } - case READ_EPHEMERAL_REQ: - case READ_REQ: - case STABLE_THEN_READ_REQ: - { - priority_bits = 1000000000000000L; - txnId = txnId0; - break; - } - case COMMIT_REQ: - { - Commit commit = (Commit) context; - if (PRIORITY_MODEL == PHASE_HLC_FIFO || commit.ballot.equals(Ballot.ZERO)) - txnId = commit.txnId; - if (commit.kind.saveStatus == SaveStatus.Stable) priority_bits = 1000000000000L; - else priority_bits = 2000000000000L; - break; - } - case ACCEPT_REQ: - { - Accept accept = (Accept) context; - if (PRIORITY_MODEL == PHASE_HLC_FIFO || accept.ballot.equals(Ballot.ZERO)) - txnId = accept.txnId; - priority_bits = 3000000000000L; - break; - } - case GET_EPHEMERAL_READ_DEPS_REQ: - case PRE_ACCEPT_REQ: - { - txnId = txnId0; - break; - } - } - } - } - break; - } - case HLC_FIFO: - { - txnId = task.preLoadContext().primaryTxnId(); - break; - } - case FIFO: - { - break; - } + task.position = position = nextPosition++; } - - if (txnId != null) + else { - long hlc = txnId.hlc(); - long delta = nextPosition - hlc; - if (delta < AGE_TO_FIFO) - { - long position = hlc; - if (delta <= 0) nextPosition = position + 1; - else if (position < minPosition) position = minPosition; - position |= priority_bits; - task.queuePosition = position; - return; - } + long delta = nextPosition - position; + if (delta >= AGE_TO_FIFO) + task.position = position = nextPosition++; + else if (delta <= 0) + nextPosition = position + 1; + else if (position < minPosition) + task.position = position = minPosition; } - - assignFifoQueuePosition(task); - } - - } - - private void assignMinQueuePosition(Task task) - { - task.queuePosition = minPosition | PRIORITY_BITS; - } - - private void assignFifoQueuePosition(Task task) - { - task.queuePosition = nextPosition++ | PRIORITY_BITS; - } - - private void updateNextPosition(Task task) - { - nextPosition = Math.max(nextPosition, (task.queuePosition & ~PRIORITY_BITS) + 1); - } - - private void inheritQueuePosition(Task parent, Task task) - { - task.queuePosition = parent.queuePosition; - } - - void completeTaskExclusive(Task task) - { - // for integration with SequentialExecutor, we must : - // - first take the position so that represents the just-executed task - // - call cleanup to submit any following task on the relevant sub-queue - // - remove the previous task from the running collection only if still present (SequentialExecutor will have removed it) - long position = task.queuePosition; - try - { - task.cleanupExclusive(this); - } - finally - { - --tasks; - if (running.contains(task)) - running.remove(task); - - if (waitingForCompletion != null && waitingForCompletion.peek().maybeNotify <= position) - maybeNotifyWaitingForCompletion(); - - cache.tryShrinkOrEvict(lock); + task.setTranche(tranches.addNew(position)); } } - private void maybeNotifyWaitingForCompletion() + final void completeTaskExclusive(Task task) { - long min = minPosition(waitingToRun.peek(), - minPosition(waitingToLoad.peek(), - minPosition(waitingToLoadRangeTxns.peek(), - minPosition(running.peek(), - minPosition(loading.peek(), - minPosition(scanningRanges.peek(), Long.MAX_VALUE)))))); - - while (!waitingForCompletion.isEmpty() && waitingForCompletion.peek().position - min <= 0) - waitingForCompletion.poll().run.run(); - if (waitingForCompletion.isEmpty()) - waitingForCompletion = null; - else - waitingForCompletion.peek().maybeNotify = min; + runnable.complete(task); + try { task.cleanupExclusive(this); } + finally { cache.tryShrinkOrEvict(lock); } } - private static long minPosition(@Nullable Task task, long min) + final void unregisterExclusive(Task task) { - return task == null ? min : Long.min(task.queuePosition, min); + int tranch = task.tranche(); + --tasks; + tranches.complete(tranch); } - void cancelExclusive(AccordTask task) + final void cancelExclusive(AccordTask task) { AccordTask.State state = task.state(); switch (state) @@ -972,7 +1171,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor> tasks = loaded.loading().copyWaiters()) + try (ArrayBuffers.BufferList> tasks = loaded.loading().copyWaiters()) { if (fail != null) { @@ -999,35 +1198,74 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor submit(Runnable run) { - PlainRunnable task = new PlainRunnable(new AsyncPromise<>(), run); - submit(task); + Task inherit = inherit(); + PlainRunnable task = inherit == null ? new PlainRunnable(new AsyncPromise<>(), run, OTHER) + : new PlainRunnable(new AsyncPromise<>(), run, OTHER, inherit.position, inherit.tranche()); + submitPlain(task); return task.result; } - public void execute(Runnable command) + public void execute(Runnable run) { - submit(new PlainRunnable(null, command)); + Task inherit = inherit(); + PlainRunnable task = inherit == null ? new PlainRunnable(null, run, OTHER) + : new PlainRunnable(null, run, OTHER, inherit.position, inherit.tranche()); + submitPlain(task); } - private Cancellable submit(Plain task) + @Override + public Cancellable execute(RunOrFail runOrFail) { - submit(AccordExecutor::submitPlainExclusive, i -> i, task); - return task; + Task inherit = inherit(); + PlainChain submit = inherit == null ? new PlainChain(runOrFail, null, ExclusiveGroup.OTHER) + : new PlainChain(runOrFail, null, ExclusiveGroup.OTHER, inherit.position, inherit.tranche()); + return submitPlain(submit); + } + + public AsyncChain buildDebuggable(Callable call, Object describe) + { + return new AsyncChains.Head<>() + { + @Override + protected Cancellable start(BiConsumer callback) + { + Task inherit = inherit(); + return submitPlain(inherit == null ? new DebuggableChain(new CallAndCallback<>(call, callback), null, describe) + : new DebuggableChain(new CallAndCallback<>(call, callback), null, inherit.position, inherit.tranche(), describe)); + } + }; } public void executeDirectlyWithLock(Runnable command) { - lock(); + AccordTaskRunner self = AccordTaskRunner.get(); + lock(self); try { + self.setAccordActiveExecutor(this); command.run(); } finally { beforeUnlockExternal(); - unlock(); + self.clearAccordLockedExecutor(); + unlock(self); } } @@ -1077,60 +1315,287 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor runningUpdater = AtomicReferenceFieldUpdater.newUpdater(TaskRunner.class, Task.class, "running"); + private volatile AccordTaskRunner wrapped; + + protected void setWrapped(AccordTaskRunner wrapped) + { + this.wrapped = wrapped; + } @Override public DebuggableTask running() { - Task running = this.running; + Task running = wrapped.accordActiveTask(); return running == null ? null : running.debuggable(); } - - Task runningTask() - { - return running; - } - - void setRunning(Task debuggable) - { - runningUpdater.lazySet(this, debuggable); - } - - void clearRunning() - { - runningUpdater.lazySet(this, null); - } } - public static abstract class Task extends IntrusivePriorityHeap.Node + public static abstract class Task extends IntrusiveHeapNode { + 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; + static final State[] VALUES = values(); + + State() + { + this.permittedFrom = 0; + } + + State(State ... permittedFroms) + { + int permittedFrom = 0; + for (State state : permittedFroms) + permittedFrom |= 1 << state.ordinal(); + this.permittedFrom = permittedFrom; + } + + boolean isPermittedFrom(int prevOrdinal) + { + return (permittedFrom & (1 << prevOrdinal)) != 0; + } + + boolean isExecuted() + { + return this.compareTo(PERSISTING) >= 0; + } + + boolean isComplete() + { + return this.compareTo(FINISHED) >= 0; + } + + static State forOrdinal(int ordinal) + { + return VALUES[ordinal]; + } + } + + enum GlobalGroup + { + COMMAND_STORE, + LOAD, + SAVE, + OTHER, + RANGE_LOAD(true), + RANGE_SCAN(true), + ; + + final int bits; + + GlobalGroup() + { + this(false); + } + + GlobalGroup(boolean isRange) + { + this.bits = (ordinal() << GLOBAL_GROUP_SHIFT) | (isRange ? RANGE_BIT : 0); + } + } + + enum ExclusiveGroup + { + APPLY, + STABLE, + COMMIT, + ACCEPT, + OTHER, + PREACCEPT, + RANGE(true), + ; + + final int bits; + + ExclusiveGroup() + { + this(false); + } + + ExclusiveGroup(boolean isRange) + { + this.bits = (ordinal() << EXCLUSIVE_GROUP_SHIFT) | (isRange ? RANGE_BIT : 0); + } + } + + public enum GroupKind + { + EXCLUSIVE(ExclusiveGroup.values().length, EXCLUSIVE_GROUP_SHIFT), + GLOBAL(GlobalGroup.values().length, GLOBAL_GROUP_SHIFT), + NONE(0, 0); + + final int count; + final byte shift; + GroupKind(int count, int shift) + { + this.count = count; + this.shift = (byte) shift; + } + } + + private static final int STATE_MASK = 0xf; + private static final int GROUP_MASK = 0x7; + private static final int GLOBAL_GROUP_SHIFT = 7; + private static final int EXCLUSIVE_GROUP_SHIFT = 4; + private static final int RANGE_BIT = 1 << 10; + private static final int CLEANUP_BIT = 1 << 11; + private static final int HAS_TRANCHE_BIT = 1 << 12; + private static final int HAS_INHERITED_BIT = 1 << 13; + private static final int TRANCHE_SHIFT = 14; + static final int MAX_TRANCHE = 0x3ff; + public final WithResources resources; Task next; - long queuePosition; + + long position; + private int info; + + // TODO (expected): do we need this? we should be able to determine the queue from state() if needed for e.g. cancellation + private TaskQueue queued; + + // TODO (expected): expose via executors vtable + // TODO (expected): use just one long and some flag bits to indicate which point it represents, and report incrementally public long createdAt = nanoTime(), waitingToRunAt, runningAt, cleanupAt; - protected Task() + protected Task(GlobalGroup group, State state) { resources = DebugTask.maybeDebug(ExecutorLocals.propagate(), this); + info = init(group, ExclusiveGroup.OTHER, state); + } + + protected Task(ExclusiveGroup group, State state) + { + resources = DebugTask.maybeDebug(ExecutorLocals.propagate(), this); + info = init(GlobalGroup.OTHER, group, state); + } + + protected Task(GlobalGroup group, State state, long position, int tranche) + { + this(group, state); + this.position = position; + setInheritedTranche(tranche); + } + + protected Task(ExclusiveGroup group, State state, long position, int tranche) + { + this(group, state); + this.position = position; + setInheritedTranche(tranche); + } + + protected Task(ExecutionContext context) + { + resources = DebugTask.maybeDebug(ExecutorLocals.propagate(), this); + ExclusiveGroup group = ExclusiveGroup.OTHER; + TxnId txnId = context.primaryTxnId(); + if (txnId != null) + { + if (txnId.is(Range)) group = ExclusiveGroup.RANGE; + else + { + switch (PRIORITY_MODEL) + { + case HLC_FIFO: + case ORIG_HLC_FIFO: + { + // TODO (expected): we should process messages for a TxnId together, to avoid processing delayed messages out of order + if (context instanceof Request) + { + MessageType type = ((Request) context).type(); + if (type instanceof StandardMessage) + { + switch ((StandardMessage)type) + { + case APPLY_REQ: + { + group = APPLY; + break; + } + case READ_EPHEMERAL_REQ: + case READ_REQ: + case STABLE_THEN_READ_REQ: + { + group = STABLE; + break; + } + case COMMIT_REQ: + { + Commit commit = (Commit) context; + if (PRIORITY_MODEL == ORIG_HLC_FIFO && !commit.ballot.equals(Ballot.ZERO)) + txnId = null; + if (commit.kind.saveStatus == SaveStatus.Stable) group = STABLE; + else group = COMMIT; + break; + } + case ACCEPT_REQ: + { + Accept accept = (Accept) context; + if (PRIORITY_MODEL == ORIG_HLC_FIFO && !accept.ballot.equals(Ballot.ZERO)) + txnId = null; + group = ExclusiveGroup.ACCEPT; + break; + } + case GET_EPHEMERAL_READ_DEPS_REQ: + case PRE_ACCEPT_REQ: + { + group = ExclusiveGroup.PREACCEPT; + break; + } + default: + { + txnId = null; + } + } + } + } + else + { + txnId = null; + } + break; + } + case FIFO: + { + txnId = null; + break; + } + } + } + } + + this.info = init(GlobalGroup.OTHER, group, INITIALIZED); + if (txnId != null) + this.position = txnId.hlc(); } public final Task unwrap() { - if (this instanceof SequentialQueueTask) - return ((SequentialQueueTask) this).queue.task; + if (this instanceof ExclusiveExecutorTask) + return ((ExclusiveExecutorTask) this).queue.task; return this; } final void setReadyToCleanup() { - queuePosition |= Long.MIN_VALUE; + info |= CLEANUP_BIT; } final boolean isReadyToCleanup() { - return 0 != (queuePosition & Long.MIN_VALUE); + return 0 != (info & CLEANUP_BIT); } final void onRunning() @@ -1165,6 +1630,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor>> GLOBAL_GROUP_SHIFT) & GROUP_MASK; + } + + final int exclusiveGroupOrdinal() + { + return (info >>> EXCLUSIVE_GROUP_SHIFT) & GROUP_MASK; + } + + @Nullable + final TaskQueue queued() + { + return queued; + } + + final void unqueueIfQueued() + { + if (queued != null) + unqueue(); + } + + final void unqueue() + { + Invariants.require(queued != null); + queued.unqueue(this); + queued = null; + } + + final void unsetQueue(TaskQueue queue) + { + Invariants.require(queued == queue); + queued = null; + } + + final void setQueue(TaskQueue queue) + { + Invariants.require(queued == null); + Invariants.require(isCompatible(queue)); + queued = queue; + } + + private boolean isCompatible(TaskQueue queue) + { + int self = stateOrdinal(); + return TinyEnumSet.contains(queue.states, self); + } + + private static int init(GlobalGroup global, ExclusiveGroup exclusive, State state) + { + return global.bits | exclusive.bits | state.ordinal(); + } + + final void setTranche(int tranche) + { + Invariants.require(tranche <= MAX_TRANCHE); + info = info | (tranche << TRANCHE_SHIFT) | HAS_TRANCHE_BIT; + } + + final void setInheritedTranche(int tranche) + { + Invariants.require(tranche <= MAX_TRANCHE); + info = info | (tranche << TRANCHE_SHIFT) | HAS_TRANCHE_BIT | HAS_INHERITED_BIT; + } + + final int tranche() + { + Invariants.require((info & HAS_TRANCHE_BIT) != 0); + return info >>> TRANCHE_SHIFT; + } + + final boolean hasInherited() + { + return (info & HAS_INHERITED_BIT) != 0; + } } // run the task even on a stopped commandStore @@ -1215,16 +1805,22 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor ownerUpdater = AtomicReferenceFieldUpdater.newUpdater(ExclusiveExecutor.class, Thread.class, "owner"); - public class ExclusiveExecutor extends TaskQueue implements accord.local.ExclusiveAsyncExecutor + public final class ExclusiveExecutor extends MultiTaskQueue implements ExclusiveAsyncExecutor { final int commandStoreId; - final SequentialQueueTask selfTask; + final ExclusiveExecutorTask selfTask; private Task task; - private volatile Thread owner, waiting; + volatile Thread owner, waiting; private boolean stopped; private volatile boolean visibleStopped; private boolean terminated; - final DebugSequentialExecutor debug; + final DebugExclusiveExecutor debug; ExclusiveExecutor(AccordExecutor executor) { @@ -1284,15 +1879,14 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor) task).preLoadContext())); + task.fail(new RejectedExecutionException(commandStoreId + " is terminated. Cannot execute " + ((AccordTask) task).executionContext())); else task.run(); // NOTE: cannot safely release owner here, in case an immediate-execution runs before we can release our references and store their changes to the cache @@ -1330,8 +1924,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor)) return true; - ExecutionContext context = ((AccordTask) task).preLoadContext(); - + ExecutionContext context = ((AccordTask) task).executionContext(); return !(terminated ? (context instanceof Unterminatable) : (context instanceof Unstoppable)); } @@ -1345,71 +1938,78 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor chain(Runnable run) { - long position = inheritQueuePosition(); - return new AsyncChains.Head<>() - { - @Override - protected Cancellable start(BiConsumer callback) - { - return execute(new RunAndCallback(run, callback), position); - } - }; + return AsyncChains.chain(this, run); } @Override - public AsyncChain chain(Callable call) + public AsyncChain chain(Callable call) { - long position = inheritQueuePosition(); - return new AsyncChains.Head<>() - { - @Override - protected Cancellable start(BiConsumer callback) - { - return execute(new CallAndCallback<>(call, callback), position); - } - }; + return AsyncChains.chain(this, call); } @Override - public AsyncChain flatChain(Callable> call) + public AsyncChain flatChain(Callable> call) { - long position = inheritQueuePosition(); - return new AsyncChains.Head<>() - { - @Override - protected Cancellable start(BiConsumer callback) - { - return execute(new FlatCallAndCallback<>(call, callback), position); - } - }; + return AsyncChains.flatChain(this, call); } - @Override - public Cancellable execute(RunOrFail runOrFail) + Task inherit() { - return execute(runOrFail, inheritQueuePosition()); - } - - private long inheritQueuePosition() - { - return inExecutor() && task != null ? task.queuePosition : 0; - } - - private Cancellable execute(RunOrFail runOrFail, long queuePosition) - { - PlainChain submit = new PlainChain(runOrFail, ExclusiveExecutor.this, queuePosition); - return AccordExecutor.this.submit(submit); + Thread thread = Thread.currentThread(); + if (thread == owner) + return task; + return AccordTaskRunner.get(thread).accordActiveSelfTask(); } @Override public void execute(Runnable run) { - PlainRunnable submit = new PlainRunnable(null, run, this, inheritQueuePosition()); - AccordExecutor.this.submit(submit); + Task inherit = inherit(); + PlainRunnable submit = inherit == null ? new PlainRunnable(null, run, this, ExclusiveGroup.OTHER) + : new PlainRunnable(null, run, this, ExclusiveGroup.OTHER, inherit.position, inherit.tranche()); + submitPlain(submit); + } + + @Override + public Cancellable execute(RunOrFail runOrFail) + { + Task inherit = inherit(); + PlainChain submit = inherit == null ? new PlainChain(runOrFail, ExclusiveExecutor.this, ExclusiveGroup.OTHER) + : new PlainChain(runOrFail, ExclusiveExecutor.this, ExclusiveGroup.OTHER, inherit.position, inherit.tranche()); + return submitPlain(submit); } @Override public boolean tryExecuteImmediately(Runnable run) { - Thread self = Thread.currentThread(); + Thread thread = Thread.currentThread(); Thread owner = this.owner; - if (owner != null ? owner != self : !ownerUpdater.compareAndSet(this, null, self)) + if (owner != null && owner != thread) return false; + AccordTaskRunner self = AccordTaskRunner.get(thread); + AccordExecutor active = self.accordActiveExecutor(); + if (active != null && active != AccordExecutor.this) + return false; // prevent cross-executor locking/execution + + if (!ownerUpdater.compareAndSet(this, null, thread)) + return false; + + if (active == null) + self.setAccordActiveExecutor(AccordExecutor.this); + try { run.run(); } catch (Throwable t) { agent.onException(t); } finally { + if (active == null) + self.setAccordActiveExecutor(null); + if (owner == null) { Thread waiting = this.waiting; @@ -1559,80 +2133,834 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor extends IntrusivePriorityHeap { - final AccordTask.State kind; + final int states; + public TaskQueue(int states) {this.states = states; } - TaskQueue(AccordTask.State kind) + void unqueue(T task) { - this.kind = kind; - } - - TaskQueue(AccordTask.State kind, boolean tiny) - { - super(tiny); - this.kind = kind; + throw new UnsupportedOperationException(); } @Override - public int compare(T o1, T o2) + public final int compare(T o1, T o2) { - return Long.compare(o1.queuePosition, o2.queuePosition); + return Long.compare(o1.position, o2.position); } - protected void append(T task) + @Override + protected final void ensureHeapified() { - super.append(task); + super.ensureHeapified(); } - protected void update(T task) + protected final boolean requeueSingle(T requeue) { - super.update(task); + int oldIndex = updateNode(requeue); + if (oldIndex < 0) + return false; + + int newIndex = heapIndex(requeue); + return Math.min(oldIndex, newIndex) == 0 && heapifiedSize() > 0; } - protected T poll() - { - ensureHeapified(); - return pollNode(); - } - - protected T peek() + final T peekSingle() { ensureHeapified(); return peekNode(); } - protected T get(int index) + final T pollSingle() { - return super.get(index); + ensureHeapified(); + return pollNode(); } - protected void remove(T remove) + final boolean isEmptySingle() { - super.remove(remove); + return isEmptyInternal(); } - @Override - protected boolean removeIfContains(T node) + final int enqueueSingle(T enqueue) { - return super.removeIfContains(node); + return insertNode(enqueue); } - protected boolean contains(T contains) + final boolean unqueueSingle(T unqueue) { - return super.contains(contains); + int heapIndex = heapIndex(unqueue); + removeNode(unqueue); + return heapIndex == 0; + } + + final boolean tryUnqueueSingle(T remove) + { + return removeNodeIfContains(remove); + } + + final T getSingle(int index) + { + return super.getNode(index); + } + + final boolean isQueuedSingle(T test) + { + return containsNode(test); + } + } + + static final class StandaloneTaskQueue extends TaskQueue + { + StandaloneTaskQueue(int states) + { + super(states); + } + + StandaloneTaskQueue(Task.State state) + { + super(TinyEnumSet.encode(state)); + } + + void enqueue(T enqueue) + { + enqueue.setQueue(this); + enqueueSingle(enqueue); + } + + void unqueue(T unqueue) + { + unqueue.unsetQueue(this); + removeNode(unqueue); + } + + T peek() + { + return peekSingle(); + } + + T poll() + { + return pollSingle(); + } + + boolean isEmpty() + { + return isEmptySingle(); + } + } + + /** + * A {@link TaskQueue} sub-divided into up to eight per-group sub-queues (groups are phases for the exclusive + * executor, or work classes globally) plus a policy for choosing which sub-queue to serve next. The policy balances + * three competing concerns: ordering/priority (run the oldest / highest-priority work), fairness (do not let one + * group monopolise the executor), and throughput (do not let an even split leave a busy group's backlog growing + * without bound). + * + *

Packed counters

+ * Per-group state is held in {@code long}s, one 7-bit lane per group; bit 7 of each byte is a spare "guard" bit used + * to run branch-free min/compare across all eight lanes at once (see {@code minCounters}). The lanes are: + *
    + *
  • {@code positions[]} - the head task's position (HLC, or submission order) per group; drives FIFO/priority.
    • + *
  • {@code recent} - a windowed count of recently served tasks per group (service received).
    • + *
  • {@code arrivals} - a windowed, size-biased count of recent arrivals per group (demand offered).
    • + *
  • {@code current} - tasks currently in flight per group, used to enforce {@code queue_active_limits}.
    • + *
  • {@code hasWork}/{@code dirty} - guard-bit masks: which groups have queued work / need a position refresh.
    • + *
+ * + *

Windowing: a bounded memory of the past

+ * {@code recent} and {@code arrivals} are not running totals. Whenever incrementing {@code recent} tips any lane to + * its maximum (0x80), both {@code recent} and {@code arrivals} are halved (see {@code incrementRecents}), + * and each lane also saturates at 0x7f. They are therefore an exponentially-decaying window keyed on service/time. + * This is what stops a one-off burst of arrivals granting a group a permanent boost: the burst saturates the lane + * briefly and then decays away as other work is served. + * + *

The fair selection counter

+ * The fair models pick the group with the smallest {@code effective} counter, where per lane + * 
effective = min(0x7f, max(0, recent - arrivals) + bias)
+ *
    + *
  • {@code max(0, recent - arrivals)} - a group whose arrivals have outpaced its recent service clamps to 0 and + * is therefore preferred; in steady state each group's service tracks its arrival rate, which bounds backlog. + * The clamp is symmetric, so a group cannot bank "credit" during a lull and later use it to monopolise service.
    • + *
+ * + *

Choosing a group ({@code minGroup} / {@code pollByBlend})

+ *
    + *
  • {@code PRIORITY_ONLY} - always the lowest {@code positions} (oldest / highest priority), ignoring fairness.
    • + *
  • {@code PHASE_OVERRIDE} - strict phase order: the lowest-index group with work, always.
    • + *
  • {@code PHASE_FAIR} - pure fairness: the smallest {@code effective}; ties broken by index, within a group by + * position. It has no priority fallback, so it is deliberately completing-first under contention.
    • + *
  • {@code PRIORITY_FAIR} (default) - a deficit round-robin blend of two strategies, chosen per poll: flow + * (smallest {@code effective}, i.e. least fairly serviced) and age (oldest {@code positions}, i.e. FIFO). + * The flow weight ramps up with the flow imbalance ({@code max-min} of {@code effective}) over + * {@code queue_flow_imbalance_onset..+width}, trading against age zero-sum; when balanced it is pure age. The + * ramp is smooth (not a hard mode switch), so there is no boundary to oscillate around and no hysteresis is + * needed. Age also drains stale/standing backlogs for free: a backlog's items are the oldest work, so age + * clears them without a separate backlog-aware strategy (the {@code effective} flow counter tracks arrival + * rate and is deliberately blind to standing stock).
    • + *
+ * + *

Concurrency limits and eligibility

+ * A group whose in-flight {@code current} count has reached its {@code queue_active_limits} limit is + * {@code saturated} and excluded from selection ({@code disabled}), as is any group with no queued work. + * + *

Note on {@code positions} freshness

+ * {@code positions} is refreshed lazily from the sub-queue heads via the {@code dirty} mask; {@code dirty} must use + * the same guard-bit layout as {@code hasWork} so the refresh loop in {@code minGroupByPriority} actually runs - + * otherwise it silently degenerates to lowest-index-with-work and starves high-index / new-work groups. + * + *

NB it extends {@link TaskQueue} to keep the type hierarchy simple for method dispatch, and for efficiency for + * anonymous ExclusiveExecutors which do not use multiple queues, while letting ExclusiveExecutor share a parent + * class for both use cases. + */ + static abstract class MultiTaskQueue extends TaskQueue + { + private static final TaskQueue[] NO_QUEUES = new TaskQueue[0]; + private static final long[] NO_POSITIONS = new long[0]; + static final long COUNTER_OVERFLOWS = 0x8080808080808080L; + static final long COUNTER_MASKS = 0x7f7f7f7f7f7f7f7fL; + static final long COUNTER_LOWBITS = 0x0101010101010101L; + + final TaskQueue[] queues; + final long[] positions; + final byte groupShift; + final long baseBudget, limits; + + /** sets overflow bits for each counter when it needs its position updated */ + long dirty; + /** sets overflow bits for each counter when there's associated work */ + long hasWork; + /** Stores recent dequeue counts for up to 8 sub queues. */ + long dispatches; + /** Stores recent enqueue counts for up to 8 sub queues. */ + long arrivals; + /** Stores currently-active counts for up to 8 sub queues. We can use this to impose limits on specific queues. */ + long active; + /** Stores a biased budget for each task type, so that we may prefer to serve one type of task over another */ + long budget; + /** deficit-round-robin credits for the two PRIORITY_FAIR strategies (flow/age). */ + int creditFlow, creditAge; + + int waitingCount; + + MultiTaskQueue(int waitingStates, GroupKind groups, long budget, long limits) + { + super(waitingStates); + this.baseBudget = budget; + this.limits = limits; + int queueCount = groups.count; + Invariants.require(queueCount <= 8); + queues = queueCount == 0 ? NO_QUEUES : new TaskQueue[queueCount]; + positions = queueCount > 0 && BALANCE_BY_POSITION ? new long[queueCount] : NO_POSITIONS; + groupShift = groups.shift; + } + + final int group(Task task) + { + if (groupShift == 0) + return -1; + + return (task.info >>> groupShift) & Task.GROUP_MASK; + } + + final TaskQueue queue(Task task) + { + int group = group(task); + if (group < 0) + return this; + + return queue(group); + } + + final TaskQueue queue(int group) + { + TaskQueue queue = queues[group]; + if (queue == null) + queues[group] = queue = new TaskQueue<>(0); + + return queue; + } + + private int pollGroup() + { + if (hasWork == 0) + return -1; + + switch (BALANCING_MODEL) + { + default: throw new UnhandledEnum(PRIORITY_MODEL); + case PRIORITY_ONLY: return pollGroupByPriority(); + case PHASE_ONLY: return pollGroupByIndex(); + case PRIORITY_BUDGET: return pollGroupByPriorityWithBudget(); + case PHASE_BUDGET: return pollGroupByIndexWithBudget(); + case PHASE_FAIR: return pollGroupByPhaseFair(); + case PHASE_BUDGET_FAIR: return pollGroupByPhaseBudgetFair(); + case BLENDED_PRIORITY_PHASE_FAIR: return pollGroupByBlended(); + case BLENDED_PRIORITY_PHASE_BUDGET_FAIR: return pollGroupByBlendedWithBudget(); + } + } + + private int pollGroupByPriority() + { + return pollGroupByPriority(unsaturatedWithWork()); + } + + private int pollGroupByPriority(long enabled) + { + long refresh = dirty & hasWork; + while (refresh != 0) + { + int bitIndex = Long.numberOfTrailingZeros(refresh); + int group = bitIndex / 8; + positions[group] = queues[group].peekSingle().position; + refresh ^= 1L << bitIndex; + } + dirty = 0; + + long minPosition = Long.MAX_VALUE; + int minGroup = -1; + long visit = enabled >>> 7; + while (visit != 0) + { + int bitIndex = Long.numberOfTrailingZeros(visit); + int group = bitIndex / 8; + long position = positions[group]; + if (position < minPosition) + { + minGroup = group; + minPosition = position; + } + visit ^= 1L << bitIndex; + } + + return minGroup; + } + + private int pollGroupByIndex() + { + long visit = (hasWork & unsaturated()) >>> 7; + if (visit == 0) + return -1; + + int bitIndex = Long.numberOfTrailingZeros(visit); + return bitIndex / 8; + } + + private int pollGroupByPriorityWithBudget() + { + return pollGroupByPriority(unsaturatedWithWorkAndBudget()); + } + + private int pollGroupByIndexWithBudget() + { + long visit = (hasWork & unsaturated()); + if (visit == 0) + return -1; + + visit = withBudgetOrReset(visit); + visit >>>= 7; + int bitIndex = Long.numberOfTrailingZeros(visit); + return bitIndex / 8; + } + + private int pollGroupByPhaseFair() + { + return minCounterIndex(recentFlowImbalances()); + } + + private int pollGroupByPhaseBudgetFair() + { + return minCounterIndex(recentFlowImbalances(), saturatedOrWithoutWorkOrWithoutBudget()); + } + + // PRIORITY_FAIR selection: a deficit round-robin blend of two strategies, chosen per poll: + // flow -> minCounterIndex(recent - arrivals + bias) (least fairly serviced) + // age -> minGroupByPriority() (earliest-queued work) + // wFlow = ramp(flow imbalance F = max-min of the selection counters); wAge = BLEND_TOTAL - wFlow. As flow gets + // uneven, polls trade from age to flow zero-sum; when balanced it is pure age (FIFO). A ramp (not a hard + // threshold) means there is no mode cliff to oscillate around, so no anti-oscillation penalty is needed. Age is + // also what drains a stale/standing backlog -- its items are the oldest -- so no separate stock strategy is needed. + private int pollGroupByBlended() + { + return pollGroupByBlended(saturatedOrWithoutWork()); + } + + private int pollGroupByBlended(long disabled) + { + long withoutWork = hasWork ^ COUNTER_OVERFLOWS; + long counters = recentFlowImbalances(); + long minMax = minMaxCounterValue(counters, withoutWork); + long min = minMax & 0x7f; + long max = minMax >>> 8; + int flowImbalance = (int) (max - min); + + int flowWeight = flowWeight(flowImbalance); + int priorityWeight = BLEND_TOTAL - flowWeight; + + creditFlow += flowWeight; + creditAge += priorityWeight; + + if (creditFlow >= creditAge) + { + creditFlow -= BLEND_TOTAL; + if (disabled != withoutWork) + min = minCounterValue(counters, disabled); + + return minCounterIndex(counters, min, disabled); + } + else + { + creditAge -= BLEND_TOTAL; + return pollGroupByPriority(disabled ^ COUNTER_OVERFLOWS); + } + } + + private int pollGroupByBlendedWithBudget() + { + return pollGroupByBlended(saturatedOrWithoutWorkOrWithoutBudget()); + } + + private long saturated() + { + return ((active | COUNTER_OVERFLOWS) - limits) & COUNTER_OVERFLOWS; + } + + private long unsaturated() + { + return saturated() ^ COUNTER_OVERFLOWS; + } + + private long unsaturatedWithWork() + { + return hasWork & unsaturated(); + } + + private long unsaturatedWithWorkAndBudget() + { + return withBudgetOrReset(hasWork & unsaturated()); + } + + private long saturatedOrWithoutWorkOrWithoutBudget() + { + return withoutBudgetOrReset(saturatedOrWithoutWork()); + } + + private long withoutBudgetOrReset(long disabled) + { + return COUNTER_OVERFLOWS ^ withBudgetOrReset(COUNTER_OVERFLOWS ^ disabled); + } + + private long withBudgetOrReset(long enabled) + { + long hasBudget = hasBudget(); + if ((hasBudget & enabled) != 0) + return hasBudget & enabled; + + budget = baseBudget; + return enabled; + } + + private long hasBudget() + { + return setOverflowWhenLessEqual(budget, 0) ^ COUNTER_OVERFLOWS; + } + + private long saturatedOrWithoutWork() + { + return (hasWork ^ COUNTER_OVERFLOWS) | saturated(); + } + + // arrivals is a windowed measure of ARRIVAL (incremented on enqueue, size-biased; decays on recent's overflow). + // Combined with recent (service) as effective = max(0, recent - arrivals): a queue whose arrivals outpace its + // service clamps to 0 and is preferred, so service converges to arrival rate (bounding the busy queue's backlog). + private long recentFlowImbalances() + { + return clampedSubtract(dispatches, arrivals); + } + + static long minCounterValue(long counters, long disabled) + { + long mins = counters; + mins |= overflowsToLowMasks(disabled); + mins = minCounters(mins, mins >>> 8); // each slot is min of slots [i..i+1] + mins = minCounters(mins, mins >>> 16); // each slot is min of slots [i..i+3] + mins = minCounters(mins, mins >>> 32); // each slot is min of slots [i..i+7] + return mins & 0x7f; + } + + static long minMaxCounterValue(long counters, long disabled) + { + long mins = counters; + long maxs = counters ^ COUNTER_MASKS; + long overflowMasks = overflowsToLowMasks(disabled); + mins |= overflowMasks; + maxs |= overflowMasks; + mins = minCounters(mins, mins >>> 8) & 0x007f007f007f007fL; // each slot is min of slots [i..i+1] + maxs = (minCounters(maxs, maxs << 8) & 0x7f007f007f007f00L); // each slot is min of slots ~[i..i+1] + long minmaxs = mins | maxs; + minmaxs = minCounters(minmaxs, minmaxs >>> 16); // each slot is min of slots [i..i+3] + minmaxs = minCounters(minmaxs, minmaxs >>> 32); // each slot is min of slots [i..i+7] + return (minmaxs ^ 0x7f00) & 0x7f7f; + } + + /** + * If provided two counters (containing 8 7 bit counters each), + * returns the minimum of each matching counter + */ + private static long minCounters(long a, long b) + { + // set overflow bits where a <= b + long selecta = setOverflowWhenLessEqual (a, b); + return selectByOverflowBits(selecta, a, b); + } + + static long setOverflowWhenLessEqual(long a, long b) + { + return ((b | COUNTER_OVERFLOWS) - a) & COUNTER_OVERFLOWS; + } + + // select a if overflow bit is set; b if it is unset + static long selectByOverflowBits(long selecta, long a, long b) + { + selecta = overflowsToLowMasks(selecta); + a &= selecta; + b &= ~selecta; + return a | b; + } + + static long overflowsToLowMasks(long v) + { + return v - (v >>> 7); + } + + private static int flowWeight(int flowImbalance) + { + if (flowImbalance <= FLOW_ONSET) return 0; + return Math.min(BLEND_TOTAL, ((flowImbalance - FLOW_ONSET) << BLEND_SHIFT) >>> FLOW_WIDTH_SHIFT); + } + + // per-lane max(0, a - b), carry-free: zero both a and b in lanes where a <= b, then subtract + private static long clampedSubtract(long a, long b) + { + long keep = ~overflowsToLowMasks(setOverflowWhenLessEqual(a, b)); + return (a & keep) - (b & keep); + } + + private int minCounterIndex(long counters) + { + return minCounterIndex(counters, saturatedOrWithoutWork()); + } + + private int minCounterIndex(long counters, long disabled) + { + return minCounterIndex(counters, minCounterValue(counters, disabled), disabled); + } + + private int minCounterIndex(long counters, long minCounterValue, long disabled) + { + long mins = minCounterValue * COUNTER_LOWBITS; + long select = ((mins | COUNTER_OVERFLOWS) - counters) & COUNTER_OVERFLOWS; + // now unset those overflow bits associated with disabled queues + select &= ~disabled; + if (select == 0) + return -1; + return (Long.numberOfTrailingZeros(select) - 7) / 8; + } + + final T pollMulti() + { + int group = pollGroup(); + if (group < 0) + { + // group < 0 can mean EITHER we don't have any nested queues OR those queues are either empty or DISABLED + T result = pollSingle(); + if (result != null) + --waitingCount; + return result; + } + + --waitingCount; + incrementActive(group); + incrementDispatches(group); + decrementBudget(group); + + TaskQueue queue = queues[group]; + T head = queue.pollSingle(); + // NOTE: must clear dirty when emptied, symmetrically with unqueue(): the fair selection paths + // never consume the dirty bit, so a group drained during a fairness episode would otherwise + // retain a stale dirty bit and NPE in minGroupByPriority.peekSingle() when balance is restored. + if (queue.isEmptySingle()) { unsetHasWork(group); unsetDirty(group); } + else setDirty(group); + return head; + } + + final void enqueueMulti(T task) + { + task.setQueue(this); + int group = group(task); + if (group < 0) + { + enqueueSingle(task); + } + else + { + TaskQueue queue = queue(group); + int result = queue.enqueueSingle(task); + incrementArrivals(group); + if (result < 0) setHasWork(group); + if (result != 0) setDirty(group); + } + ++waitingCount; + } + + final void requeue(T task) + { + int group = group(task); + if (group < 0) requeueSingle(task); + else + { + TaskQueue queue = queue(group); + Invariants.require(queue != null && queue.isQueuedSingle(task)); + if (queue.requeueSingle(task)) + setDirty(group); + } + } + + final void unqueueMulti(T task) + { + int group = group(task); + TaskQueue queue = group < 0 ? this : queue(task); + Invariants.require(queue.isQueuedSingle(task)); + unqueue(task, group, queue); + } + + // if there is an active collection, we return false and do not remove ourselves from it + boolean tryUnqueueWaiting(T task) + { + int group = group(task); + TaskQueue queue = group < 0 ? this : queue(task); + if (!queue.isQueuedSingle(task)) + return false; + + unqueue(task, group, queue); + return true; + } + + private void unqueue(T task, int group, TaskQueue queue) + { + task.unsetQueue(this); + boolean dirty = queue.unqueueSingle(task); + --waitingCount; + if (group >= 0) + { + if (queue.isEmptySingle()) + { + unsetHasWork(group); + unsetDirty(group); + } + else if (dirty) setDirty(group); + } + } + + final void incrementActive(int group) + { + active += lowBit(group); + } + + final void decrementActive(int group) + { + active -= lowBit(group); + } + + final void incrementDispatches(Task task) + { + int group = group(task); + if (group >= 0) + incrementDispatches(group); + } + + final void incrementDispatches(int group) + { + dispatches += lowBit(group); + if ((dispatches & COUNTER_OVERFLOWS) != 0) + { + dispatches = (dispatches >>> 1) & COUNTER_MASKS; + arrivals = (arrivals >>> 1) & COUNTER_MASKS; // arrivals (arrival) decays on the service/time clock + } + } + + final void decrementBudget(int group) + { + // no need to manage underflow; if the budget is being used, + // a queue should only dispatch work when there's non-zero budget + budget -= lowBit(group); + } + + final void decrementDispatches(Task task) + { + int group = group(task); + if (group >= 0) + decrementDispatches(group); + } + + final void decrementDispatches(int group) + { + long lowBit = lowBit(group); + dispatches -= lowBit; + dispatches += (dispatches >>> 7) & lowBit; + } + + final void incrementArrivals(int group) + { + int shift = group * 8; + long overflowBit = 0x80L << shift; + arrivals += 1L << shift; + // if we overflow, unset the overflow bit and set all other bits for the counter + long overflow = arrivals & overflowBit; + arrivals ^= overflow; + arrivals |= overflow - (overflow >>> 7); + } + + final void setHasWork(int group) + { + hasWork |= overflowBit(group); + } + + final void unsetHasWork(int group) + { + hasWork &= ~overflowBit(group); + } + + final void setDirty(int group) + { + dirty |= overflowBit(group); + } + + final void unsetDirty(int group) + { + dirty &= ~overflowBit(group); + } + + final boolean hasWaiting() + { + return waitingCount > 0; + } + + final boolean isWaiting(T task) + { + return queue(task).isQueuedSingle(task); + } + + final int waitingCount() + { + return waitingCount; + } + + final long lowBit(int group) + { + return 1L << (group * 8); + } + + final long overflowBit(int group) + { + return 0x80L << (group * 8); + } + } + + static final class RunnableTaskQueue extends MultiTaskQueue + { + static final int RUNNABLE = TinyEnumSet.encode(WAITING_TO_RUN, ASSIGNED, RUNNING); + + final TaskQueue assigned; + + RunnableTaskQueue() + { + super(RUNNABLE, GroupKind.GLOBAL, GLOBAL_QUEUE_BUDGETS, GLOBAL_QUEUE_LIMITS); + this.assigned = new TaskQueue<>(0); + } + + T poll() + { + T next = pollMulti(); + if (next == null) + return null; + + next.setState(ASSIGNED); + assigned.enqueueSingle(next); + + return next; + } + + void enqueue(T enqueue) + { + enqueueMulti(enqueue); + } + + void unqueue(T unqueue) + { + if (assigned.isQueuedSingle(unqueue)) + { + int group = group(unqueue); + if (group >= 0) + decrementActive(group); + + unqueue.unsetQueue(this); + assigned.unqueueSingle(unqueue); + } + else + { + super.unqueueMulti(unqueue); + } + } + + int waitingOrAssignedCount() + { + return waitingCount + assigned.size(); + } + + boolean hasAssignedOrWaiting() + { + return waitingCount > 0 || !assigned.isEmptySingle(); + } + + boolean hasAssigned() + { + return !assigned.isEmptySingle(); + } + + boolean isAssigned(T task) + { + return assigned.isQueuedSingle(task); + } + + void complete(T task) + { + if (assigned.tryUnqueueSingle(task)) + { + int group = group(task); + if (group >= 0) + decrementActive(group); + task.unsetQueue(this); + } } } static class CancelTask extends Task { final Task cancel; - private CancelTask(Task cancel) { this.cancel = cancel; } + private CancelTask(Task cancel) + { + super(GlobalGroup.OTHER, WAITING_TO_RUN); + this.cancel = cancel; + } + @Override void submitExclusive(AccordExecutor owner) { cancel.cancelExclusive(owner); } @Override protected void preRunExclusive() { throw new UnsupportedOperationException(); } @Override protected void run() { throw new UnsupportedOperationException(); } @Override protected void fail(Throwable fail) { throw new UnsupportedOperationException(); } - @Override protected void addToQueue(TaskQueue queue) { throw new UnsupportedOperationException(); } + @Override protected boolean isNewWork() { return false; } + @Override String toDescription() { return "Cancel " + cancel.toDescription(); } } static IntFunction constant(O out) @@ -1642,18 +2970,31 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor result, Runnable run, GlobalGroup group, long position, int tranche) { - this(null, run); + super(group, position, tranche); + this.result = result; + this.run = run; + this.executor = null; } - PlainRunnable(AsyncPromise result, Runnable run) + PlainRunnable(AsyncPromise result, Runnable run, GlobalGroup group) { - this(result, run, null, 0); + super(group); + this.result = result; + this.run = run; + this.executor = null; } - PlainRunnable(AsyncPromise result, Runnable run, @Nullable ExclusiveExecutor executor, long queuePosition) + PlainRunnable(AsyncPromise result, Runnable run, ExclusiveExecutor executor, ExclusiveGroup group, long position, int tranche) { + super(group, position, tranche); this.result = result; this.run = run; this.executor = executor; - this.queuePosition = queuePosition; + } + + PlainRunnable(AsyncPromise result, Runnable run, ExclusiveExecutor executor, ExclusiveGroup group) + { + super(group); + this.result = result; + this.run = run; + this.executor = executor; + } + + @Override + String toDescription() + { + // TODO (expected): ensure this is usefully descriptive, or accept a separate description + return run.toString(); } @Override @@ -1735,17 +3101,18 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor LoadRunnable newLoad(AccordCacheEntry entry, boolean isForRange) { - return isForRange ? new LoadRangeRunnable<>(entry) : new LoadRunnable<>(entry); + return new LoadRunnable<>(entry, isForRange ? RANGE_LOAD : LOAD); } class LoadRunnable extends IOTask @@ -1794,12 +3161,14 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor entry; Object result = FailureHolder.NOT_STARTED; - LoadRunnable(AccordCacheEntry entry) + LoadRunnable(AccordCacheEntry entry, GlobalGroup group) { + super(group); + Invariants.require(group == LOAD || group == RANGE_LOAD); this.entry = entry; } - boolean isForRange() { return false; } + boolean isForRange() { return is(RANGE_LOAD); } void postRunExclusive() { @@ -1807,6 +3176,12 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor extends LoadRunnable - { - LoadRangeRunnable(AccordCacheEntry entry) { super(entry); } - @Override boolean isForRange() { return true; } - } - static abstract class AbstractIOTask { abstract protected void runInternal(); @@ -1849,11 +3218,18 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor entry, UniqueSave identity, Runnable run) { + super(SAVE); this.entry = entry; this.identity = identity; this.run = run; @@ -1905,6 +3282,12 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor) task).preLoadContext().reason(); + return ((AccordTask) task).executionContext().reason(); if (task instanceof DebuggableTask) return ((DebuggableTask) task).description(); @@ -2075,7 +3463,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor) task).preLoadContext(); + return ((AccordTask) task).executionContext(); if (task instanceof WrappedIOTask && ((WrappedIOTask) task).wrapped instanceof AccordTask.RangeTxnScanner) return ((AccordTask.RangeTxnScanner) ((WrappedIOTask) task).wrapped).preLoadContext(); return null; @@ -2093,19 +3481,24 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor taskSnapshot() { List result = new ArrayList<>(); - lock(); + AccordTaskRunner self = AccordTaskRunner.get(); + lock(self); try { - addToSnapshot(result, waitingToLoad, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD); - addToSnapshot(result, waitingToLoadRangeTxns, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD); addToSnapshot(result, scanningRanges, TaskInfo.Status.SCANNING_RANGES, TaskInfo.Status.SCANNING_RANGES); - addToSnapshot(result, loading, TaskInfo.Status.LOADING, TaskInfo.Status.LOADING); - addToSnapshot(result, waitingToRun, TaskInfo.Status.WAITING_TO_RUN, TaskInfo.Status.WAITING_TO_RUN); - addToSnapshot(result, running, TaskInfo.Status.RUNNING, TaskInfo.Status.WAITING_TO_RUN); + addToSnapshot(result, waitingToLoadRangeTxns, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD); + addToSnapshot(result, waitingToLoad, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD); + addToSnapshot(result, loading, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD); + for (TaskQueue queue : runnable.queues) + { + if (queue != null) + addToSnapshot(result, queue, TaskInfo.Status.WAITING_TO_RUN, TaskInfo.Status.WAITING_TO_RUN); + } + addToSnapshot(result, runnable.assigned, TaskInfo.Status.RUNNING, TaskInfo.Status.WAITING_TO_RUN); } finally { - unlock(); + unlock(self); } result.sort(TaskInfo::compareTo); return result; @@ -2115,13 +3508,19 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor q0 : q.queues) + { + if (q0 != null) + { + for (int j = 0 ; j < q0.size() ; ++j) + snapshot.add(new TaskInfo(ifQueued, q.commandStoreId, q0.getSingle(j))); + } + } } else { @@ -2133,17 +3532,65 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor> get, String input, String describe) + { + long result = 0; + for (String spec : input.split(",")) + { + if (spec.trim().isEmpty()) continue; + + String[] split = spec.split(":"); + if (split.length != 2) + throw new IllegalArgumentException("Invalid specifier " + spec + " in " + describe + ": " + input); + + try + { + Enum queue = get.apply(split[0]); + long value = Long.parseLong(split[1]); + if (value <= 0 || value >= 128) + throw new IllegalArgumentException("Invalid limit " + value + " in queue_active_limits: " + input); + + result |= value << (queue.ordinal() * 8); + } + catch (Throwable t) + { + throw new IllegalArgumentException("Invalid queue identifier " + split[0] + " in " + describe + ": " + input); + } + } + + return result; + } + + private static long encodeCounters(Map, Long> counters) + { + long result = 0; + for (Map.Entry, Long> e : counters.entrySet()) + result |= e.getValue() << (e.getKey().ordinal() * 8); + return result; } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java index d8051f40dc..8e3e41f7c3 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java +++ b/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 void submitExternal(QuintConsumer sync, QuadFunction async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4); - void submit(QuintConsumer sync, QuadFunction async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4) + final void submit(QuintConsumer sync, QuadFunction 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); } - void submitExternalExclusive(QuintConsumer sync, P1s p1s, P2 p2, P3 p3, P4 p4) + final void submitExternalExclusive(QuintConsumer 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) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLoop.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLoop.java index a76d1bac18..3cbd3db144 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLoop.java +++ b/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 unqueuedUpdater = AtomicReferenceFieldUpdater.newUpdater(AccordExecutorAbstractLoop.class, Task.class, "unqueued"); + volatile Task unqueued; + static final AtomicReferenceFieldUpdater 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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSemiSyncSubmit.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSemiSyncSubmit.java index f97e206ef5..67e897d479 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSemiSyncSubmit.java +++ b/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 { diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSignalLoop.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSignalLoop.java index 9ec7fa95d2..ce7de7fefc 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSignalLoop.java +++ b/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); + } } } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java index ab6c8c25e7..ee37744d8d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java +++ b/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() { diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSyncSubmit.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSyncSubmit.java index 0d852ccde3..e4efac3ad4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSyncSubmit.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorSyncSubmit.java @@ -91,14 +91,15 @@ public class AccordExecutorSyncSubmit extends AccordExecutorAbstractLockLoop void submitExternal(QuintConsumer sync, QuadFunction 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); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java index 45aeaa2f18..e268113b86 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java +++ b/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 diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 6a9817bc97..d0399e1b92 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -64,7 +64,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore KeyBarriers.await(node, node.someSequentialExecutor(), found, syncLocal, syncRemote)) + .flatMap(found -> KeyBarriers.await(node, node.someExclusiveExecutor(), found, syncLocal, syncRemote)) .flatMap(success -> { if (success) return null; diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index fe2889e2a6..90781c740f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/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 extends Task implements Function function; - public ForFunction(AccordCommandStore commandStore, ExecutionContext loadCtx, Function function) + public ForFunction(AccordCommandStore commandStore, ExecutionContext context, Function function) { - super(commandStore, loadCtx); + super(commandStore, context); this.function = function; } @@ -129,9 +127,9 @@ public abstract class AccordTask extends Task implements Function consumer; - private ForConsumer(AccordCommandStore commandStore, ExecutionContext loadCtx, Consumer consumer) + private ForConsumer(AccordCommandStore commandStore, ExecutionContext context, Consumer consumer) { - super(commandStore, loadCtx); + super(commandStore, context); this.consumer = consumer; } @@ -143,63 +141,16 @@ public abstract class AccordTask extends Task implements Function AccordTask create(CommandStore commandStore, ExecutionContext ctx, Function function) + public static AccordTask create(CommandStore commandStore, ExecutionContext context, Function function) { - return new ForFunction<>((AccordCommandStore) commandStore, ctx, function); + return new ForFunction<>((AccordCommandStore) commandStore, context, function); } - public static AccordTask create(CommandStore commandStore, ExecutionContext ctx, Consumer consumer) + public static AccordTask create(CommandStore commandStore, ExecutionContext context, Consumer 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 extends Task implements Function> waitingToLoad; @Nullable RangeTxnScanner rangeScanner; - boolean hasRanges; @Nullable CommandSummaries commandsForRanges; - @Nullable private TaskQueue queued; private BiConsumer 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 extends Task implements Function extends Task implements Function no loading or waiting; found %s", this, AccordTask::toDescription); - } - } - Unseekables keys() { return executionContext.keys(); @@ -334,20 +273,6 @@ public abstract class AccordTask extends Task implements Function priorityChain() - { - return new AsyncChains.Head<>() - { - @Override - protected Cancellable start(BiConsumer callback) - { - preSetup(callback); - commandStore.executor().submitPriority(AccordTask.this); - return AccordTask.this; - } - }; - } - private void preSetup(BiConsumer callback) { Invariants.require(this.callback == null); @@ -358,7 +283,7 @@ public abstract class AccordTask extends Task implements Function 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 extends Task implements Function extends Task implements Function extends Task implements Function extends Task implements Function extends Task implements Function> 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 extends Task implements Function 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 extends Task implements Function extends Task implements Function extends Task implements Function finish(result, null); safeStore.persistFieldUpdatesInternal(changes == null ? onFlush : null); if (changes != null) save(changes, onFlush); @@ -735,12 +657,12 @@ public abstract class AccordTask extends Task implements Function extends Task implements Function extends Task implements Function extends Task implements Function extends Task implements Function extends Task implements Function 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 @@ -965,7 +851,6 @@ public abstract class AccordTask extends Task implements Function intersectingKeys = new ObjectHashSet<>(); final KeyWatcher keyWatcher = new KeyWatcher(); final Ranges ranges = ((AbstractRanges) executionContext.keys()).toRanges(); @@ -1102,9 +987,9 @@ public abstract class AccordTask extends Task implements Function extends Task implements Function= 1 && increment <= MAX_SIGNAL_COUNT); while (true) { long cur = state; diff --git a/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java b/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java index de181ef688..e1a50740ef 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java @@ -401,10 +401,13 @@ public class ActionSchedule implements CloseableIterator, 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()); diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java index eadb7a4f55..52a2b85409 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java +++ b/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>> 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> start() + { + ConcurrentLinkedQueue> 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 + { + 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> consequences, Consumer>> 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 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> 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> 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> await = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> awaitConsequences = new ConcurrentLinkedQueue<>(); while (outerLoop-- > 0) { + List>> allAwaitSubmitted = new ArrayList<>(); for (int i = 0; i < innerLoop; ++i) - submitRecursive(lock, executor, sequentialExecutor, 1 + i, await, sleepChance, lockChance); + { + Collection> 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> 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> await, float sleepChance, float lockChance) + private static void submitRecursive(Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, int count, Collection> consequences, Collection> 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 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> await, @Nullable Class ignore) throws InterruptedException, ExecutionException + { + for (Future future : await) + { + try { future.get(); } + catch (ExecutionException e) + { + if (ignore == null || !(ignore.isInstance(e.getCause()))) + throw e; + } + } + } } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index a9a5c6db58..5d6408f676 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/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; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index dcd3486852..3509c2e506 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/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; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java index ede89a4d77..54dee77c04 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java +++ b/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; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index fc4fdd4367..ae43676851 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/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; } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 228e2e564a..00705915dd 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/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; } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 5130dce553..9975fd6b2b 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/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; } diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index 0e7d7a3685..a05b644b53 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/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: