From 0e9f0fab780d52d9d5496927f1117a112c746c30 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Mon, 30 Jun 2025 10:35:43 +0100 Subject: [PATCH] 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 --- modules/accord | 2 +- .../service/accord/AccordCommandStore.java | 2 +- .../service/accord/AccordService.java | 7 ++- .../cassandra/service/accord/AccordTask.java | 57 +++++++++---------- .../service/accord/CommandsForRanges.java | 7 ++- .../test/accord/AccordBootstrapTest.java | 6 +- .../test/accord/AccordDropTableBase.java | 5 +- .../accord/AccordIncrementalRepairTest.java | 7 ++- .../CompactionAccordIteratorsTest.java | 15 ++--- .../db/virtual/AccordDebugKeyspaceTest.java | 3 +- .../accord/AccordCommandStoreTest.java | 5 +- .../service/accord/AccordCommandTest.java | 9 +-- .../service/accord/AccordTaskTest.java | 17 +++--- .../service/accord/AccordTestUtils.java | 4 +- ...SimpleSimulatedAccordCommandStoreTest.java | 3 +- .../accord/SimulatedAccordTaskTest.java | 2 +- .../CommandsForKeySerializerTest.java | 2 +- 17 files changed, 79 insertions(+), 74 deletions(-) diff --git a/modules/accord b/modules/accord index 1c203efdd4..a1c1ed91cf 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 1c203efdd463f2c452e8072267370f9e8dfdb8f3 +Subproject commit a1c1ed91cfeb904be9b490d498c1c52aba2253c6 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 6847437f2a..e25f7a3827 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -507,7 +507,7 @@ public class AccordCommandStore extends CommandStore @Override public AsyncChain 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(); }); diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 786ab75058..1d1928e4a6 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -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(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index 3d54a289b1..4edae67a2a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -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 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 extends SubmittableTask implements Runnable, private void setupKeyLoadsExclusive(Caches caches, Iterable 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 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 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 extends SubmittableTask implements Runnable, @Override public String description() { - return toString(); + return preLoadContext.describe(); } } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index d51b297932..880afa9942 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -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 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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java index 30f4346cf8..10333ad0a1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java @@ -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)) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java index 416fe065fb..9c9835e4a8 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java @@ -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()) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java index 387d8ce872..04a6dca76d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java @@ -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 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)) diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index f458ac8461..3b5292ef55 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 08163de516..56abfe10e8 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -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); })); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java index 05ca3adc00..a44196591b 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index af94147517..7a612c1aa8 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java index f5f36b4b2c..0dbc166b43 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java @@ -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 consumer = Mockito.mock(Consumer.class); Map 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 consumer = Mockito.mock(Consumer.class); String errorMsg = "txn_ids " + ids; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 10afacf74c..543cd0af70 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -216,7 +216,7 @@ public class AccordTestUtils public static AsyncChain> processTxnResult(AccordCommandStore commandStore, TxnId txnId, PartialTxn txn, Timestamp executeAt) throws Throwable { AtomicReference>> 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; } diff --git a/test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java index a8523e469e..b8016430c1 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java index 035df8d1e2..d5454e7f77 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java @@ -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); } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 03c327bca8..42e62de3ee 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -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); }