natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Benedict Elliott Smith 2025-07-05 09:58:59 +01:00
parent a866f92fcf
commit cba4208568
9 changed files with 103 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 71d235d56cb315fa5ae01ec24d3d9f08dd08ac6a
Subproject commit e587fc4c090aeb74a363af43295e0f7c7825445a

7
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -946,7 +946,7 @@ public class Journal<K, V> implements Shutdownable
*/
public static class KeyRefs<K>
{
long segments[];
long[] segments;
K key;
int size;
@ -966,6 +966,11 @@ public class Journal<K, V> implements Shutdownable
consumer.accept(segments[i]);
}
public long[] copyOfSegments()
{
return segments == null ? new long[0] : Arrays.copyOf(segments, size);
}
public K key()
{
return key;

21
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -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<Command> load(TxnId txnId)
public AsyncChain<Route> 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();
});
}
}

38
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -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<Future<?>> 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

35
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

@ -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<Condition> 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)

6
src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
View File

@ -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();
}

5
src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java
View File

@ -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(); }

72
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -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<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator())
{
JournalKey prev = null;
while (iter.hasNext())
{
Journal.KeyRefs<JournalKey> 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<JournalKey> 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);
});
}
}
}

2
src/java/org/apache/cassandra/service/accord/AccordTask.java
View File

@ -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<R> 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);