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

Improve: 

- PreLoadContext descriptions
 - Introduce LoadKeysFor so we can avoid loading range transactions except where necessary

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20758
This commit is contained in:
Benedict Elliott Smith 2025-06-30 10:35:43 +01:00
parent 9df548109c
commit 0e9f0fab78
17 changed files with 79 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 1c203efdd463f2c452e8072267370f9e8dfdb8f3
Subproject commit a1c1ed91cfeb904be9b490d498c1c52aba2253c6

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

@ -507,7 +507,7 @@ public class AccordCommandStore extends CommandStore
@Override
public AsyncChain<Route> load(TxnId txnId)
{
return store.submit(txnId, safeStore -> {
return store.submit(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
initialiseState(safeStore, txnId);
return safeStore.unsafeGet(txnId).current().route();
});

7
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -56,7 +56,6 @@ import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.KeyHistory;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
@ -134,6 +133,8 @@ import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.api.ProtocolModifiers.Toggles.FastExec.MAY_BYPASS_SAFESTORE;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
@ -791,7 +792,7 @@ public class AccordService implements IAccordService, Shutdownable
private static void populateAsync(CommandStoreTxnBlockedGraph.Builder state, CommandStore store, TxnId txnId)
{
state.asyncTxns.incrementAndGet();
store.execute(txnId, in -> {
store.execute(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), in -> {
populateSync(state, (AccordSafeCommandStore) in, txnId);
if (0 == state.asyncTxns.decrementAndGet() && 0 == state.asyncKeys.get())
state.complete();
@ -841,7 +842,7 @@ public class AccordService implements IAccordService, Shutdownable
private static void populateAsync(CommandStoreTxnBlockedGraph.Builder state, CommandStore commandStore, TokenKey blockedBy, TxnId txnId, Timestamp executeAt)
{
state.asyncKeys.incrementAndGet();
commandStore.execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(blockedBy.toUnseekable()), KeyHistory.SYNC), in -> {
commandStore.execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(blockedBy.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), in -> {
populateSync(state, (AccordSafeCommandStore) in, blockedBy, txnId, executeAt);
if (0 == state.asyncKeys.decrementAndGet() && 0 == state.asyncTxns.get())
state.complete();

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

@ -42,6 +42,7 @@ import accord.api.Journal;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.LoadKeys;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
@ -52,6 +53,7 @@ import accord.primitives.Ranges;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.Cancellable;
@ -69,6 +71,9 @@ import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.Condition;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.LoadKeysFor.RECOVERY;
import static accord.local.LoadKeysFor.WRITE;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.utils.Invariants.illegalState;
@ -342,14 +347,13 @@ public abstract class AccordTask<R> extends SubmittableTask implements Runnable,
if (parent.commandsForKey == null) return;
if (preLoadContext.keys().domain() != Key) return;
switch (preLoadContext.keyHistory())
switch (preLoadContext.loadKeys())
{
default: throw new AssertionError("Unhandled KeyHistory: " + preLoadContext.keyHistory());
default: throw new UnhandledEnum(preLoadContext.loadKeys());
case NONE:
break;
case ASYNC:
case RECOVER:
case INCR:
case SYNC:
for (RoutingKey key : (AbstractUnseekableKeys)preLoadContext.keys())
@ -396,39 +400,31 @@ public abstract class AccordTask<R> extends SubmittableTask implements Runnable,
private void setupKeyLoadsExclusive(Caches caches, Iterable<? extends RoutingKey> keys, boolean isToCompleteRangeScan)
{
switch (preLoadContext.keyHistory())
if (preLoadContext.loadKeys() == LoadKeys.NONE)
return;
if (!isToCompleteRangeScan && preLoadContext.loadKeysFor() == RECOVERY)
{
default: throw new AssertionError("Unhandled KeyHistory: " + preLoadContext.keyHistory());
case NONE:
break;
Invariants.require(rangeScanner == null);
rangeScanner = new RangeTxnScanner();
}
case RECOVER:
if (!isToCompleteRangeScan)
{
Invariants.require(rangeScanner == null);
rangeScanner = new RangeTxnScanner();
}
case ASYNC:
case INCR:
case SYNC:
{
boolean hasPreSetup = commandsForKey != null;
for (RoutingKey key : keys)
{
if (hasPreSetup && completePresetupExclusive(key, commandsForKey, caches.commandsForKeys())) continue;
setupExclusive(key, AccordTask::ensureCommandsForKey, caches.commandsForKeys());
}
break;
}
boolean hasPreSetup = commandsForKey != null;
for (RoutingKey key : keys)
{
if (hasPreSetup && completePresetupExclusive(key, commandsForKey, caches.commandsForKeys())) continue;
setupExclusive(key, AccordTask::ensureCommandsForKey, caches.commandsForKeys());
}
}
private void setupRangeLoadsExclusive(Caches caches)
{
switch (preLoadContext.keyHistory())
if (preLoadContext.loadKeysFor() == WRITE)
return;
switch (preLoadContext.loadKeys())
{
default: throw new AssertionError("Unhandled KeyHistory: " + preLoadContext.keyHistory());
default: throw new UnhandledEnum(preLoadContext.loadKeys());
case NONE:
case ASYNC:
break;
@ -436,7 +432,6 @@ public abstract class AccordTask<R> extends SubmittableTask implements Runnable,
case INCR:
throw new AssertionError("Incremental mode should only be used with an explicit list of keys");
case RECOVER:
case SYNC:
hasRanges = true;
rangeScanner = new RangeTxnAndKeyScanner(caches.commandsForKeys());
@ -1091,7 +1086,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Runnable,
void startInternal(Caches caches)
{
summaryLoader = commandStore.commandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.keyHistory(), keysOrRanges);
summaryLoader = commandStore.commandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.loadKeysFor(), keysOrRanges);
summaryLoader.forEachInCache(keysOrRanges, summary -> summaries.put(summary.txnId, summary), caches);
caches.commands().register(commandWatcher);
}
@ -1149,6 +1144,6 @@ public abstract class AccordTask<R> extends SubmittableTask implements Runnable,
@Override
public String description()
{
return toString();
return preLoadContext.describe();
}
}

7
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
View File

@ -39,7 +39,8 @@ import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandSummaries;
import accord.local.CommandSummaries.Summary;
import accord.local.KeyHistory;
import accord.local.LoadKeys;
import accord.local.LoadKeysFor;
import accord.local.MaxDecidedRX;
import accord.local.RedundantBefore;
import accord.primitives.AbstractRanges;
@ -320,10 +321,10 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
}
}
public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
{
RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
return Loader.loader(redundantBefore, primaryTxnId, keyHistory, keysOrRanges, this::newLoader);
return Loader.loader(redundantBefore, primaryTxnId, loadKeysFor, keysOrRanges, this::newLoader);
}
private Loader newLoader(@Nullable TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)

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

@ -278,7 +278,7 @@ public class AccordBootstrapTest extends TestBaseImpl
Assert.assertTrue(session.getNumKeyspaceTransfers() > 0);
});
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach(safeStore -> {
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach((PreLoadContext.Empty)()->"Test", safeStore -> {
AccordSafeCommandStore ss = (AccordSafeCommandStore) safeStore;
Assert.assertEquals(Timestamp.NONE, getOnlyElement(ss.bootstrapBeganAt().keySet()));
Assert.assertEquals(Timestamp.NONE, getOnlyElement(ss.safeToReadAt().keySet()));
@ -321,7 +321,7 @@ public class AccordBootstrapTest extends TestBaseImpl
Assert.assertEquals(key, row.getInt("c"));
Assert.assertEquals(key, row.getInt("v"));
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach(safeStore -> {
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach((PreLoadContext.Empty)()->"Test", safeStore -> {
if (safeStore.ranges().currentRanges().contains(partitionKey))
{
AccordSafeCommandStore ss = (AccordSafeCommandStore) safeStore;
@ -464,7 +464,7 @@ public class AccordBootstrapTest extends TestBaseImpl
PartitionKey partitionKey = new PartitionKey(tableId, dk);
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach(PreLoadContext.contextFor(partitionKey.toUnseekable()),
awaitUninterruptiblyAndRethrow(service().node().commandStores().forEach((PreLoadContext.Empty)()->"Test",
partitionKey.toUnseekable(), moveMax, moveMax,
safeStore -> {
if (!safeStore.ranges().allAt(preMove).contains(partitionKey))

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

@ -22,7 +22,7 @@ import com.google.common.base.Throwables;
import accord.api.RoutingKey;
import accord.local.CommandStores;
import accord.local.KeyHistory;
import accord.local.LoadKeys;
import accord.local.PreLoadContext;
import accord.local.cfk.CommandsForKey;
import accord.primitives.Ranges;
@ -43,6 +43,7 @@ import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.TokenRange;
import org.assertj.core.api.Assertions;
import static accord.local.LoadKeysFor.READ_WRITE;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
public class AccordDropTableBase extends TestBaseImpl
@ -128,7 +129,7 @@ public class AccordDropTableBase extends TestBaseImpl
inst.runOnInstance(() -> {
TableId tableId = TableId.fromString(s);
AccordService accord = (AccordService) AccordService.instance();
PreLoadContext ctx = PreLoadContext.contextFor(Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), KeyHistory.SYNC);
PreLoadContext ctx = PreLoadContext.contextFor(Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), LoadKeys.SYNC, READ_WRITE, "Test");
CommandStores stores = accord.node().commandStores();
for (int storeId : stores.ids())
{

7
test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
View File

@ -72,7 +72,8 @@ import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.local.KeyHistory.SYNC;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static java.lang.String.format;
public class AccordIncrementalRepairTest extends AccordTestBase
@ -214,7 +215,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
{
Node node = accordService().node();
AtomicReference<TxnId> waitFor = new AtomicReference<>(null);
AsyncChains.awaitUninterruptibly(node.commandStores().ifLocal(PreLoadContext.contextFor(key, SYNC), key.toUnseekable(), 0, Long.MAX_VALUE, safeStore -> {
AsyncChains.awaitUninterruptibly(node.commandStores().ifLocal(PreLoadContext.contextFor(key, SYNC, READ_WRITE, "Test"), key.toUnseekable(), 0, Long.MAX_VALUE, safeStore -> {
AccordSafeCommandStore store = (AccordSafeCommandStore) safeStore;
SafeCommandsForKey safeCfk = store.ifLoadedAndInitialised(key);
if (safeCfk == null)
@ -236,7 +237,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
long now = Clock.Global.currentTimeMillis();
if (now - start > TimeUnit.MINUTES.toMillis(1))
throw new AssertionError("Timeout");
AsyncChains.awaitUninterruptibly(node.commandStores().ifLocal(txnId, key.toUnseekable(), 0, Long.MAX_VALUE, safeStore -> {
AsyncChains.awaitUninterruptibly(node.commandStores().ifLocal(PreLoadContext.contextFor(txnId, "Test"), key.toUnseekable(), 0, Long.MAX_VALUE, safeStore -> {
SafeCommand command = safeStore.get(txnId, StoreParticipants.empty(txnId));
Assert.assertNotNull(command.current());
if (command.current().status().hasBeen(Status.Applied))

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

@ -87,7 +87,8 @@ 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.KeyHistory.SYNC;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.PreLoadContext.contextFor;
import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_APPLIED;
import static accord.primitives.Routable.Domain.Range;
@ -313,28 +314,28 @@ public class CompactionAccordIteratorsTest
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.slice(commandStore.unsafeGetRangesForEpoch().currentRanges());
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, txnId, partialDeps, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC), safe -> {
getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
return AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
}).flatMap(i -> i).flatMap(result -> commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
}).flatMap(i -> i).flatMap(result -> commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, (a, b) -> {});
})));
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied
// in order to have the updated state in the system table
spinAssertEquals(true, 5, () -> {
return getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC), safe -> {
return getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
StoreParticipants participants = StoreParticipants.all(route);
Command command = safe.get(txnId, participants).current();
return command.hasBeen(Status.Applied);

3
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -32,6 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ProtocolModifiers;
import accord.local.PreLoadContext;
import accord.messages.TxnRequest;
import accord.primitives.Ranges;
import accord.primitives.Routable;
@ -193,7 +194,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
TxnId syncId2 = new TxnId(101, 300, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, accord.nodeId());
Ranges ranges1 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(1)), new TokenKey(tableId, new LongToken(100))));
Ranges ranges2 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(100)), new TokenKey(tableId, new LongToken(200))));
AsyncChains.getBlocking(accord.node().commandStores().forEach(safeStore -> {
AsyncChains.getBlocking(accord.node().commandStores().forEach((PreLoadContext.Empty)() -> "Test", safeStore -> {
safeStore.commandStore().markShardDurable(safeStore, syncId1, ranges1, Status.Durability.Universal);
safeStore.commandStore().markShardDurable(safeStore, syncId2, ranges2, Status.Durability.Majority);
}));

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

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

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

@ -29,7 +29,7 @@ import accord.api.RoutingKey;
import accord.local.StoreParticipants;
import accord.local.cfk.CommandsForKey;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@ -57,6 +57,7 @@ import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.utils.ByteBufferUtil;
import static accord.api.ProtocolModifiers.Toggles.filterDuplicateDependenciesFromAcceptReply;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.messages.Accept.Kind.SLOW;
import static accord.utils.async.AsyncChains.getUninterruptibly;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
@ -96,7 +97,7 @@ public class AccordCommandTest
public void basicCycleTest() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
getUninterruptibly(commandStore.execute((PreLoadContext.Empty)() -> "Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(1);
@ -171,7 +172,7 @@ public class AccordCommandTest
Commit commit = Commit.SerializerSupport.create(txnId, route, 1, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute);
getUninterruptibly(commandStore.execute(commit, commit::apply));
getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(txnId, Keys.of(key).toParticipants(), KeyHistory.SYNC), safeStore -> {
getUninterruptibly(commandStore.execute(PreLoadContext.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));
@ -188,7 +189,7 @@ public class AccordCommandTest
public void computeDeps() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
getUninterruptibly(commandStore.execute((PreLoadContext.Empty)()->"Test", unused -> commandStore.executor().cacheUnsafe().setCapacity(0)));
TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(2);

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

@ -86,7 +86,8 @@ import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
import static accord.local.KeyHistory.SYNC;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.PreLoadContext.contextFor;
import static accord.utils.Property.qt;
import static accord.utils.async.AsyncChains.getUninterruptibly;
@ -127,7 +128,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getUninterruptibly(commandStore.execute(txnId, instance -> {
getUninterruptibly(commandStore.execute(PreLoadContext.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 +142,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getUninterruptibly(commandStore.execute(txnId, safe -> {
getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(txnId, "Test"), safe -> {
StoreParticipants participants = StoreParticipants.empty(txnId);
SafeCommand command = safe.get(txnId, participants);
Assert.assertNotNull(command);
@ -155,7 +156,7 @@ public class AccordTaskTest
Txn txn = AccordTestUtils.createWriteTxn((int)clock.incrementAndGet());
TokenKey key = ((PartitionKey) Iterables.getOnlyElement(txn.keys())).toUnseekable();
getUninterruptibly(commandStore.execute(contextFor(key), instance -> {
getUninterruptibly(commandStore.execute((PreLoadContext.Empty)() -> "Test", instance -> {
SafeCommandsForKey cfk = instance.ifLoadedAndInitialised(key);
Assert.assertNull(cfk);
}));
@ -200,7 +201,7 @@ public class AccordTaskTest
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC), safe -> {
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
@ -250,7 +251,7 @@ public class AccordTaskTest
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC), safe -> {
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
@ -302,7 +303,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);
PreLoadContext 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);
@ -367,7 +368,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);
PreLoadContext 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;

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

@ -216,7 +216,7 @@ public class AccordTestUtils
public static AsyncChain<Pair<Writes, Result>> processTxnResult(AccordCommandStore commandStore, TxnId txnId, PartialTxn txn, Timestamp executeAt) throws Throwable
{
AtomicReference<AsyncChain<Pair<Writes, Result>>> result = new AtomicReference<>();
getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(txn.keys().toParticipants()),
getUninterruptibly(commandStore.execute((PreLoadContext.Empty)() -> "Test",
safeStore -> result.set(processTxnResultDirect(safeStore, txnId, txn, executeAt))));
return result.get();
}
@ -394,7 +394,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), Collections.emptySet()));
AccordCommandStore store = createAccordCommandStore(node, now, topology, loadExecutor, saveExecutor);
store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).executor().cacheUnsafe().setCapacity(1 << 20));
store.execute((PreLoadContext.Empty)()->"Test", safeStore -> ((AccordCommandStore)safeStore.commandStore()).executor().cacheUnsafe().setCapacity(1 << 20));
return store;
}

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

@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord;
import org.junit.Test;
import accord.local.PreLoadContext;
import accord.local.StoreParticipants;
import accord.primitives.SaveStatus;
import accord.primitives.TxnId;
@ -40,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(id, (safe) -> {
instance.process(PreLoadContext.contextFor(id, "Test"), (safe) -> {
var safeCommand = safe.get(id, StoreParticipants.empty(id));
var command = safeCommand.current();
Assertions.assertThat(command.saveStatus()).isEqualTo(SaveStatus.Uninitialised);

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

@ -106,7 +106,7 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
{
case Task:
{
PreLoadContext ctx = PreLoadContext.contextFor(unseekablesGen.next(rs));
PreLoadContext ctx = (PreLoadContext.Empty)()->"Test";
instance.maybeCacheEvict(ctx.keys());
operation(instance, ctx, actionGen.next(rs), rs::nextBoolean).chain().begin(counter);
}

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

@ -521,7 +521,7 @@ public class CommandsForKeySerializerTest
{
int next = source.nextInt(commands.size());
Command command = commands.get(next);
cfk = cfk.update(new TestSafeCommandStore(command.txnId()), command).cfk();
cfk = cfk.update(new TestSafeCommandStore(PreLoadContext.contextFor(command.txnId(), "Test")), command).cfk();
commands.set(next, commands.get(commands.size() - 1));
commands.remove(commands.size() - 1);
}