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

rename PreLoadContext to ExecutionContext, and SequentialAsyncExecutor -> ExclusiveAsyncExecutor

This commit is contained in:
Benedict Elliott Smith 2026-07-09 11:25:51 +01:00
parent bfcdf927f8
commit 13956ddf21
25 changed files with 159 additions and 161 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitmodules vendored
View File

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

2
doc/modules/cassandra/pages/architecture/accord-architecture.adoc
View File

@ -87,7 +87,7 @@ finally {
}
....
In other words, `CommandStore` collects the `PreLoadContext`, state
In other words, `CommandStore` collects the `ExecutionContext`, state
required to be in memory for command execution (possible dependencies,
such as `TxnId`s, and `Key`s of commands, but also `CommandsForKeys`
that will be needed during execution). Once the context is collected and

18
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -71,9 +71,9 @@ import accord.local.CommandStores.LatentStoreSelector;
import accord.local.Commands;
import accord.local.Commands.NotifyWaitingOnPlus;
import accord.local.DurableBefore;
import accord.local.ExecutionContext;
import accord.local.MaxConflicts;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
@ -337,16 +337,16 @@ public class AccordDebugKeyspace extends VirtualKeyspace
if (prev != null && info.status() == prev.status() && info.position() == prev.position()) ++uniquePos;
else uniquePos = 0;
prev = info;
PreLoadContext preLoadContext = info.preLoadContext();
ExecutionContext executionContext = info.preLoadContext();
rows.add(info.status().name(), info.position(), uniquePos)
.lazyCollect(columns -> {
columns.add("description", info.describe())
.add("command_store_id", info.commandStoreId())
.add("txn_id", preLoadContext, PreLoadContext::primaryTxnId, TO_STRING)
.add("txn_id_additional", preLoadContext, PreLoadContext::additionalTxnId, TO_STRING)
.add("keys", preLoadContext, PreLoadContext::keys, TO_STRING)
.add("keys_loading", preLoadContext, PreLoadContext::loadKeys, TO_STRING)
.add("keys_loading_for", preLoadContext, PreLoadContext::loadKeysFor, TO_STRING);
.add("txn_id", executionContext, ExecutionContext::primaryTxnId, TO_STRING)
.add("txn_id_additional", executionContext, ExecutionContext::additionalTxnId, TO_STRING)
.add("keys", executionContext, ExecutionContext::keys, TO_STRING)
.add("keys_loading", executionContext, ExecutionContext::loadKeys, TO_STRING)
.add("keys_loading_for", executionContext, ExecutionContext::loadKeysFor, TO_STRING);
});
}
});
@ -740,7 +740,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
collector.partition(commandStore.id())
.collect(rows -> {
// TODO (desired): support maybe execute immediately with safeStore
Future<?> future = toFuture(commandStore.chain((PreLoadContext.Empty) metadata::toString, safeStore -> { addRows(safeStore, rows); }));
Future<?> future = toFuture(commandStore.chain((ExecutionContext.Empty) metadata::toString, safeStore -> { addRows(safeStore, rows); }));
if (!future.awaitUntilThrowUncheckedOnInterrupt(collector.deadlineNanos()))
throw new InternalTimeoutException();
});
@ -1895,7 +1895,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
AccordService.getBlocking(accord.node()
.commandStores()
.forId(commandStoreId)
.chain(PreLoadContext.contextFor(txnId, TXN_OPS), apply)
.chain(ExecutionContext.contextFor(txnId, TXN_OPS), apply)
.flatMap(i -> i));
}

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

