From 54b2f6acbcec5b795835c8ad7ee0486c943e8202 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Tue, 29 Jul 2025 10:32:24 +0100 Subject: [PATCH] Improve: - system_accord_debug.txn_update->txn_ops + force_apply, force_update, try_execute - system_accord_debug.commands_for_key - system_accord_debug.commands_for_key_unmanaged - system_accord_debug.redundant_before + locally_durable_to_X_store - flush accord caches on shutdown patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20814 --- .../apache/cassandra/config/AccordSpec.java | 7 + .../db/virtual/AccordDebugKeyspace.java | 354 +++++++++++++++--- .../org/apache/cassandra/journal/Params.java | 3 + .../cassandra/service/StorageService.java | 14 +- .../service/accord/AccordCommandStore.java | 16 +- .../service/accord/AccordKeyspace.java | 10 + .../service/accord/AccordService.java | 43 ++- .../service/accord/IAccordService.java | 25 ++ .../service/accord/api/TokenKey.java | 26 ++ .../config/DatabaseDescriptorRefTest.java | 1 + .../db/virtual/AccordDebugKeyspaceTest.java | 9 +- .../apache/cassandra/journal/TestParams.java | 6 + 12 files changed, 453 insertions(+), 61 deletions(-) diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index 79e6d62211..c34d3fab49 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -188,6 +188,7 @@ public class AccordSpec { public int segmentSize = 32 << 20; public FailurePolicy failurePolicy = FailurePolicy.STOP; + public ReplayMode replayMode = ReplayMode.ONLY_NON_DURABLE; public FlushMode flushMode = FlushMode.PERIODIC; public volatile DurationSpec flushPeriod; // pulls default from 'commitlog_sync_period' public DurationSpec periodicFlushLagBlock = new DurationSpec.IntMillisecondsBound("1500ms"); @@ -220,6 +221,12 @@ public class AccordSpec return failurePolicy; } + @Override + public ReplayMode replayMode() + { + return replayMode; + } + @Override public FlushMode flushMode() { diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index da7029c1fe..ae72d01d36 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -31,6 +31,7 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.function.Function; import java.util.stream.Collectors; @@ -51,18 +52,26 @@ import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.Commands; import accord.local.DurableBefore; +import accord.local.LoadKeys; +import accord.local.LoadKeysFor; import accord.local.MaxConflicts; import accord.local.PreLoadContext; import accord.local.RejectBefore; import accord.local.SafeCommand; import accord.local.SafeCommandStore; import accord.local.StoreParticipants; +import accord.local.cfk.CommandsForKey; +import accord.local.cfk.SafeCommandsForKey; import accord.local.durability.ShardDurability; import accord.primitives.Participants; +import accord.primitives.SaveStatus; import accord.primitives.Status; import accord.primitives.TxnId; import accord.utils.Invariants; +import accord.utils.UnhandledEnum; +import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DataRange; @@ -97,12 +106,15 @@ import org.apache.cassandra.utils.LocalizeString; import static accord.local.RedundantStatus.Property.GC_BEFORE; import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED; +import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE; +import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE; import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT; import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED; import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED; import static accord.local.RedundantStatus.Property.MAJORITY_APPLIED; import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP; import static accord.local.RedundantStatus.Property.SHARD_APPLIED; +import static accord.utils.async.AsyncChains.getBlockingAndRethrow; import static com.google.common.collect.ImmutableList.toImmutableList; import static java.lang.String.format; import static java.util.concurrent.TimeUnit.NANOSECONDS; @@ -112,29 +124,34 @@ import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime; public class AccordDebugKeyspace extends VirtualKeyspace { + public static final String COMMANDS_FOR_KEY = "commands_for_key"; + public static final String COMMANDS_FOR_KEY_UNMANAGED = "commands_for_key_unmanaged"; public static final String DURABILITY_SERVICE = "durability_service"; public static final String DURABLE_BEFORE = "durable_before"; public static final String EXECUTOR_CACHE = "executor_cache"; + public static final String JOURNAL = "journal"; public static final String MAX_CONFLICTS = "max_conflicts"; public static final String MIGRATION_STATE = "migration_state"; public static final String PROGRESS_LOG = "progress_log"; public static final String REDUNDANT_BEFORE = "redundant_before"; public static final String REJECT_BEFORE = "reject_before"; - public static final String TXN_BLOCKED_BY = "txn_blocked_by"; - public static final String JOURNAL = "journal"; public static final String TXN = "txn"; - public static final String TXN_UPDATE = "txn_update"; + public static final String TXN_BLOCKED_BY = "txn_blocked_by"; public static final String TXN_TRACE = "txn_trace"; public static final String TXN_TRACES = "txn_traces"; + public static final String TXN_OPS = "txn_ops"; public static final AccordDebugKeyspace instance = new AccordDebugKeyspace(); private AccordDebugKeyspace() { super(VIRTUAL_ACCORD_DEBUG, List.of( + new CommandsForKeyTable(), + new CommandsForKeyUnmanagedTable(), new DurabilityServiceTable(), new DurableBeforeTable(), new ExecutorCacheTable(), + new JournalTable(), new MaxConflictsTable(), new MigrationStateTable(), new ProgressLogTable(), @@ -142,13 +159,212 @@ public class AccordDebugKeyspace extends VirtualKeyspace new RejectBeforeTable(), new TxnBlockedByTable(), new TxnTable(), - new JournalTable(), - new TxnUpdateTable(), new TxnTraceTable(), - new TxnTracesTable() + new TxnTracesTable(), + new TxnOpsTable() )); } + // TODO (desired): don't report null as "null" + public static final class CommandsForKeyTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet + { + static class Entry + { + final int commandStoreId; + final CommandsForKey cfk; + + Entry(int commandStoreId, CommandsForKey cfk) + { + this.commandStoreId = commandStoreId; + this.cfk = cfk; + } + } + private CommandsForKeyTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, COMMANDS_FOR_KEY, + "Accord per-CommandStore CommandsForKey Managed Transaction State", + "CREATE TABLE %s (\n" + + " key text,\n" + + " command_store_id int,\n" + + " txn_id 'TxnIdUtf8Type',\n" + + " ballot text,\n" + + " deps_known_before text,\n" + + " execute_at text,\n" + + " flags text,\n" + + " missing text,\n" + + " status text,\n" + + " status_overrides text,\n" + + " PRIMARY KEY (key, command_store_id, txn_id)" + + ')', UTF8Type.instance)); + } + + @Override + public DataSet data() + { + return this; + } + + @Override + public boolean isEmpty() + { + return false; + } + + @Override + public Partition getPartition(DecoratedKey partitionKey) + { + String keyStr = UTF8Type.instance.compose(partitionKey.getKey()); + TokenKey key = TokenKey.parse(keyStr, DatabaseDescriptor.getPartitioner()); + + List cfks = new CopyOnWriteArrayList<>(); + PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key table query"); + CommandStores commandStores = AccordService.instance().node().commandStores(); + getBlockingAndRethrow(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> { + SafeCommandsForKey safeCfk = safeStore.get(key); + CommandsForKey cfk = safeCfk.current(); + if (cfk == null) + return; + + cfks.add(new Entry(safeStore.commandStore().id(), cfk)); + }).beginAsResult()); + + if (cfks.isEmpty()) + return null; + + SimpleDataSet ds = new SimpleDataSet(metadata); + for (Entry e : cfks) + { + CommandsForKey cfk = e.cfk; + for (int i = 0 ; i < cfk.size() ; ++i) + { + CommandsForKey.TxnInfo txn = cfk.get(i); + ds.row(keyStr, e.commandStoreId, toStringOrNull(txn.plainTxnId())) + .column("ballot", toStringOrNull(txn.ballot())) + .column("deps_known_before", toStringOrNull(txn.depsKnownUntilExecuteAt())) + .column("flags", flags(txn)) + .column("execute_at", toStringOrNull(txn.plainExecuteAt())) + .column("missing", Arrays.toString(txn.missing())) + .column("status", toStringOrNull(txn.status())) + .column("status_overrides", txn.statusOverrides() == 0 ? null : ("0x" + Integer.toHexString(txn.statusOverrides()))); + } + } + + return ds.getPartition(partitionKey); + } + + @Override + public Iterator getPartitions(DataRange range) + { + throw new UnsupportedOperationException(); + } + + private static String flags(CommandsForKey.TxnInfo txn) + { + StringBuilder sb = new StringBuilder(); + if (!txn.mayExecute()) + { + sb.append("NO EXECUTE"); + } + if (txn.hasNotifiedReady()) + { + if (sb.length() > 0) sb.append(", "); + sb.append("NOTIFIED READY"); + } + if (txn.hasNotifiedWaiting()) + { + if (sb.length() > 0) sb.append(", "); + sb.append("NOTIFIED WAITING"); + } + return sb.toString(); + } + } + + + // TODO (expected): test this table + public static final class CommandsForKeyUnmanagedTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet + { + static class Entry + { + final int commandStoreId; + final CommandsForKey cfk; + + Entry(int commandStoreId, CommandsForKey cfk) + { + this.commandStoreId = commandStoreId; + this.cfk = cfk; + } + } + private CommandsForKeyUnmanagedTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, COMMANDS_FOR_KEY_UNMANAGED, + "Accord per-CommandStore CommandsForKey Unmanaged Transaction State", + "CREATE TABLE %s (\n" + + " key text,\n" + + " command_store_id int,\n" + + " txn_id 'TxnIdUtf8Type',\n" + + " waiting_until text,\n" + + " waiting_until_status text,\n" + + " PRIMARY KEY (key, command_store_id, txn_id)" + + ')', UTF8Type.instance)); + } + + @Override + public DataSet data() + { + return this; + } + + @Override + public boolean isEmpty() + { + return false; + } + + @Override + public Partition getPartition(DecoratedKey partitionKey) + { + String keyStr = UTF8Type.instance.compose(partitionKey.getKey()); + TokenKey key = TokenKey.parse(keyStr, DatabaseDescriptor.getPartitioner()); + + List cfks = new CopyOnWriteArrayList<>(); + PreLoadContext context = PreLoadContext.contextFor(key, LoadKeys.SYNC, LoadKeysFor.READ_WRITE, "commands_for_key_unmanaged table query"); + CommandStores commandStores = AccordService.instance().node().commandStores(); + getBlockingAndRethrow(commandStores.forEach(context, key, Long.MIN_VALUE, Long.MAX_VALUE, safeStore -> { + SafeCommandsForKey safeCfk = safeStore.get(key); + CommandsForKey cfk = safeCfk.current(); + if (cfk == null) + return; + + cfks.add(new Entry(safeStore.commandStore().id(), cfk)); + })); + + if (cfks.isEmpty()) + return null; + + SimpleDataSet ds = new SimpleDataSet(metadata); + for (Entry e : cfks) + { + CommandsForKey cfk = e.cfk; + for (int i = 0 ; i < cfk.unmanagedCount() ; ++i) + { + CommandsForKey.Unmanaged txn = cfk.getUnmanaged(i); + ds.row(keyStr, e.commandStoreId, toStringOrNull(txn.txnId)) + .column("waiting_until", toStringOrNull(txn.waitingUntil)) + .column("waiting_until_status", toStringOrNull(txn.pending)); + } + } + + return ds.getPartition(partitionKey); + } + + @Override + public Iterator getPartitions(DataRange range) + { + throw new UnsupportedOperationException(); + } + } + + public static final class DurabilityServiceTable extends AbstractVirtualTable { private DurabilityServiceTable() @@ -436,6 +652,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + + " table_id text,\n" + " command_store_id int,\n" + " txn_id 'TxnIdUtf8Type',\n" + // Timer + BaseTxnState @@ -453,7 +670,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace " home_progress text,\n" + " home_retry_counter int,\n" + " home_scheduled_at timestamp,\n" + - " PRIMARY KEY (keyspace_name, table_name, command_store_id, txn_id)" + + " PRIMARY KEY (keyspace_name, table_name, table_id, command_store_id, txn_id)" + ')', UTF8Type.instance)); } @@ -466,10 +683,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace { DefaultProgressLog.ImmutableView view = ((DefaultProgressLog) commandStore.unsafeProgressLog()).immutableView(); TableId tableId = ((AccordCommandStore)commandStore).tableId(); + String tableIdStr = tableId.toString(); TableMetadata tableMetadata = tableMetadata(tableId); while (view.advance()) { - ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), view.commandStoreId(), view.txnId().toString()) + ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), tableIdStr, view.commandStoreId(), view.txnId().toString()) .column("contact_everyone", view.contactEveryone()) .column("waiting_is_uninitialised", view.isWaitingUninitialised()) .column("waiting_blocked_until", view.waitingIsBlockedUntil().name()) @@ -507,6 +725,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + + " table_id text,\n" + " token_start 'TokenUtf8Type',\n" + " token_end 'TokenUtf8Type',\n" + " command_store_id int,\n" + @@ -516,12 +735,14 @@ public class AccordDebugKeyspace extends VirtualKeyspace " shard_applied 'TxnIdUtf8Type',\n" + " majority_applied 'TxnIdUtf8Type',\n" + " locally_applied 'TxnIdUtf8Type',\n" + - " locally_synced 'TxnIdUtf8Type',\n" + + " locally_durable_to_command_store 'TxnIdUtf8Type',\n" + + " locally_durable_to_data_store 'TxnIdUtf8Type',\n" + " locally_redundant 'TxnIdUtf8Type',\n" + + " locally_synced 'TxnIdUtf8Type',\n" + " locally_witnessed 'TxnIdUtf8Type',\n" + " pre_bootstrap 'TxnIdUtf8Type',\n" + " stale_until_at_least 'TxnIdUtf8Type',\n" + - " PRIMARY KEY (keyspace_name, table_name, command_store_id, token_start)" + + " PRIMARY KEY (keyspace_name, table_name, table_id, command_store_id, token_start)" + ')', UTF8Type.instance)); } @@ -535,12 +756,13 @@ public class AccordDebugKeyspace extends VirtualKeyspace { int commandStoreId = commandStore.id(); TableId tableId = ((AccordCommandStore)commandStore).tableId(); + String tableIdStr = tableId.toString(); TableMetadata tableMetadata = tableMetadata(tableId); String keyspace = keyspace(tableMetadata); String table = table(tableId, tableMetadata); commandStore.unsafeGetRedundantBefore().foldl( (entry, ds) -> { - ds.row(keyspace, table, commandStoreId, printToken(entry.range.start())) + ds.row(keyspace, table, tableIdStr, commandStoreId, printToken(entry.range.start())) .column("token_end", printToken(entry.range.end())) .column("start_epoch", entry.startEpoch) .column("end_epoch", entry.endEpoch) @@ -548,8 +770,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace .column("shard_applied", entry.maxBound(SHARD_APPLIED).toString()) .column("majority_applied", entry.maxBound(MAJORITY_APPLIED).toString()) .column("locally_applied", entry.maxBound(LOCALLY_APPLIED).toString()) - .column("locally_synced", entry.maxBound(LOCALLY_SYNCED).toString()) + .column("locally_durable_to_command_store", entry.maxBound(LOCALLY_DURABLE_TO_COMMAND_STORE).toString()) + .column("locally_durable_to_data_store", entry.maxBound(LOCALLY_DURABLE_TO_DATA_STORE).toString()) .column("locally_redundant", entry.maxBound(LOCALLY_REDUNDANT).toString()) + .column("locally_synced", entry.maxBound(LOCALLY_SYNCED).toString()) .column("locally_witnessed", entry.maxBound(LOCALLY_WITNESSED).toString()) .column("pre_bootstrap", entry.maxBound(PRE_BOOTSTRAP).toString()) .column("stale_until_at_least", entry.staleUntilAtLeast != null ? entry.staleUntilAtLeast.toString() : null); @@ -572,11 +796,12 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + + " table_id text,\n" + " command_store_id int,\n" + " token_start 'TokenUtf8Type',\n" + " token_end 'TokenUtf8Type',\n" + " timestamp text,\n" + - " PRIMARY KEY (keyspace_name, table_name, command_store_id, token_start)" + + " PRIMARY KEY (keyspace_name, table_name, table_id, command_store_id, token_start)" + ')', UTF8Type.instance)); } @@ -592,12 +817,12 @@ public class AccordDebugKeyspace extends VirtualKeyspace continue; TableId tableId = ((AccordCommandStore)commandStore).tableId(); + String tableIdStr = tableId.toString(); TableMetadata tableMetadata = tableMetadata(tableId); String keyspace = keyspace(tableMetadata); String table = table(tableId, tableMetadata); rejectBefore.foldlWithBounds( - (timestamp, ds, start, end) -> ds.row(keyspace, table, sortToken(start), commandStore.id()) - .column("token_start", printToken(start)) + (timestamp, ds, start, end) -> ds.row(keyspace, table, tableIdStr, commandStore.id(), printToken(start)) .column("token_end", printToken(end)) .column("timestamp", timestamp.toString()) , @@ -775,8 +1000,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace super(parse(VIRTUAL_ACCORD_DEBUG, TXN, "Accord per-CommandStore Transaction State", "CREATE TABLE %s (\n" + - " txn_id text,\n" + " command_store_id int,\n" + + " txn_id text,\n" + " save_status text,\n" + " route text,\n" + " durability text,\n" + @@ -814,7 +1039,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace String txnIdStr = UTF8Type.instance.compose(partitionKey.getKey()); TxnId txnId = TxnId.parse(txnIdStr); - List commands = new ArrayList<>(); + List commands = new CopyOnWriteArrayList<>(); AccordService.instance().node().commandStores().forEachCommandStore(store -> { Command command = ((AccordCommandStore)store).loadCommand(txnId); if (command != null) @@ -973,17 +1198,18 @@ public class AccordDebugKeyspace extends VirtualKeyspace */ // Had to be separate from the "regular" journal table since it does not have segment and position, and command store id is inferred // TODO (required): add access control - public static final class TxnUpdateTable extends AbstractMutableVirtualTable implements AbstractVirtualTable.DataSet + public static final class TxnOpsTable extends AbstractMutableVirtualTable implements AbstractVirtualTable.DataSet { - private static Set ALLOWED = new HashSet<>(Arrays.asList("ERASE_VESTIGIAL", "INVALIDATE")); - private TxnUpdateTable() + // TODO (expected): test each of these operations + enum Op { ERASE_VESTIGIAL, INVALIDATE, TRY_EXECUTE, FORCE_APPLY, FORCE_UPDATE } + private TxnOpsTable() { - super(parse(VIRTUAL_ACCORD_DEBUG, TXN_UPDATE, + super(parse(VIRTUAL_ACCORD_DEBUG, TXN_OPS, "Update Accord Command State", "CREATE TABLE %s (\n" + " txn_id text,\n" + " command_store_id int,\n" + - " cleanup text,\n" + + " op text,\n" + " PRIMARY KEY (txn_id, command_store_id)" + ')', UTF8Type.instance)); } @@ -991,7 +1217,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace @Override public DataSet data() { - throw new UnsupportedOperationException(TXN_UPDATE + " is a write-only table"); + throw new UnsupportedOperationException(TXN_OPS + " is a write-only table"); } @Override @@ -1003,13 +1229,13 @@ public class AccordDebugKeyspace extends VirtualKeyspace @Override public Partition getPartition(DecoratedKey partitionKey) { - throw new UnsupportedOperationException(TXN_UPDATE + " is a write-only table"); + throw new UnsupportedOperationException(TXN_OPS + " is a write-only table"); } @Override public Iterator getPartitions(DataRange range) { - throw new UnsupportedOperationException(TXN_UPDATE + " is a write-only table"); + throw new UnsupportedOperationException(TXN_OPS + " is a write-only table"); } @@ -1017,40 +1243,66 @@ public class AccordDebugKeyspace extends VirtualKeyspace protected void applyColumnUpdate(ColumnValues partitionKey, ColumnValues clusteringColumns, Optional columnValue) { TxnId txnId = TxnId.parse(partitionKey.value(0)); - Invariants.require(columnValue.isPresent()); - Cleanup cleanup; - switch ((String) columnValue.get().value()) - { - default: throw new IllegalStateException(String.format("Unknown Cleanup %s. Allowed: %s", columnValue.get().value(), ALLOWED)); - case "ERASE_VESTIGIAL": cleanup = Cleanup.VESTIGIAL; break; - case "INVALIDATE": cleanup = Cleanup.INVALIDATE; break; - } - int commandStoreId = clusteringColumns.value(0); - Invariants.require(commandStoreId >= 0); + Invariants.require(columnValue.isPresent()); + Op op = Op.valueOf(columnValue.get().value()); + switch (op) + { + default: throw new UnhandledEnum(op); + case ERASE_VESTIGIAL: + cleanup(txnId, commandStoreId, Cleanup.VESTIGIAL); + break; + case INVALIDATE: + cleanup(txnId, commandStoreId, Cleanup.INVALIDATE); + break; + case TRY_EXECUTE: + run(txnId, commandStoreId, safeStore -> { + SafeCommand safeCommand = safeStore.unsafeGet(txnId); + Commands.maybeExecute(safeStore, safeCommand, true, true); + return AsyncChains.success(null); + }); + break; + case FORCE_UPDATE: + run(txnId, commandStoreId, safeStore -> { + SafeCommand safeCommand = safeStore.unsafeGet(txnId); + safeCommand.update(safeStore, safeCommand.current(), true); + return AsyncChains.success(null); + }); + break; + case FORCE_APPLY: + run(txnId, commandStoreId, safeStore -> { + SafeCommand safeCommand = safeStore.unsafeGet(txnId); + Command command = safeCommand.current(); + // TODO (expected): we can call applyChain with TruncatedApplyWithOutcome in theory, but the type signature prevents it + if (command.saveStatus().compareTo(SaveStatus.PreApplied) < 0 || command.saveStatus().compareTo(SaveStatus.TruncatedApplyWithOutcome) >= 0) + throw new UnsupportedOperationException("Cannot apply a transaction with saveStatus " + command.saveStatus()); + return Commands.applyChain(safeStore, (Command.Executed) command); + }); + break; + } + } + private void run(TxnId txnId, int commandStoreId, Function> apply) + { AccordService accord = (AccordService) AccordService.instance(); AsyncChains.awaitUninterruptibly(accord.node() .commandStores() .forId(commandStoreId) - .submit(PreLoadContext.contextFor(txnId, "txn_update"), new Function() - { - @Override - public Void apply(SafeCommandStore safeStore) - { - SafeCommand safeCommand = safeStore.unsafeGet(txnId); - Command command = safeCommand.current(); - Command updated = Commands.purge(safeStore, - command, - command.participants(), - cleanup, - true); - safeCommand.update(safeStore, updated); - return null; - } - }) + .submit(PreLoadContext.contextFor(txnId, TXN_OPS), apply) + .flatMap(i -> i) .beginAsResult()); } + + private void cleanup(TxnId txnId, int commandStoreId, Cleanup cleanup) + { + run(txnId, commandStoreId, safeStore -> { + SafeCommand safeCommand = safeStore.unsafeGet(txnId); + Command command = safeCommand.current(); + Command updated = Commands.purge(safeStore, command, command.participants(), cleanup, true); + safeCommand.update(safeStore, updated); + return AsyncChains.success(null); + }); + } } public static class TxnBlockedByTable extends AbstractVirtualTable diff --git a/src/java/org/apache/cassandra/journal/Params.java b/src/java/org/apache/cassandra/journal/Params.java index 29ba152e17..1e898ebce4 100644 --- a/src/java/org/apache/cassandra/journal/Params.java +++ b/src/java/org/apache/cassandra/journal/Params.java @@ -24,6 +24,7 @@ public interface Params enum FlushMode { BATCH, GROUP, PERIODIC } enum FailurePolicy { STOP, STOP_JOURNAL, IGNORE, ALLOW_UNSAFE_STARTUP, DIE } + enum ReplayMode { RESET, ALL, ONLY_NON_DURABLE } /** * @return maximum segment size @@ -40,6 +41,8 @@ public interface Params */ FlushMode flushMode(); + ReplayMode replayMode(); + boolean enableCompaction(); long compactionPeriod(TimeUnit units); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 54f6c0674a..87ce92f8f7 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -171,6 +171,7 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.TableMetadataRef; import org.apache.cassandra.schema.ViewMetadata; +import org.apache.cassandra.service.accord.AccordKeyspace.AccordColumnFamilyStores; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget; @@ -250,6 +251,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.PAXOS_REPA import static org.apache.cassandra.config.CassandraRelevantProperties.PAXOS_REPAIR_ON_TOPOLOGY_CHANGE_RETRY_DELAY_SECONDS; import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACE_ADDRESS_FIRST_BOOT; import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_WRITE_SURVEY; +import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.INTERNALLY_FORCED; import static org.apache.cassandra.index.SecondaryIndexManager.getIndexName; import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily; import static org.apache.cassandra.io.util.FileUtils.ONE_MIB; @@ -3823,7 +3825,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } if (AccordService.isSetup()) - AccordService.instance().shutdownAndWait(1, MINUTES); + AccordService.instance().markShuttingDown(); // In-progress writes originating here could generate hints to be written, // which is currently scheduled on the mutation stage. So shut down MessagingService @@ -3839,6 +3841,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE logger.error("Messaging service timed out shutting down", t); } + if (AccordService.isSetup()) + { + logger.info("Flushing Accord caches"); + if (!AccordService.instance().flushCaches().awaitUninterruptibly(1, MINUTES)) + logger.error("Could not flush Accord caches promptly"); + if (AccordColumnFamilyStores.commandsForKey != null) + AccordColumnFamilyStores.commandsForKey.forceBlockingFlush(INTERNALLY_FORCED); + AccordService.instance().shutdownAndWait(1, MINUTES); + } + // ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it. ScheduledExecutors.optionalTasks.shutdown(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index ef3a5b848a..f337f8f738 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -83,6 +83,7 @@ import static accord.api.Journal.CommandUpdate; import static accord.api.Journal.FieldUpdates; import static accord.api.Journal.Load.MINIMAL; import static accord.utils.Invariants.require; +import static org.apache.cassandra.journal.Params.ReplayMode.ONLY_NON_DURABLE; public class AccordCommandStore extends CommandStore { @@ -496,7 +497,10 @@ public class AccordCommandStore extends CommandStore public AccordCommandStoreReplayer replayer() { - return new AccordCommandStoreReplayer(this); + boolean replayOnlyDurable = true; + if (journal instanceof AccordJournal) + replayOnlyDurable = ((AccordJournal)journal).configuration().replayMode() == ONLY_NON_DURABLE; + return new AccordCommandStoreReplayer(this, replayOnlyDurable); } static final AtomicLong nextDurabilityLoggingId = new AtomicLong(); @@ -560,23 +564,23 @@ public class AccordCommandStore extends CommandStore public static class AccordCommandStoreReplayer extends AbstractReplayer { private final AccordCommandStore store; + private final boolean onlyNonDurable; - private AccordCommandStoreReplayer(AccordCommandStore store) + private AccordCommandStoreReplayer(AccordCommandStore store, boolean onlyNonDurable) { super(store.unsafeGetRedundantBefore()); this.store = store; + this.onlyNonDurable = onlyNonDurable; } @Override public AsyncChain replay(TxnId txnId) { - if (!maybeShouldReplay(txnId)) - { + if (onlyNonDurable && !maybeShouldReplay(txnId)) return AsyncChains.success(null); - } return store.submit(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> { - if (!shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants())) + if (onlyNonDurable && !shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants())) return null; initialiseState(safeStore, txnId); diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index 9646d63c6c..356200da23 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -510,6 +510,16 @@ public class AccordKeyspace return TABLES; } + public static void truncateCommandsForKey() + { + Keyspace ks = Keyspace.open(ACCORD_KEYSPACE_NAME); + for (String table : new String[]{ CommandsForKeys.name }) + { + if (!ks.getColumnFamilyStore(table).isEmpty()) + ks.getColumnFamilyStore(table).truncateBlocking(); + } + } + private static ByteBuffer cellValue(Cell cell) { return cell.accessor().toBuffer(cell.value()); diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 4432beb0aa..8db4a44784 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -87,6 +87,7 @@ import accord.utils.Invariants; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; +import accord.utils.async.AsyncResults; import org.apache.cassandra.concurrent.Shutdownable; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.DatabaseDescriptor; @@ -149,6 +150,7 @@ import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurab import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurabilityMaxSplits; import static org.apache.cassandra.config.DatabaseDescriptor.getAccordShardDurabilityTargetSplits; import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; +import static org.apache.cassandra.journal.Params.ReplayMode.RESET; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadBookkeeping; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteBookkeeping; import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.ImmutableTopoloyImage; @@ -262,6 +264,9 @@ public class AccordService implements IAccordService, Shutdownable CommandsForKey.disableLinearizabilityViolationsReporting(); try { + if (as.journalConfiguration().replayMode() == RESET) + AccordKeyspace.truncateCommandsForKey(); + as.node.commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().stop()); as.journal().replay(as.node().commandStores()); logger.info("Waiting for command stores to quiesce."); @@ -696,10 +701,46 @@ public class AccordService implements IAccordService, Shutdownable return scheduler.isTerminated(); } + public synchronized Future flushCaches() + { + class Ready extends AsyncResults.CountingResult implements Runnable + { + public Ready() { super(1); } + @Override public void run() { decrement(); } + } + Ready ready = new Ready(); + AccordCommandStores commandStores = (AccordCommandStores) node.commandStores(); + AsyncChains.getBlockingAndRethrow(commandStores.forEach((PreLoadContext.Empty)() -> "Flush Caches", safeStore -> { + AccordCommandStore commandStore = (AccordCommandStore)safeStore.commandStore(); + try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches()) + { + caches.commandsForKeys().forEach(entry -> { + if (entry.isModified()) + { + ready.increment(); + caches.global().saveWhenReadyExclusive(entry, ready); + } + }); + } + })); + ready.decrement(); + AsyncPromise result = new AsyncPromise<>(); + ready.begin((success, fail) -> { + if (fail != null) result.tryFailure(fail); + else result.trySuccess(null); + }); + return result; + } + + public synchronized void markShuttingDown() + { + state = State.SHUTTING_DOWN; + } + @Override public synchronized void shutdown() { - if (state != State.STARTED) + if (state != State.STARTED && state != State.SHUTTING_DOWN) return; state = State.SHUTTING_DOWN; shutdownAndWait(1, TimeUnit.MINUTES); diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 2e742fd934..40c0e57a4c 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -114,6 +114,8 @@ public interface IAccordService void startup(); + Future flushCaches(); + void markShuttingDown(); void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException; AccordScheduler scheduler(); @@ -274,6 +276,17 @@ public interface IAccordService } } + @Override + public void markShuttingDown() + { + } + + @Override + public Future flushCaches() + { + return ImmediateFuture.success(null); + } + @Override public void shutdownAndWait(long timeout, TimeUnit unit) { } @@ -455,6 +468,18 @@ public interface IAccordService delegate.startup(); } + @Override + public Future flushCaches() + { + return delegate.flushCaches(); + } + + @Override + public void markShuttingDown() + { + delegate.markShuttingDown(); + } + @Override public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException { diff --git a/src/java/org/apache/cassandra/service/accord/api/TokenKey.java b/src/java/org/apache/cassandra/service/accord/api/TokenKey.java index 96bb73f418..b71be2f333 100644 --- a/src/java/org/apache/cassandra/service/accord/api/TokenKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/TokenKey.java @@ -147,6 +147,32 @@ public final class TokenKey extends AccordRoutableKey implements RoutingKey, Ran return prefix() + ":" + printableSuffix(); } + public static TokenKey parse(String str, IPartitioner partitioner) + { + TableId tableId; + { + int split = str.indexOf(':', str.startsWith("tid:") ? 4 : 0); + tableId = TableId.fromString(str.substring(0, split)); + str = str.substring(split + 1); + } + if (str.endsWith("Inf")) + { + return new TokenKey(tableId, str.charAt(0) == '-' ? MIN_TABLE_SENTINEL : MAX_TABLE_SENTINEL, partitioner.getMinimumToken()); + } + if (str.endsWith(")")) + { + boolean isBefore = str.startsWith("before("); + boolean isAfter = str.startsWith("after("); + if (isBefore || isAfter) + { + byte sentinel = isBefore ? BEFORE_TOKEN_SENTINEL : AFTER_TOKEN_SENTINEL; + str = str.substring(isBefore ? 7 : 6, str.length() - 1); + return new TokenKey(tableId, sentinel, partitioner.getTokenFactory().fromString(str)); + } + } + return new TokenKey(tableId, partitioner.getTokenFactory().fromString(str)); + } + public long estimatedSizeOnHeap() { return EMPTY_SIZE + token().getHeapSize(); diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index fd1b64083a..b16cdc9734 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -297,6 +297,7 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.journal.Params", "org.apache.cassandra.journal.Params$FailurePolicy", "org.apache.cassandra.journal.Params$FlushMode", + "org.apache.cassandra.journal.Params$ReplayMode", "org.apache.cassandra.locator.Endpoint", "org.apache.cassandra.locator.IEndpointSnitch", "org.apache.cassandra.locator.InetAddressAndPort", diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 039444d66e..19fcbfe529 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -75,6 +75,9 @@ public class AccordDebugKeyspaceTest extends CQLTester private static final String QUERY_TXN_BLOCKED_BY = String.format("SELECT * FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_BLOCKED_BY); + private static final String QUERY_COMMANDS_FOR_KEY = + String.format("SELECT txn_id, status FROM %s.%s WHERE key=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.COMMANDS_FOR_KEY); + private static final String QUERY_TXN = String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN); @@ -214,12 +217,14 @@ public class AccordDebugKeyspaceTest extends CQLTester AccordService accord = accord(); TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0); + String keyStr = txn.keys().get(0).toUnseekable().toString(); AsyncChains.getBlocking(accord.node().coordinate(id, txn)); spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()), row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name())))); assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied")); assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null)); + assertRows(execute(QUERY_COMMANDS_FOR_KEY, keyStr), row(id.toString(), "APPLIED_DURABLE")); } @Test @@ -330,7 +335,7 @@ public class AccordDebugKeyspaceTest extends CQLTester } catch (Throwable t) { - Assert.assertTrue(t.getMessage().contains("Unknown")); + Assert.assertTrue(t.getMessage().contains("No enum constant")); } } @@ -358,7 +363,7 @@ public class AccordDebugKeyspaceTest extends CQLTester assertRows(rs, row(id.toString(), "PreAccepted", anyNonNull())); int commandStoreId = rs.one().getInt("command_store_id"); - String PATCH_JOURNAL = String.format("UPDATE %s.%s SET cleanup = ? WHERE txn_id=? AND command_store_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_UPDATE); + String PATCH_JOURNAL = String.format("UPDATE %s.%s SET op = ? WHERE txn_id=? AND command_store_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_OPS); execute(PATCH_JOURNAL, cleanupAction, id.toString(), commandStoreId); assertRows(execute(QUERY_TXN, id.toString()), diff --git a/test/unit/org/apache/cassandra/journal/TestParams.java b/test/unit/org/apache/cassandra/journal/TestParams.java index 72c64b0e51..edf357a790 100644 --- a/test/unit/org/apache/cassandra/journal/TestParams.java +++ b/test/unit/org/apache/cassandra/journal/TestParams.java @@ -43,6 +43,12 @@ public class TestParams implements Params return FlushMode.GROUP; } + @Override + public ReplayMode replayMode() + { + return null; + } + @Override public boolean enableCompaction() {