From cba42085687d3b1476ec1715261ce20464358431 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Sat, 5 Jul 2025 09:58:59 +0100 Subject: [PATCH] Standardise Replay logic to ensure SafeCommandStore.update is called Also Fix: - Initialise home state when calling waiting() if not already initialised - Don't reportClosed/reportRetired for epochs that are already closed/retired Also Improve: - Implement waitForQuiescence in AccordExecutor - Permit replay parallelism patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20750 --- modules/accord | 2 +- .../org/apache/cassandra/journal/Journal.java | 7 +- .../service/accord/AccordCommandStore.java | 21 +++--- .../service/accord/AccordCommandStores.java | 38 +--------- .../service/accord/AccordExecutor.java | 35 +++++++++ .../AccordExecutorAbstractLockLoop.java | 6 +- .../service/accord/AccordExecutorSimple.java | 5 +- .../service/accord/AccordJournal.java | 72 +++++++++++-------- .../cassandra/service/accord/AccordTask.java | 2 - 9 files changed, 103 insertions(+), 85 deletions(-) diff --git a/modules/accord b/modules/accord index 71d235d56c..e587fc4c09 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 71d235d56cb315fa5ae01ec24d3d9f08dd08ac6a +Subproject commit e587fc4c090aeb74a363af43295e0f7c7825445a diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 727360a3f0..d07cae1c59 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -946,7 +946,7 @@ public class Journal implements Shutdownable */ public static class KeyRefs { - long segments[]; + long[] segments; K key; int size; @@ -966,6 +966,11 @@ public class Journal implements Shutdownable consumer.accept(segments[i]); } + public long[] copyOfSegments() + { + return segments == null ? new long[0] : Arrays.copyOf(segments, size); + } + public K key() { return key; diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index e166b94b87..7c1e66857e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -45,7 +45,6 @@ import accord.impl.AbstractSafeCommandStore.CommandStoreCaches; import accord.local.Command; import accord.local.CommandStore; import accord.local.CommandStores; -import accord.local.Commands; import accord.local.NodeCommandStoreService; import accord.local.PreLoadContext; import accord.local.RedundantBefore; @@ -55,6 +54,7 @@ import accord.primitives.PartialTxn; import accord.primitives.RangeDeps; import accord.primitives.Ranges; import accord.primitives.RoutableKey; +import accord.primitives.Route; import accord.primitives.Status; import accord.primitives.Timestamp; import accord.primitives.TxnId; @@ -71,7 +71,6 @@ import org.apache.cassandra.utils.Clock; import static accord.api.Journal.CommandUpdate; import static accord.api.Journal.FieldUpdates; import static accord.api.Journal.Load.MINIMAL; -import static accord.api.Journal.Loader; import static accord.utils.Invariants.require; public class AccordCommandStore extends CommandStore @@ -153,7 +152,7 @@ public class AccordCommandStore extends CommandStore private AccordSafeCommandStore current; - private final CommandStoreLoader loader; + private final AccordCommandStoreLoader loader; public AccordCommandStore(int id, NodeCommandStoreService node, @@ -182,7 +181,7 @@ public class AccordCommandStore extends CommandStore this.exclusiveExecutor = sharedExecutor.executor(); this.commandsForRanges = new CommandsForRanges.Manager(this); - this.loader = new CommandStoreLoader(this); + this.loader = new AccordCommandStoreLoader(this); maybeLoadRedundantBefore(journal.loadRedundantBefore(id())); maybeLoadBootstrapBeganAt(journal.loadBootstrapBeganAt(id())); @@ -485,7 +484,7 @@ public class AccordCommandStore extends CommandStore return rangeSearcher; } - public Loader loader() + public AccordCommandStoreLoader loader() { return loader; } @@ -496,23 +495,21 @@ public class AccordCommandStore extends CommandStore super.unsafeUpsertRedundantBefore(addRedundantBefore); } - private static class CommandStoreLoader extends AbstractLoader + public static class AccordCommandStoreLoader extends AbstractLoader { private final AccordCommandStore store; - private CommandStoreLoader(AccordCommandStore store) + private AccordCommandStoreLoader(AccordCommandStore store) { this.store = store; } @Override - public AsyncChain load(TxnId txnId) + public AsyncChain load(TxnId txnId) { return store.submit(txnId, safeStore -> { - maybeApplyWrites(txnId, safeStore, (safeCommand, cmd) -> { - Commands.applyWrites(safeStore, txnId, cmd).begin(store.agent); - }); - return safeStore.unsafeGet(txnId).current(); + maybeApplyWrites(safeStore, txnId); + return safeStore.unsafeGet(txnId).current().route(); }); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index 11276a7a6c..3fa4bb0bdc 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -17,11 +17,8 @@ */ package org.apache.cassandra.service.accord; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import accord.api.Agent; @@ -46,7 +43,6 @@ import org.apache.cassandra.metrics.CacheSizeMetrics; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory; import org.apache.cassandra.service.accord.api.TokenKey; -import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static org.apache.cassandra.config.AccordSpec.QueueShardModel.THREAD_PER_SHARD; import static org.apache.cassandra.config.DatabaseDescriptor.getAccordQueueShardCount; @@ -229,38 +225,8 @@ public class AccordCommandStores extends CommandStores implements CacheSize public void waitForQuiescense() { - boolean runAgain = true; - try - { - while (true) - { - boolean hasTasks = false; - List> futures = new ArrayList<>(); - for (AccordExecutor executor : this.executors) - { - hasTasks |= executor.hasTasks(); - hasTasks |= Stage.MUTATION.executor().getPendingTaskCount() > 0; - hasTasks |= Stage.MUTATION.executor().getActiveTaskCount() > 0; - futures.add(executor.submit(() -> {})); - } - for (Future future : futures) - future.get(); - futures.clear(); - - if (!runAgain) - return; - - runAgain = hasTasks; - } - } - catch (ExecutionException e) - { - throw new IllegalStateException("Should have never been thrown", e); - } - catch (InterruptedException e) - { - throw new UncheckedInterruptedException(e); - } + for (AccordExecutor executor : this.executors) + executor.waitForQuiescence(); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index c61f5e7a9b..1ece0d3c3c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -18,6 +18,8 @@ package org.apache.cassandra.service.accord; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.CancellationException; import java.util.concurrent.TimeUnit; @@ -62,6 +64,7 @@ import org.apache.cassandra.concurrent.Shutdownable; import org.apache.cassandra.metrics.AccordCacheMetrics; import org.apache.cassandra.utils.MonotonicClock; import org.apache.cassandra.utils.concurrent.AsyncPromise; +import org.apache.cassandra.utils.concurrent.Condition; import org.apache.cassandra.utils.concurrent.Future; import static org.apache.cassandra.service.accord.AccordCache.CommandAdapter.COMMAND_ADAPTER; @@ -138,6 +141,8 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo private final AccordCacheEntry.OnLoaded onRangeLoaded = this::onRangeLoaded; private final ExclusiveGlobalCaches caches; + private List waitingForQuiescence; + /** * The maximum total number of loads we can queue at once - this includes loads for range transactions, * which are subject to this limit as well as that imposed by {@link #maxQueuedRangeLoads} @@ -224,6 +229,36 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo return Stream.of(); } + public void waitForQuiescence() + { + Condition condition; + lock.lock(); + try + { + if (tasks == 0 && running == 0) + return; + + if (waitingForQuiescence == null) + waitingForQuiescence = new ArrayList<>(); + condition = Condition.newOneTimeCondition(); + waitingForQuiescence.add(condition); + } + finally + { + lock.unlock(); + } + condition.awaitThrowUncheckedOnInterrupt(); + } + + protected void signalQuiescentExclusive() + { + if (waitingForQuiescence != null) + { + waitingForQuiescence.forEach(Condition::signalAll); + waitingForQuiescence = null; + } + } + void maybeUnpauseLoading() { if (!hasPausedLoading) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java index f442473a46..63fe294439 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java @@ -125,7 +125,8 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor private void pauseExclusive() { isHeldByExecutor = false; - --running; + if (--running == 0 && tasks == 0) + signalQuiescentExclusive(); } private void resumeExclusive() @@ -234,6 +235,8 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor break; } + pauseExclusive(); + if (shutdown) { exitLockExclusive(); @@ -241,7 +244,6 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor return; } - pauseExclusive(); awaitExclusive(); resumeExclusive(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java index ac0069fdc9..7e74e09b32 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java @@ -70,7 +70,6 @@ class AccordExecutorSimple extends AccordExecutor Invariants.requireArgument(threads == 1); this.lock = lock; this.executor = executorFactory().sequential(name.apply(0)); - } @Override @@ -89,7 +88,11 @@ class AccordExecutorSimple extends AccordExecutor { Task task = pollWaitingToRunExclusive(); if (task == null) + { + running = 0; + signalQuiescentExclusive(); return; + } --tasks; try { task.preRunExclusive(null); task.run(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index c597c11f40..aa1d8425c0 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.NavigableMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import java.util.function.Supplier; import javax.annotation.Nullable; @@ -52,7 +53,6 @@ import accord.primitives.TxnId; import accord.utils.Invariants; import accord.utils.PersistentField; import accord.utils.UnhandledEnum; -import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; import accord.utils.async.AsyncResults; import org.apache.cassandra.concurrent.Shutdownable; @@ -70,6 +70,7 @@ import org.apache.cassandra.journal.RecordPointer; import org.apache.cassandra.journal.SegmentCompactor; import org.apache.cassandra.journal.StaticSegment; import org.apache.cassandra.journal.ValueSerializer; +import org.apache.cassandra.service.accord.AccordCommandStore.AccordCommandStoreLoader; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightImage; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator; import org.apache.cassandra.service.accord.JournalKey.JournalKeySupport; @@ -82,6 +83,8 @@ import org.apache.cassandra.service.accord.serializers.Version; import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer; import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.ExecutorUtils; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.concurrent.Semaphore; import static accord.impl.CommandChange.Field.CLEANUP; import static accord.impl.CommandChange.anyFieldChanged; @@ -484,43 +487,52 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @Override public void replay(CommandStores commandStores) { + final Semaphore concurrency = Semaphore.newSemaphore(FBUtilities.getAvailableProcessors()); + final AtomicBoolean abort = new AtomicBoolean(); + try (CloseableIterator> iter = journalTable.keyIterator()) { JournalKey prev = null; while (iter.hasNext()) { - Journal.KeyRefs ref = iter.next(); + if (abort.get()) + break; - if (ref.key().type != JournalKey.Type.COMMAND_DIFF) - continue; - CommandStore commandStore = commandStores.forId(ref.key().commandStoreId); - Loader loader = commandStore.loader(); - TxnId txnId = ref.key().id; - try + JournalKey key; + long[] segments; { - Invariants.require(prev == null || - ref.key().commandStoreId != prev.commandStoreId || - ref.key().id.compareTo(prev.id) != 0, - "duplicate key detected %s == %s", ref.key(), prev); - prev = ref.key(); - AsyncChains.getUnchecked(loader.load(txnId) - .map(command -> { - if (journalTable.shouldIndex(ref.key()) - && command.participants() != null - && command.participants().route() != null) - { - ref.segments(segment -> { - journalTable.safeNotify(index -> index.update(segment, ref.key().commandStoreId, txnId, command.participants().route())); - }); - } - return command; - }) - .beginAsResult()); - } - catch (Throwable t) - { - journal.handleError("Could not replay command " + ref.key().id, t); + Journal.KeyRefs ref = iter.next(); + key = ref.key(); + if (key.type != JournalKey.Type.COMMAND_DIFF) + continue; + + segments = journalTable.shouldIndex(key) ? ref.copyOfSegments() : null; } + + CommandStore commandStore = commandStores.forId(key.commandStoreId); + AccordCommandStoreLoader loader = (AccordCommandStoreLoader) commandStore.loader(); + + TxnId txnId = key.id; + Invariants.require(prev == null || + key.commandStoreId != prev.commandStoreId || + key.id.compareTo(prev.id) != 0, + "duplicate key detected %s == %s", key, prev); + prev = key; + + concurrency.acquireThrowUncheckedOnInterrupt(1); + loader.load(txnId) + .map(route -> { + if (segments != null) + { + for (long segment : segments) + journalTable.safeNotify(index -> index.update(segment, key.commandStoreId, txnId, route)); + } + return null; + }).begin((success, fail) -> { + concurrency.release(1); + if (fail != null && !journal.handleError("Could not replay command " + txnId, fail)) + abort.set(true); + }); } } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index f037ab8df1..3d54a289b1 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -42,7 +42,6 @@ import accord.api.Journal; import accord.api.RoutingKey; import accord.local.Command; import accord.local.CommandStore; -import accord.local.MaxDecidedRX; import accord.local.PreLoadContext; import accord.local.SafeCommandStore; import accord.local.cfk.CommandsForKey; @@ -1092,7 +1091,6 @@ public abstract class AccordTask extends SubmittableTask implements Runnable, void startInternal(Caches caches) { - MaxDecidedRX maxDecidedRX = commandStore.unsafeGetMaxDecidedRX(); summaryLoader = commandStore.commandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.keyHistory(), keysOrRanges); summaryLoader.forEachInCache(keysOrRanges, summary -> summaries.put(summary.txnId, summary), caches); caches.commands().register(commandWatcher);