@ -58,13 +58,13 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.CommandSummaries;
import accord.local.ExecutionContext;
import accord.local.MaxConflicts;
import accord.local.MaxDecidedRX;
import accord.local.MinimalCommand;
import accord.local.MinimalCommand.MinimalWithDeps;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.PreLoadContext.Empty;
import accord.local.ExecutionContext.Empty;
import accord.local.RedundantBefore;
import accord.local.RedundantBefore.Bounds;
import accord.local.RedundantStatus.Property;
@ -216,7 +216,7 @@ public class AccordCommandStore extends CommandStore
public final String loggingId;
public final Journal journal;
private final AccordExecutor sharedExecutor;
final AccordExecutor.SequentialExecutor exclusiveExecutor;
final AccordExecutor.ExclusiveExecutor exclusiveExecutor;
private final ExclusiveCaches caches;
private final RangeIndex rangeIndex;
private final TableId tableId;
@ -403,27 +403,27 @@ public class AccordCommandStore extends CommandStore
return lastSystemTimestampMicros.accumulateAndGet(node.now(), (a, b) -> Math.max(a + 1, b));
}
@Override
public <T> AsyncChain<T> chain(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
public <T> AsyncChain<T> chain(ExecutionContext loadCtx, Function<? super SafeCommandStore, T> function)
{
return AccordTask.create(this, loadCtx, function).chain();
}
@Override
public AsyncChain<Void> chain(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
public AsyncChain<Void> chain(ExecutionContext executionContext, Consumer<? super SafeCommandStore> consumer)
{
return AccordTask.create(this, preLoadContext, consumer).chain();
return AccordTask.create(this, executionContext, consumer).chain();
}
@Override
public AsyncChain<Void> priorityChain(PreLoadContext preLoadContext, Consumer<? super SafeCommandStore> consumer)
public AsyncChain<Void> priorityChain(ExecutionContext executionContext, Consumer<? super SafeCommandStore> consumer)
{
return AccordTask.create(this, preLoadContext, consumer).priorityChain();
return AccordTask.create(this, executionContext, consumer).priorityChain();
}
@Override
public <T> AsyncChain<T> priorityChain(PreLoadContext preLoadContext, Function<? super SafeCommandStore, T> function)
public <T> AsyncChain<T> priorityChain(ExecutionContext executionContext, Function<? super SafeCommandStore, T> function)
{
return AccordTask.create(this, preLoadContext, function).priorityChain();
return AccordTask.create(this, executionContext, function).priorityChain();
}
@Override
@ -729,7 +729,7 @@ public class AccordCommandStore extends CommandStore
if (!maybeShouldReplay(txnId))
return AsyncChains.success(null);
return commandStore.chain(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Replay"), safeStore -> {
Replay replay = shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants());
if (replay == Replay.NONE)
return null;

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

@ -36,7 +36,7 @@ import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.local.CommandStores;
import accord.local.NodeCommandStoreService;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.local.ShardDistributor;
import accord.utils.RandomSource;
import accord.utils.Reduce;
@ -148,7 +148,7 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
}
@Override
public SequentialAsyncExecutor someSequentialExecutor()
public ExclusiveAsyncExecutor someSequentialExecutor()
{
int idx = ((int) Thread.currentThread().getId()) & mask;
return executors[idx].newSequentialExecutor();

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

@ -44,8 +44,7 @@ import accord.api.Agent;
import accord.api.RoutingKey;
import accord.impl.AbstractAsyncExecutor;
import accord.local.Command;
import accord.local.PreLoadContext;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExecutionContext;
import accord.local.cfk.CommandsForKey;
import accord.messages.Accept;
import accord.messages.Commit;
@ -611,25 +610,25 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
task.addToQueue(task.commandStore.exclusiveExecutor);
}
private void waitingToRun(Task task, @Nullable SequentialExecutor queue)
private void waitingToRun(Task task, @Nullable ExclusiveExecutor queue)
{
task.onWaitingToRun();
task.addToQueue(queue == null ? waitingToRun : queue);
}
public SequentialExecutor executor()
public ExclusiveExecutor executor()
{
return new SequentialExecutor(this);
return new ExclusiveExecutor(this);
}
public SequentialExecutor executor(int commandStoreId)
public ExclusiveExecutor executor(int commandStoreId)
{
return new SequentialExecutor(this, commandStoreId);
return new ExclusiveExecutor(this, commandStoreId);
}
public SequentialAsyncExecutor newSequentialExecutor()
public accord.local.ExclusiveAsyncExecutor newSequentialExecutor()
{
return new SequentialExecutor(this);
return new ExclusiveExecutor(this);
}
public <R> void cancel(AccordTask<R> task)
@ -751,7 +750,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
case PHASE_HLC_FIFO:
{
// TODO (expected): we should process messages for a TxnId together, to avoid processing delayed messages out of order
PreLoadContext context = task.preLoadContext();
ExecutionContext context = task.preLoadContext();
if (context instanceof Request)
{
MessageType type = ((Request) context).type();
@ -1207,7 +1206,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
// run the task even on a stopped commandStore
public interface Unstoppable extends PreLoadContext.Empty
public interface Unstoppable extends ExecutionContext.Empty
{
}
@ -1218,9 +1217,9 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
static final class SequentialQueueTask extends Task
{
private final SequentialExecutor queue;
private final ExclusiveExecutor queue;
SequentialQueueTask(SequentialExecutor queue)
SequentialQueueTask(ExclusiveExecutor queue)
{
super();
this.queue = queue;
@ -1265,8 +1264,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
}
private static final AtomicReferenceFieldUpdater<SequentialExecutor, Thread> ownerUpdater = AtomicReferenceFieldUpdater.newUpdater(SequentialExecutor.class, Thread.class, "owner");
public class SequentialExecutor extends TaskQueue<Task> implements SequentialAsyncExecutor
private static final AtomicReferenceFieldUpdater<ExclusiveExecutor, Thread> ownerUpdater = AtomicReferenceFieldUpdater.newUpdater(ExclusiveExecutor.class, Thread.class, "owner");
public class ExclusiveExecutor extends TaskQueue<Task> implements accord.local.ExclusiveAsyncExecutor
{
final int commandStoreId;
final SequentialQueueTask selfTask;
@ -1278,12 +1277,12 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
final DebugSequentialExecutor debug;
SequentialExecutor(AccordExecutor executor)
ExclusiveExecutor(AccordExecutor executor)
{
this(executor, -1);
}
SequentialExecutor(AccordExecutor executor, int commandStoreId)
ExclusiveExecutor(AccordExecutor executor, int commandStoreId)
{
super(WAITING_TO_RUN, commandStoreId < 0);
this.commandStoreId = commandStoreId;
@ -1331,7 +1330,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
if (!(task instanceof AccordTask<?>))
return true;
PreLoadContext context = ((AccordTask<?>) task).preLoadContext();
ExecutionContext context = ((AccordTask<?>) task).preLoadContext();
return !(terminated ? (context instanceof Unterminatable) : (context instanceof Unstoppable));
}
@ -1522,7 +1521,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
private Cancellable execute(RunOrFail runOrFail, long queuePosition)
{
PlainChain submit = new PlainChain(runOrFail, SequentialExecutor.this, queuePosition);
PlainChain submit = new PlainChain(runOrFail, ExclusiveExecutor.this, queuePosition);
return AccordExecutor.this.submit(submit);
}
@ -1643,7 +1642,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
abstract class Plain extends Task implements Cancellable
{
abstract SequentialExecutor executor();
abstract ExclusiveExecutor executor();
@Override
protected void preRunExclusive() {}
@ -1663,7 +1662,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
void cancelExclusive(AccordExecutor owner)
{
SequentialExecutor executor = executor();
ExclusiveExecutor executor = executor();
TaskQueue queue = executor == null ? waitingToRun : executor;
if (queue.contains(this))
{
@ -1685,7 +1684,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
{
final @Nullable AsyncPromise<Void> result;
final Runnable run;
final @Nullable SequentialExecutor executor;
final @Nullable ExclusiveExecutor executor;
PlainRunnable(Runnable run)
{
@ -1697,7 +1696,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
this(result, run, null, 0);
}
PlainRunnable(AsyncPromise<Void> result, Runnable run, @Nullable SequentialExecutor executor, long queuePosition)
PlainRunnable(AsyncPromise<Void> result, Runnable run, @Nullable ExclusiveExecutor executor, long queuePosition)
{
this.result = result;
this.run = run;
@ -1727,7 +1726,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
SequentialExecutor executor()
ExclusiveExecutor executor()
{
return executor;
}
@ -1755,7 +1754,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
SequentialExecutor executor()
ExclusiveExecutor executor()
{
return null;
}
@ -1934,14 +1933,14 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
class PlainChain extends Plain
{
final RunOrFail runOrFail;
final @Nullable SequentialExecutor executor;
final @Nullable ExclusiveExecutor executor;
PlainChain(RunOrFail runOrFail)
{
this(runOrFail, null, 0);
}
PlainChain(RunOrFail runOrFail, @Nullable SequentialExecutor executor, long queuePosition)
PlainChain(RunOrFail runOrFail, @Nullable ExclusiveExecutor executor, long queuePosition)
{
this.runOrFail = runOrFail;
this.executor = executor;
@ -1949,7 +1948,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
SequentialExecutor executor()
ExclusiveExecutor executor()
{
return executor;
}
@ -1991,7 +1990,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
long startedAtNanos;
final Object describe;
DebuggableChain(RunOrFail runOrFail, @Nullable SequentialExecutor executor, int queuePosition, Object describe)
DebuggableChain(RunOrFail runOrFail, @Nullable ExclusiveExecutor executor, int queuePosition, Object describe)
{
super(runOrFail, executor, queuePosition);
this.createdAtNanos = MonotonicClock.Global.approxTime.now();
@ -2073,7 +2072,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return null;
}
public @Nullable PreLoadContext preLoadContext()
public @Nullable ExecutionContext preLoadContext()
{
if (task instanceof AccordTask)
return ((AccordTask<?>) task).preLoadContext();
@ -2119,7 +2118,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
Task t = queue.get(i);
if (t instanceof SequentialQueueTask)
{
SequentialExecutor q = ((SequentialQueueTask) t).queue;
ExclusiveExecutor q = ((SequentialQueueTask) t).queue;
snapshot.add(new TaskInfo(ifCurrent, q.commandStoreId, q.task));
for (int j = 0 ; j < q.size() ; ++j)
snapshot.add(new TaskInfo(ifQueued, q.commandStoreId, q.get(j)));

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

@ -49,8 +49,8 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandSummaries;
import accord.local.CommandSummaries.Summary;
import accord.local.ExecutionContext;
import accord.local.LoadKeys;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.AbstractRanges;
@ -111,7 +111,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
{
private final Function<? super SafeCommandStore, R> function;
public ForFunction(AccordCommandStore commandStore, PreLoadContext loadCtx, Function<? super SafeCommandStore, R> function)
public ForFunction(AccordCommandStore commandStore, ExecutionContext loadCtx, Function<? super SafeCommandStore, R> function)
{
super(commandStore, loadCtx);
this.function = function;
@ -129,7 +129,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
{
private final Consumer<? super SafeCommandStore> consumer;
private ForConsumer(AccordCommandStore commandStore, PreLoadContext loadCtx, Consumer<? super SafeCommandStore> consumer)
private ForConsumer(AccordCommandStore commandStore, ExecutionContext loadCtx, Consumer<? super SafeCommandStore> consumer)
{
super(commandStore, loadCtx);
this.consumer = consumer;
@ -143,12 +143,12 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
}
}
public static <T> AccordTask<T> create(CommandStore commandStore, PreLoadContext ctx, Function<? super SafeCommandStore, T> function)
public static <T> AccordTask<T> create(CommandStore commandStore, ExecutionContext ctx, Function<? super SafeCommandStore, T> function)
{
return new ForFunction<>((AccordCommandStore) commandStore, ctx, function);
}
public static AccordTask<Void> create(CommandStore commandStore, PreLoadContext ctx, Consumer<? super SafeCommandStore> consumer)
public static AccordTask<Void> create(CommandStore commandStore, ExecutionContext ctx, Consumer<? super SafeCommandStore> consumer)
{
return new ForConsumer((AccordCommandStore) commandStore, ctx, consumer);
}
@ -201,7 +201,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
private State state = INITIALIZED;
final AccordCommandStore commandStore;
private final PreLoadContext preLoadContext;
private final ExecutionContext executionContext;
private volatile String loggingId;
private static final AtomicLong nextLoggingId = new AtomicLong(Clock.Global.currentTimeMillis());
private static final AtomicReferenceFieldUpdater<AccordTask, String> loggingIdUpdater = AtomicReferenceFieldUpdater.newUpdater(AccordTask.class, String.class, "loggingId");
@ -220,10 +220,10 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
private BiConsumer<? super R, Throwable> callback;
public AccordTask(@Nonnull AccordCommandStore commandStore, PreLoadContext preLoadContext)
public AccordTask(@Nonnull AccordCommandStore commandStore, ExecutionContext executionContext)
{
this.commandStore = commandStore;
this.preLoadContext = preLoadContext;
this.executionContext = executionContext;
this.loggingId = "0x" + Long.toHexString(nextLoggingId.incrementAndGet());
if (logger.isTraceEnabled())
@ -245,7 +245,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
@Override
public String toString()
{
return preLoadContext.describe() + ' ' + toBriefString();
return executionContext.describe() + ' ' + toBriefString();
}
public String toBriefString()
@ -257,7 +257,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
{
return toBriefString() + ": "
+ (queued == null ? "unqueued" : queued.kind)
+ ", primaryTxnId: " + preLoadContext.primaryTxnId()
+ ", primaryTxnId: " + executionContext.primaryTxnId()
+ ", waitingToLoad: " + summarise(waitingToLoad)
+ ", loading:" + summarise(loading, AccordSafeState::global)
+ ", cfks:" + summarise(commandsForKey, AccordSafeState::global)
@ -315,7 +315,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
Unseekables<?> keys()
{
return preLoadContext.keys();
return executionContext.keys();
}
// TODO (expected): try to execute immediately BUT consider ordering requirements
@ -363,22 +363,22 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
// so we do not mutate anything, except the atomic counter of references
if (parent.commands != null)
{
for (TxnId txnId : preLoadContext.txnIds())
for (TxnId txnId : executionContext.txnIds())
presetupExclusive(txnId, AccordTask::ensureCommands, parent.commands, commandStore.cachesUnsafe().commands());
}
if (parent.commandsForKey == null) return;
if (preLoadContext.keys().domain() != Key) return;
switch (preLoadContext.loadKeys())
if (executionContext.keys().domain() != Key) return;
switch (executionContext.loadKeys())
{
default: throw new UnhandledEnum(preLoadContext.loadKeys());
default: throw new UnhandledEnum(executionContext.loadKeys());
case NONE:
break;
case ASYNC:
case INCR:
case SYNC:
for (RoutingKey key : (AbstractUnseekableKeys)preLoadContext.keys())
for (RoutingKey key : (AbstractUnseekableKeys) executionContext.keys())
presetupExclusive(key, AccordTask::ensureCommandsForKey, parent.commandsForKey, commandStore.cachesUnsafe().commandsForKeys());
break;
}
@ -402,7 +402,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
{
{
boolean hasPreSetup = commands != null;
for (TxnId txnId : preLoadContext.txnIds())
for (TxnId txnId : executionContext.txnIds())
{
if (hasPreSetup && completePresetupExclusive(txnId, commands, caches.commands()))
continue;
@ -410,22 +410,22 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
}
}
if (preLoadContext.keys().isEmpty())
if (executionContext.keys().isEmpty())
return;
switch (preLoadContext.keys().domain())
switch (executionContext.keys().domain())
{
case Key: setupKeyLoadsExclusive(caches, (AbstractUnseekableKeys)preLoadContext.keys(), false); break;
case Key: setupKeyLoadsExclusive(caches, (AbstractUnseekableKeys) executionContext.keys(), false); break;
case Range: setupRangeLoadsExclusive(caches);
}
}
private void setupKeyLoadsExclusive(Caches caches, Iterable<? extends RoutingKey> keys, boolean isToCompleteRangeScan)
{
if (preLoadContext.loadKeys() == LoadKeys.NONE)
if (executionContext.loadKeys() == LoadKeys.NONE)
return;
if (!isToCompleteRangeScan && preLoadContext.loadKeysFor() == RECOVERY)
if (!isToCompleteRangeScan && executionContext.loadKeysFor() == RECOVERY)
{
Invariants.require(rangeScanner == null);
rangeScanner = new RangeTxnScanner();
@ -441,12 +441,12 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
private void setupRangeLoadsExclusive(Caches caches)
{
if (preLoadContext.loadKeysFor() == WRITE)
if (executionContext.loadKeysFor() == WRITE)
return;
switch (preLoadContext.loadKeys())
switch (executionContext.loadKeys())
{
default: throw new UnhandledEnum(preLoadContext.loadKeys());
default: throw new UnhandledEnum(executionContext.loadKeys());
case NONE:
case ASYNC:
break;
@ -564,9 +564,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
return true;
}
public PreLoadContext preLoadContext()
public ExecutionContext preLoadContext()
{
return preLoadContext;
return executionContext;
}
public Map<TxnId, AccordSafeCommand> commands()
@ -680,7 +680,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
try (Closeable close = resources.get())
{
if (Tracing.isTracing())
Tracing.trace(preLoadContext.describe());
Tracing.trace(executionContext.describe());
state(RUNNING);
@ -789,7 +789,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
void cancelExclusive()
{
logger.info("Cancelling {}", preLoadContext);
logger.info("Cancelling {}", executionContext);
state(CANCELLED);
if (rangeScanner != null)
rangeScanner.cancelled = true;
@ -968,7 +968,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
// TODO (expected): produce key summaries to avoid locking all in memory
final Set<TokenKey> intersectingKeys = new ObjectHashSet<>();
final KeyWatcher keyWatcher = new KeyWatcher();
final Ranges ranges = ((AbstractRanges) preLoadContext.keys()).toRanges();
final Ranges ranges = ((AbstractRanges) executionContext.keys()).toRanges();
final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache;
public RangeTxnAndKeyScanner(AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache)
@ -1082,9 +1082,9 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
loader.load(guardedSummaries, () -> cancelled);
}
PreLoadContext preLoadContext()
ExecutionContext preLoadContext()
{
return preLoadContext;
return executionContext;
}
@Override
@ -1109,7 +1109,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
void startInternal(Caches caches)
{
loader = commandStore.rangeIndex().loader(preLoadContext.primaryTxnId(), preLoadContext.executeAt(), preLoadContext.loadKeysFor(), preLoadContext.keys());
loader = commandStore.rangeIndex().loader(executionContext.primaryTxnId(), executionContext.executeAt(), executionContext.loadKeysFor(), executionContext.keys());
loader.loadExclusive(guardedSummaries, caches);
}
@ -1143,7 +1143,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
@Override
public String description()
{
return "Scanning range intersections for " + preLoadContext.reason() + ' ' + toBriefString();
return "Scanning range intersections for " + executionContext.reason() + ' ' + toBriefString();
}
@Override
@ -1174,6 +1174,6 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
@Override
public String description()
{
return preLoadContext.describe();
return executionContext.describe();
}
}

8
src/java/org/apache/cassandra/service/accord/debug/DebugBlockedTxns.java
View File

@ -37,7 +37,7 @@ import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommandStore;
import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.RoutingKeys;
@ -178,7 +178,7 @@ public class DebugBlockedTxns
private AsyncChain<Txn> visitRootTxnAsync(CommandStore commandStore, TxnId txnId)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
@ -188,7 +188,7 @@ public class DebugBlockedTxns
private AsyncChain<Txn> visitTxnAsync(CommandStore commandStore, TxnId txnId, Timestamp rootExecuteAt, @Nullable TokenKey byKey, int depth, boolean recurse)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
@ -231,7 +231,7 @@ public class DebugBlockedTxns
private AsyncChain<Void> visitKeysAsync(CommandStore commandStore, TokenKey key, Timestamp rootExecuteAt, int depth)
{
return commandStore.chain(PreLoadContext.contextFor(RoutingKeys.of(key.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(RoutingKeys.of(key.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), safeStore -> {
visitKeysSync(safeStore, key, rootExecuteAt, depth);
});
}

8
src/java/org/apache/cassandra/service/accord/debug/DebugTxnGraph.java
View File

@ -40,7 +40,7 @@ import javax.annotation.Nullable;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommandStore;
import accord.primitives.PartialDeps;
import accord.primitives.Participants;
@ -221,7 +221,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<TxnInfos<T>> submitRoot(CommandStore commandStore, TxnId txnId)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return AsyncChains.<TxnInfos<T>>success(null);
@ -232,7 +232,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<TxnInfos<T>> submitParent(CommandStore commandStore, TxnId txnId, P param, Map<TxnId, SaveInfo> infos, Set<TxnId> visitedParent, int depth)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return AsyncChains.<TxnInfos<T>>success(null);
@ -344,7 +344,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<Void> populateTxnAsync(CommandStore commandStore, TxnId txnId, Map<TxnId, SaveInfo> visited)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
visited.putIfAbsent(txnId, command == null || command.saveStatus() == SaveStatus.Uninitialised ? SaveInfo.NONE : new SaveInfo(command.saveStatus(), command.executeAtIfKnown()));
});

10
src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
View File

@ -30,7 +30,7 @@ import accord.coordinate.CoordinationAdapter.Adapters.TxnAdapter;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
import accord.coordinate.ExecutePath;
import accord.local.Node;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.messages.Apply;
import accord.primitives.Ballot;
import accord.primitives.Deps;
@ -83,14 +83,14 @@ public class AccordInteropAdapter extends TxnAdapter
}
@Override
public void execute(Node node, SequentialAsyncExecutor executor, Topologies any, FullRoute<?> route, Ballot ballot, ExecutePath path, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
public void execute(Node node, ExclusiveAsyncExecutor executor, Topologies any, FullRoute<?> route, Ballot ballot, ExecutePath path, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
{
if (!doInteropExecute(node, executor, route, ballot, txnId, txn, executeAt, stableDeps, callback))
super.execute(node, executor, any, route, ballot, path, flags, txnId, txn, executeAt, stableDeps, sendDeps, callback);
}
@Override
public void persist(Node node, SequentialAsyncExecutor executor, Topologies any, Route<?> require, Route<?> sendTo, FullRoute<?> route, Ballot ballot, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
public void persist(Node node, ExclusiveAsyncExecutor executor, Topologies any, Route<?> require, Route<?> sendTo, FullRoute<?> route, Ballot ballot, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
{
if (applyKind == Minimal && doInteropPersist(node, executor, any, require, sendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback))
return;
@ -98,7 +98,7 @@ public class AccordInteropAdapter extends TxnAdapter
super.persist(node, executor, any, require, sendTo, route, ballot, flags, txnId, txn, executeAt, deps, writes, result, informDurableOnDone, callback);
}
private boolean doInteropExecute(Node node, SequentialAsyncExecutor executor, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
private boolean doInteropExecute(Node node, ExclusiveAsyncExecutor executor, FullRoute<?> route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
{
// Unrecoverable repair always needs to be run by AccordInteropExecution
AccordUpdate.Kind updateKind = AccordUpdate.kind(txn.update());
@ -121,7 +121,7 @@ public class AccordInteropAdapter extends TxnAdapter
return true;
}
private boolean doInteropPersist(Node node, SequentialAsyncExecutor executor, Topologies any, Route<?> require, Route<?> sendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
private boolean doInteropPersist(Node node, ExclusiveAsyncExecutor executor, Topologies any, Route<?> require, Route<?> sendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
{
Update update = txn.update();
ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null;

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

@ -33,7 +33,7 @@ import accord.coordinate.CoordinationAdapter;
import accord.coordinate.ExecuteFlag.CoordinationFlags;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.messages.Commit;
import accord.messages.Commit.Kind;
import accord.primitives.AbstractRanges;
@ -125,7 +125,7 @@ public class AccordInteropExecution implements ReadCoordinator
private final Timestamp executeAt;
private final Deps deps;
private final BiConsumer<? super Result, Throwable> callback;
private final SequentialAsyncExecutor executor;
private final ExclusiveAsyncExecutor executor;
private final ConsistencyLevel consistencyLevel;
private final AccordEndpointMapper endpointMapper;
@ -141,7 +141,7 @@ public class AccordInteropExecution implements ReadCoordinator
private volatile long uniqueHlc;
public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute<?> route, Ballot ballot, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback,
SequentialAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) throws TopologyException
ExclusiveAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) throws TopologyException
{
requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR);
this.node = node;

4
src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
View File

@ -26,7 +26,7 @@ import accord.coordinate.Persist;
import accord.coordinate.tracking.AllTracker;
import accord.coordinate.tracking.QuorumTracker;
import accord.local.Node;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.messages.Apply;
import accord.primitives.Ballot;
import accord.primitives.Deps;
@ -53,7 +53,7 @@ import static org.apache.cassandra.db.ConsistencyLevel.SERIAL;
*/
public class AccordInteropPersist extends Persist
{
public AccordInteropPersist(Node node, SequentialAsyncExecutor executor, Topologies topologies, TxnId txnId, Route<?> sendTo, Ballot ballot, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, ExecuteFlag.CoordinationFlags flags, boolean informDurableOnDone, Apply.Kind applyKind, BiConsumer<? super Result, Throwable> callback)
public AccordInteropPersist(Node node, ExclusiveAsyncExecutor executor, Topologies topologies, TxnId txnId, Route<?> sendTo, Ballot ballot, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ConsistencyLevel consistencyLevel, ExecuteFlag.CoordinationFlags flags, boolean informDurableOnDone, Apply.Kind applyKind, BiConsumer<? super Result, Throwable> callback)
{
super(node, executor, topologies, txnId, ballot, sendTo, txn, executeAt, deps, writes, result.toPersistable(), fullRoute, flags, informDurableOnDone, AccordInteropApply.FACTORY, applyKind, consistencyLevel == ALL ? AllTracker::new : QuorumTracker::new, (ignore, fail) -> {
if (fail != null) callback.accept(null, fail);

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
View File

@ -24,9 +24,9 @@ import org.assertj.core.api.Assertions;
import accord.api.RoutingKey;
import accord.local.CommandStores;
import accord.local.ExecutionContext;
import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.cfk.CommandsForKey;
import accord.primitives.Ranges;
import accord.primitives.Routable;
@ -135,7 +135,7 @@ public class AccordDropTableBase extends TestBaseImpl
TableId tableId = TableId.fromString(s);
AccordService accord = (AccordService) AccordService.instance();
TxnId syntheticTxnId = new TxnId(TxnId.MAX_EPOCH, 0, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, new Node.Id(1));
PreLoadContext ctx = PreLoadContext.contextFor(syntheticTxnId, Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), LoadKeys.SYNC, READ_WRITE, "Test");
ExecutionContext ctx = ExecutionContext.contextFor(syntheticTxnId, Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), LoadKeys.SYNC, READ_WRITE, "Test");
CommandStores stores = accord.node().commandStores();
for (int storeId : stores.ids())
{

6
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

@ -56,7 +56,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.CommandStore;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommand;
import accord.primitives.PartialDeps;
import accord.primitives.TxnId;
@ -832,7 +832,7 @@ public class AccordLoadTest extends AccordTestBase
if (storeId.get() >= 0)
{
CommandStore commandStore = service.node().commandStores().forId(storeId.get());
List<List<String>> result = AccordService.getBlocking(commandStore.submit(PreLoadContext.contextFor(candidate, "LoadTest"), safeStore -> {
List<List<String>> result = AccordService.getBlocking(commandStore.submit(ExecutionContext.contextFor(candidate, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(candidate);
PartialDeps deps = safeCommand.current().partialDeps();
if (deps == null)
@ -855,7 +855,7 @@ public class AccordLoadTest extends AccordTestBase
for (List<String> info : result)
{
TxnId txnId = TxnId.parse(info.get(0));
AccordService.getBlocking(commandStore.execute(PreLoadContext.contextFor(txnId, "LoadTest"), safeStore -> {
AccordService.getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
if (safeCommand.current().executeAt != null)
info.add(safeCommand.current().executeAt.toString());

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordRegainRangesTest.java
View File

@ -29,7 +29,7 @@ import org.slf4j.LoggerFactory;
import accord.local.CommandStore;
import accord.local.CommandStores.PreviouslyOwned;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.primitives.AbstractRanges;
import accord.primitives.Ranges;
import accord.topology.ActiveEpochs;
@ -121,7 +121,7 @@ public class AccordRegainRangesTest extends AccordTestBase
Ranges range = Ranges.EMPTY;
for (CommandStore commandStore : AccordService.instance().node().commandStores().all()) {
Ranges safeToReadRanges = getBlocking(commandStore.submit((PreLoadContext.Empty) () -> "No overlapping safeToReadRanges", safeCommandStore -> {
Ranges safeToReadRanges = getBlocking(commandStore.submit((ExecutionContext.Empty) () -> "No overlapping safeToReadRanges", safeCommandStore -> {
Ranges mergedRanges = Ranges.EMPTY;
for (Ranges r : safeCommandStore.safeToReadAt().values())
mergedRanges = mergedRanges.union(AbstractRanges.UnionMode.MERGE_ADJACENT, r);

9
test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java
View File

@ -35,9 +35,8 @@ import java.util.function.BooleanSupplier;
import org.junit.Test;
import accord.api.AsyncExecutor;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import org.apache.cassandra.concurrent.ExecutorFactory;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableSupplier;
@ -85,7 +84,7 @@ public class AccordExecutorTest extends SimulationTestBase
ExecutorPlus submit = executorFactory().pooled("submit-test", submissionThreads);
AccordExecutor executor = supplier.get();
Lock lock = executor.unsafeLock();
SequentialAsyncExecutor sequentialExecutor = executor.newSequentialExecutor();
ExclusiveAsyncExecutor sequentialExecutor = executor.newSequentialExecutor();
Executor lockExecutor = executorFactory().sequential("lock");
for (float sleepChance : new float[] { 0f, 0.01f, 0.1f })
@ -120,7 +119,7 @@ public class AccordExecutorTest extends SimulationTestBase
() -> {}, 1L);
}
private static void submitLoop(int id, Lock lock, AccordExecutor executor, SequentialAsyncExecutor 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) throws ExecutionException, InterruptedException
{
ConcurrentLinkedQueue<Future<?>> await = new ConcurrentLinkedQueue<>();
while (outerLoop-- > 0)
@ -137,7 +136,7 @@ public class AccordExecutorTest extends SimulationTestBase
}
}
private static void submitRecursive(Lock lock, AccordExecutor executor, SequentialAsyncExecutor sequentialExecutor, int count, Collection<Future<?>> await, float sleepChance, float lockChance)
private static void submitRecursive(Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, int count, Collection<Future<?>> await, float sleepChance, float lockChance)
{
AsyncExecutor submitTo = ThreadLocalRandom.current().nextBoolean() ? executor : sequentialExecutor;
await.add(toFuture(submitTo.chain(() -> {

2
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -92,7 +92,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.PreLoadContext.contextFor;
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;

6
test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
View File

@ -32,7 +32,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.Result.PersistableResult;
import accord.local.Command;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.StoreParticipants;
import accord.local.cfk.CommandsForKey;
import accord.primitives.Ballot;
@ -170,8 +170,8 @@ public class AccordCommandStoreTest
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();
cfk.set(cfk.current().update(new TestSafeCommandStore(PreLoadContext.contextFor(command1.txnId(), "Test")), command1).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(PreLoadContext.contextFor(command1.txnId(), "Test")), command2).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.contextFor(command1.txnId(), "Test")), command1).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.contextFor(command1.txnId(), "Test")), command2).cfk());
CommandsForKeyAccessor.systemTableUpdater(commandStore.id(), (TokenKey)cfk.key(), cfk.current(), null, commandStore.nextSystemTimestampMicros()).run();
logger.info("E: {}", cfk);

8
test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
View File

@ -29,7 +29,7 @@ import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
@ -98,7 +98,7 @@ public class AccordCommandTest
public void basicCycleTest() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getBlocking(commandStore.execute((PreLoadContext.Empty)() -> "Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
getBlocking(commandStore.execute((ExecutionContext.Empty)() -> "Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(1);
@ -174,7 +174,7 @@ public class AccordCommandTest
Commit commit = Commit.SerializerSupport.create(txnId, route, 1, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute);
getBlocking(commandStore.execute(commit, commit::apply));
getBlocking(commandStore.execute(PreLoadContext.contextFor(txnId, Keys.of(key).toParticipants(), LoadKeys.SYNC, READ_WRITE, "Test"), safeStore -> {
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, Keys.of(key).toParticipants(), LoadKeys.SYNC, READ_WRITE, "Test"), safeStore -> {
Command before = safeStore.ifInitialised(txnId).current();
Assert.assertEquals(commit.executeAt, before.executeAt());
Assert.assertTrue(before.hasBeen(Status.Committed));
@ -191,7 +191,7 @@ public class AccordCommandTest
public void computeDeps() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getBlocking(commandStore.execute((PreLoadContext.Empty)()->"Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
getBlocking(commandStore.execute((ExecutionContext.Empty)()->"Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(2);

14
test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
View File

@ -46,7 +46,7 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
@ -87,7 +87,7 @@ import org.apache.cassandra.utils.concurrent.Condition;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.PreLoadContext.contextFor;
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;
@ -127,7 +127,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getBlocking(commandStore.execute(PreLoadContext.contextFor(txnId, "Test"), instance -> {
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "Test"), instance -> {
// TODO review: This change to `ifInitialized` was done in a lot of places and it doesn't preserve this property
// I fixed this reference to point to `ifLoadedAndInitialised` and but didn't update other places
Assert.assertNull(instance.ifInitialised(txnId));
@ -141,7 +141,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getBlocking(commandStore.execute(PreLoadContext.contextFor(txnId, "Test"), safe -> {
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "Test"), safe -> {
StoreParticipants participants = StoreParticipants.empty(txnId);
SafeCommand command = safe.get(txnId, participants);
Assert.assertNotNull(command);
@ -155,7 +155,7 @@ public class AccordTaskTest
Txn txn = AccordTestUtils.createWriteTxn((int)clock.incrementAndGet());
TokenKey key = ((PartitionKey) Iterables.getOnlyElement(txn.keys())).toUnseekable();
getBlocking(commandStore.execute((PreLoadContext.Empty)() -> "Test", instance -> {
getBlocking(commandStore.execute((ExecutionContext.Empty)() -> "Test", instance -> {
SafeCommandsForKey cfk = instance.ifLoadedAndInitialised(key);
Assert.assertNull(cfk);
}));
@ -298,7 +298,7 @@ public class AccordTaskTest
awaitDone(commandStore, ids, participants);
assertNoReferences(commandStore, ids, participants);
PreLoadContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
ExecutionContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
Consumer<SafeCommandStore> consumer = Mockito.mock(Consumer.class);
Map<TxnId, Boolean> failed = selectFailedTxn(rs, ids);
@ -363,7 +363,7 @@ public class AccordTaskTest
assertNoReferences(commandStore, ids, participants);
createCommand(commandStore, rs, ids);
PreLoadContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
ExecutionContext ctx = contextFor(ids.get(0), ids.size() == 1 ? null : ids.get(1), participants, SYNC, READ_WRITE, "Test");
Consumer<SafeCommandStore> consumer = Mockito.mock(Consumer.class);
String errorMsg = "txn_ids " + ids;

10
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -51,12 +51,12 @@ import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.ExecutionContext;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.local.StoreParticipants;
import accord.local.TimeService;
import accord.local.durability.DurabilityService;
@ -244,7 +244,7 @@ public class AccordTestUtils
public static AsyncChain<Pair<Writes, PersistableResult>> processTxnResult(AccordCommandStore commandStore, TxnId txnId, PartialTxn txn, Timestamp executeAt) throws Throwable
{
AtomicReference<AsyncChain<Pair<Writes, PersistableResult>>> result = new AtomicReference<>();
getBlocking(commandStore.execute((PreLoadContext.Empty)() -> "Test",
getBlocking(commandStore.execute((ExecutionContext.Empty)() -> "Test",
safeStore -> result.set(processTxnResultDirect(safeStore, txnId, txn, executeAt))));
return result.get();
}
@ -371,7 +371,7 @@ public class AccordTestUtils
}
@Override
public SequentialAsyncExecutor someSequentialExecutor()
public ExclusiveAsyncExecutor someSequentialExecutor()
{
return null;
}
@ -419,7 +419,7 @@ public class AccordTestUtils
Node.Id node = new Id(1);
Topology topology = new Topology(1, Shard.create(range, new SortedArrayList<>(new Id[] { node }), Sets.newHashSet(node)));
AccordCommandStore store = createAccordCommandStore(node, now, topology);
store.execute((PreLoadContext.Empty)()->"Test", safeStore -> ((AccordCommandStore)safeStore.commandStore()).executor().cacheUnsafe().setCapacity(1 << 20));
store.execute((ExecutionContext.Empty)()->"Test", safeStore -> ((AccordCommandStore)safeStore.commandStore()).executor().cacheUnsafe().setCapacity(1 << 20));
return store;
}

4
test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java
View File

@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord;
import org.assertj.core.api.Assertions;
import org.junit.Test;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.StoreParticipants;
import accord.primitives.SaveStatus;
import accord.primitives.TxnId;
@ -41,7 +41,7 @@ public class SimpleSimulatedAccordCommandStoreTest extends SimulatedAccordComman
for (int i = 0, examples = 100; i < examples; i++)
{
TxnId id = AccordGens.txnIds().next(rs);
instance.process(PreLoadContext.contextFor(id, "Test"), (safe) -> {
instance.process(ExecutionContext.contextFor(id, "Test"), (safe) -> {
var safeCommand = safe.get(id, StoreParticipants.empty(id));
var command = safeCommand.current();
Assertions.assertThat(command.saveStatus()).isEqualTo(SaveStatus.Uninitialised);

10
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -54,12 +54,12 @@ import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.ExecutionContext;
import accord.local.Node;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.SequentialAsyncExecutor;
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 SequentialAsyncExecutor someSequentialExecutor()
public ExclusiveAsyncExecutor someSequentialExecutor()
{
return null;
}
@ -448,7 +448,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
return process(request, request);
}
public <T extends Reply> T process(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function) throws ExecutionException, InterruptedException
public <T extends Reply> T process(ExecutionContext loadCtx, Function<? super SafeCommandStore, T> function) throws ExecutionException, InterruptedException
{
var result = processAsync(loadCtx, function);
processAll();
@ -460,7 +460,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
return processAsync(request, request);
}
public <T extends Reply> AsyncResult<T> processAsync(PreLoadContext loadCtx, Function<? super SafeCommandStore, T> function)
public <T extends Reply> AsyncResult<T> processAsync(ExecutionContext loadCtx, Function<? super SafeCommandStore, T> function)
{
return commandStore.submit(loadCtx, function);
}

10
test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java
View File

@ -32,7 +32,7 @@ import org.junit.Test;
import accord.api.RoutingKey;
import accord.impl.basic.SimulatedFault;
import accord.local.PreLoadContext;
import accord.local.ExecutionContext;
import accord.local.SafeCommandStore;
import accord.messages.PreAccept;
import accord.primitives.FullRoute;
@ -107,7 +107,7 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
{
case Task:
{
PreLoadContext ctx = (PreLoadContext.Empty)()->"Test";
ExecutionContext ctx = (ExecutionContext.Empty)()->"Test";
instance.maybeCacheEvict(ctx.keys());
operation(instance, ctx, actionGen.next(rs), rs::nextBoolean).chain().begin(counter);
}
@ -178,7 +178,7 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
private enum Action { SUCCESS, FAILURE, LOAD_FAILURE }
private static AccordTask<Void> operation(SimulatedAccordCommandStore instance, PreLoadContext ctx, Action action, BooleanSupplier delay)
private static AccordTask<Void> operation(SimulatedAccordCommandStore instance, ExecutionContext ctx, Action action, BooleanSupplier delay)
{
return new SimulatedOperation(instance.commandStore, ctx, action == Action.FAILURE ? SimulatedOperation.Action.FAILURE : SimulatedOperation.Action.SUCCESS);
}
@ -201,9 +201,9 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
enum Action { SUCCESS, FAILURE}
private final Action action;
public SimulatedOperation(AccordCommandStore commandStore, PreLoadContext preLoadContext, Action action)
public SimulatedOperation(AccordCommandStore commandStore, ExecutionContext executionContext, Action action)
{
super(commandStore, preLoadContext);
super(commandStore, executionContext);
this.action = action;
}

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

@ -66,13 +66,13 @@ import accord.local.CommandBuilder;
import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.ExecutionContext;
import accord.local.Node;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.SequentialAsyncExecutor;
import accord.local.ExclusiveAsyncExecutor;
import accord.local.StoreParticipants;
import accord.local.TimeService;
import accord.local.cfk.CommandsForKey;
@ -505,8 +505,8 @@ public class CommandsForKeySerializerTest
{
int next = source.nextInt(commands.size());
Command command = commands.get(next);
if (command.txnId.isSyncPoint()) cfk = cfk.registerUnmanaged(new TestSafeCommandStore(PreLoadContext.contextFor(command.txnId(), "Test")), new TestSafeCommand(command), REGISTER).cfk();
else cfk = cfk.update(new TestSafeCommandStore(PreLoadContext.contextFor(command.txnId(), "Test")), command).cfk();
if (command.txnId.isSyncPoint()) cfk = cfk.registerUnmanaged(new TestSafeCommandStore(ExecutionContext.contextFor(command.txnId(), "Test")), new TestSafeCommand(command), REGISTER).cfk();
else cfk = cfk.update(new TestSafeCommandStore(ExecutionContext.contextFor(command.txnId(), "Test")), command).cfk();
commands.set(next, commands.get(commands.size() - 1));
commands.remove(commands.size() - 1);
}
@ -667,8 +667,8 @@ public class CommandsForKeySerializerTest
}
@Override public boolean inStore() { return true; }
@Override public AsyncChain<Void> chain(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { throw new UnsupportedOperationException();}
@Override public <T> AsyncChain<T> chain(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public AsyncChain<Void> chain(ExecutionContext context, Consumer<? super SafeCommandStore> consumer) { throw new UnsupportedOperationException();}
@Override public <T> AsyncChain<T> chain(ExecutionContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public Journal.Replayer replayer(AbstractReplayer.Mode mode) { throw new UnsupportedOperationException(); }
@ -702,7 +702,7 @@ public class CommandsForKeySerializerTest
public static class TestSafeCommandStore extends AbstractSafeCommandStore
{
public TestSafeCommandStore(PreLoadContext context)
public TestSafeCommandStore(ExecutionContext context)
{
super(context, TestCommandStore.INSTANCE);
}
@ -718,7 +718,7 @@ public class CommandsForKeySerializerTest
@Override public NodeCommandStoreService node() { return new NodeCommandStoreService()
{
@Override public AsyncExecutor someExecutor() { throw new UnsupportedOperationException(); }
@Override public SequentialAsyncExecutor someSequentialExecutor() { throw new UnsupportedOperationException(); }
@Override public ExclusiveAsyncExecutor someSequentialExecutor() { throw new UnsupportedOperationException(); }
@Override public long epoch() { return 0;}
@Override public Node.Id id() { return Node.Id.NONE; }
@Override public Timeouts timeouts() { return null; }