You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/08/03 14:16:06 UTC
(cassandra) branch trunk updated (eb65c0e600 -> 420823afc5)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from eb65c0e600 Support manual secondary index selection at the CQL level
new 54b2f6acbc 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
new f02020f06d Improve ProgressLog: - Schedule a fallback timeout for active requests to ensure progress with lost callbacks - Clear active task for a txnId when new RunInvoker is registered - Consult DurableBefore prior to invoking recovery - Support user invoked reset of a command, clearing any active work and requeueing it - (Testing): Treat progress log for RX as recurring tasks for burn test termination Also: - Add TxnOps RECOVER, FETCH, RESET_PROGRESS_LOG
new 420823afc5 Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively, since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions Also Improve: - Apply MaxDecidedRX filtering to CommandsForKey RX dependencies - Optimise AbstractRanges.sliceMinimal (burn test hotspot) - Don't start shard schedulers on topology changes if not already started - [...]
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 11 +-
.../cassandra/config/DatabaseDescriptor.java | 22 +-
.../cassandra/db/virtual/AccordDebugKeyspace.java | 436 ++++++++++++++++++---
src/java/org/apache/cassandra/journal/Params.java | 3 +
.../apache/cassandra/service/StorageService.java | 14 +-
.../service/accord/AccordCommandStore.java | 71 +---
.../service/accord/AccordExecutorLoops.java | 7 -
.../cassandra/service/accord/AccordKeyspace.java | 13 +-
.../cassandra/service/accord/AccordService.java | 43 +-
.../cassandra/service/accord/AccordTask.java | 7 +-
.../service/accord/CommandsForRanges.java | 128 +-----
.../cassandra/service/accord/IAccordService.java | 25 ++
.../cassandra/service/accord/api/AccordAgent.java | 3 +-
.../cassandra/service/accord/api/TokenKey.java | 26 ++
.../test/accord/AccordDropTableBase.java | 6 +-
.../simulator/test/AccordHarrySimulationTest.java | 7 +
.../config/DatabaseDescriptorRefTest.java | 1 +
.../db/virtual/AccordDebugKeyspaceTest.java | 9 +-
.../org/apache/cassandra/journal/TestParams.java | 6 +
.../serializers/CommandsForKeySerializerTest.java | 1 -
21 files changed, 601 insertions(+), 240 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/03: 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
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 54b2f6acbcec5b795835c8ad7ee0486c943e8202
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Tue Jul 29 10:32:24 2025 +0100
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
---
.../org/apache/cassandra/config/AccordSpec.java | 7 +
.../cassandra/db/virtual/AccordDebugKeyspace.java | 350 ++++++++++++++++++---
src/java/org/apache/cassandra/journal/Params.java | 3 +
.../apache/cassandra/service/StorageService.java | 14 +-
.../service/accord/AccordCommandStore.java | 16 +-
.../cassandra/service/accord/AccordKeyspace.java | 10 +
.../cassandra/service/accord/AccordService.java | 43 ++-
.../cassandra/service/accord/IAccordService.java | 25 ++
.../cassandra/service/accord/api/TokenKey.java | 26 ++
.../config/DatabaseDescriptorRefTest.java | 1 +
.../db/virtual/AccordDebugKeyspaceTest.java | 9 +-
.../org/apache/cassandra/journal/TestParams.java | 6 +
12 files changed, 451 insertions(+), 59 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<Entry> 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<Partition> 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<Entry> 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<Partition> 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<Entry> commands = new ArrayList<>();
+ List<Entry> 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<String> 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<Partition> 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> columnValue)
{
TxnId txnId = TxnId.parse(partitionKey.value(0));
+ int commandStoreId = clusteringColumns.value(0);
Invariants.require(columnValue.isPresent());
- Cleanup cleanup;
- switch ((String) columnValue.get().value())
+ Op op = Op.valueOf(columnValue.get().value());
+ switch (op)
{
- 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;
+ 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;
}
+ }
- int commandStoreId = clusteringColumns.value(0);
- Invariants.require(commandStoreId >= 0);
-
+ private void run(TxnId txnId, int commandStoreId, Function<SafeCommandStore, AsyncChain<Void>> apply)
+ {
AccordService accord = (AccordService) AccordService.instance();
AsyncChains.awaitUninterruptibly(accord.node()
.commandStores()
.forId(commandStoreId)
- .submit(PreLoadContext.contextFor(txnId, "txn_update"), new Function<SafeCommandStore, Void>()
- {
- @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<Route> 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 <T> ByteBuffer cellValue(Cell<T> 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<Void> 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<Void> 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<Void> 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<Void> 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<Void> 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()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 03/03: Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively, since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions Also Improve: - Apply MaxDecidedRX filtering to CommandsForKey RX dependencies - Optimise AbstractRanges.sliceMinimal (burn test hotspot) - Don't start shard schedulers on topology changes if not already started - Only registerTransit [...]
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 420823afc59fc4d16d7168ad383a2f1d0109fabc
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Fri Aug 1 11:03:43 2025 +0100
Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively,
since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions
Also Improve:
- Apply MaxDecidedRX filtering to CommandsForKey RX dependencies
- Optimise AbstractRanges.sliceMinimal (burn test hotspot)
- Don't start shard schedulers on topology changes if not already started
- Only registerTransitive if waitingOnSync
- Reserve DurabilityRequest until ShardDurability is started
- Don't notify progress log if stopped
- Remove duplicated CFK unmanaged filtering
Also fix:
- Edge case in notification across bootstrap boundary by resubmitting Unmanaged to be recomputed
- Serialization of CommandsForKey.TxnInfo.missing() collection when non-identity flag bits are present
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20811
---
modules/accord | 2 +-
.../service/accord/AccordCommandStore.java | 51 +-------
.../service/accord/AccordExecutorLoops.java | 7 --
.../cassandra/service/accord/AccordKeyspace.java | 3 +-
.../cassandra/service/accord/AccordTask.java | 7 +-
.../service/accord/CommandsForRanges.java | 128 +++------------------
.../cassandra/service/accord/api/AccordAgent.java | 3 +-
.../test/accord/AccordDropTableBase.java | 6 +-
.../simulator/test/AccordHarrySimulationTest.java | 7 ++
.../serializers/CommandsForKeySerializerTest.java | 1 -
10 files changed, 38 insertions(+), 177 deletions(-)
diff --git a/modules/accord b/modules/accord
index 20df7e5e5a..d5de79b56f 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 20df7e5e5ad7bae442ff6fe4ee716eb7104336c1
+Subproject commit d5de79b56f792d3b6d868ca46845b830e8906828
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 972a7488e0..95aba5de55 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -55,11 +55,9 @@ import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.PartialTxn;
-import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
import accord.primitives.RoutableKey;
import accord.primitives.Route;
-import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@@ -226,14 +224,6 @@ public class AccordCommandStore extends CommandStore
return commandsForRanges;
}
- @Override
- public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges, Status.Durability durability)
- {
- super.markShardDurable(safeStore, globalSyncId, ranges, durability);
- if (durability.compareTo(Status.Durability.UniversalOrInvalidated) >= 0)
- commandsForRanges.gcBefore(globalSyncId, ranges);
- }
-
@Override
public boolean inStore()
{
@@ -400,43 +390,6 @@ public class AccordCommandStore extends CommandStore
{
}
- public void registerTransitive(SafeCommandStore safeStore, RangeDeps rangeDeps)
- {
- if (rangeDeps.isEmpty())
- return;
-
- RedundantBefore redundantBefore = unsafeGetRedundantBefore();
- CommandStores.RangesForEpoch ranges = safeStore.ranges();
- // used in places such as accord.local.CommandStore.fetchMajorityDeps
- // We find a set of dependencies for a range then update CommandsFor to know about them
- Ranges allRanges = safeStore.ranges().all();
- Ranges coordinateRanges = Ranges.EMPTY;
- long coordinateEpoch = -1;
- try (ExclusiveCaches caches = lockCaches())
- {
- for (int i = 0; i < rangeDeps.txnIdCount(); i++)
- {
- TxnId txnId = rangeDeps.txnId(i);
- AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
- if (state != null && state.isLoaded() && state.getExclusive() != null && state.getExclusive().known().isDefinitionKnown())
- continue;
-
- Ranges addRanges = rangeDeps.ranges(i).slice(allRanges);
- if (addRanges.isEmpty()) continue;
-
- if (coordinateEpoch != txnId.epoch())
- {
- coordinateEpoch = txnId.epoch();
- coordinateRanges = ranges.allAt(txnId.epoch());
- }
- if (addRanges.intersects(coordinateRanges)) continue;
- addRanges = redundantBefore.removeGcBefore(txnId, txnId, addRanges);
- if (addRanges.isEmpty()) continue;
- commandsForRanges().mergeTransitive(txnId, addRanges, Ranges::with);
- }
- }
- }
-
public void appendCommands(List<CommandUpdate> diffs, Runnable onFlush)
{
for (int i = 0; i < diffs.size(); i++)
@@ -668,8 +621,8 @@ public class AccordCommandStore extends CommandStore
StringBuilder sb = new StringBuilder("[");
if (metadata != null)
sb.append(metadata).append('|');
- sb.append(tableId.lsb());
- sb.append(',').append(id).append(',').append(node.id().id).append(']');
+ sb.append(tableId);
+ sb.append('|').append(id).append(',').append(node.id().id).append(']');
return sb.toString();
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java
index ff9169d08b..803839de94 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java
@@ -85,13 +85,6 @@ class AccordExecutorLoops
};
}
- AccordExecutor.Task activeTaskForCurrentThread()
- {
- Thread thread = Thread.currentThread();
- LoopTask task = tasks.get(thread.getId());
- return task == null ? null : task.runningTask();
- }
-
public Stream<? extends DebuggableTaskRunner> active()
{
return tasks.values().stream();
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index 356200da23..f0dfcf4c66 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -302,7 +302,8 @@ public class AccordKeyspace
if (current == null)
return null;
- CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore.gcBefore());
+ // TODO (desired): consider whether better to not compact any validation failures, since we expect is already overwritten
+ CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore.gcBefore(), false);
if (current == updated)
return row;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java
index 7a4fd0e8cf..f04d4ad7c0 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTask.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java
@@ -1038,7 +1038,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
{
CommandsForRanges.Summary summary = summaryLoader.ifRelevant(state);
if (summary != null)
- summaries.put(summary.txnId, summary);
+ summaries.put(summary.plainTxnId(), summary);
}
}
@@ -1059,7 +1059,10 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
CommandsForRanges.Summary summary = summaryLoader.load(txnId);
if (summary != null)
+ {
summaries.putIfAbsent(txnId, summary);
+ summaryLoader.maybeRecordFutureRx(summary);
+ }
});
}
@@ -1086,7 +1089,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
void startInternal(Caches caches)
{
summaryLoader = commandStore.commandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.loadKeysFor(), keysOrRanges);
- summaryLoader.forEachInCache(keysOrRanges, summary -> summaries.put(summary.txnId, summary), caches);
+ summaryLoader.forEachInCache(keysOrRanges, summary -> summaries.put(summary.plainTxnId(), summary), caches);
caches.commands().register(commandWatcher);
}
diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
index eddafc8398..b905ec519a 100644
--- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
+++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
@@ -21,18 +21,14 @@ package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.BiFunction;
import java.util.function.Consumer;
-import java.util.function.UnaryOperator;
import javax.annotation.Nullable;
import accord.api.RoutingKey;
@@ -49,7 +45,6 @@ import accord.primitives.RangeRoute;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Timestamp;
-import accord.primitives.Txn;
import accord.primitives.Txn.Kind.Kinds;
import accord.primitives.TxnId;
import accord.primitives.Unseekable;
@@ -65,7 +60,6 @@ import org.apache.cassandra.utils.btree.BTreeSet;
import org.apache.cassandra.utils.btree.IntervalBTree;
import org.apache.cassandra.utils.concurrent.IntrusiveStack;
-import static accord.local.CommandSummaries.SummaryStatus.NOT_DIRECTLY_WITNESSED;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.endWithStart;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.keyEndWithStart;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.keyStartWithEnd;
@@ -176,7 +170,6 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
private final AccordCommandStore commandStore;
private final RangeSearcher searcher;
- private final AtomicReference<NavigableMap<TxnId, Ranges>> transitive = new AtomicReference<>(new TreeMap<>());
// TODO (desired): manage memory consumed by this auxillary information
private final Object2ObjectHashMap<TxnId, RangeRoute> cachedRangeTxnsById = new Object2ObjectHashMap<>();
private Object[] cachedRangeTxnsByRange = IntervalBTree.empty();
@@ -356,80 +349,28 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
{
RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
- return Loader.loader(redundantBefore, primaryTxnId, loadKeysFor, keysOrRanges, this::newLoader);
+ MaxDecidedRX maxDecidedRX = commandStore.unsafeGetMaxDecidedRX();
+ return SummaryLoader.loader(redundantBefore, maxDecidedRX, primaryTxnId, loadKeysFor, keysOrRanges, this::newLoader);
}
- private Loader newLoader(@Nullable TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ private Loader newLoader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, @Nullable TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
{
- MaxDecidedRX maxDecidedRX = null;
- if (primaryTxnId != null && primaryTxnId.is(Txn.Kind.ExclusiveSyncPoint) && findAsDep == null)
- maxDecidedRX = commandStore.unsafeGetMaxDecidedRX();
- return new Loader(this, primaryTxnId, searchKeysOrRanges, redundantBefore, testKind, minTxnId, maxTxnId, findAsDep, maxDecidedRX);
- }
-
- private void updateTransitive(UnaryOperator<NavigableMap<TxnId, Ranges>> update)
- {
- while (true)
- {
- NavigableMap<TxnId, Ranges> prev = transitive.get();
- NavigableMap<TxnId, Ranges> next = update.apply(prev);
- if (next == null || prev == next)
- return;
- if (transitive.compareAndSet(prev, next))
- return;
- }
- }
-
- public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
- {
- updateTransitive(transitive -> {
- NavigableMap<TxnId, Ranges> next = new TreeMap<>(transitive);
- next.merge(txnId, ranges, remappingFunction);
- return next;
- });
- }
-
- public void gcBefore(TxnId gcBefore, Ranges ranges)
- {
- updateTransitive(transitive -> {
- NavigableMap<TxnId, Ranges> next = null;
- Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
- while (iterator.hasNext())
- {
- Map.Entry<TxnId, Ranges> e = iterator.next();
- Ranges newRanges = e.getValue().without(ranges);
- if (!newRanges.isEmpty())
- {
- if (next == null)
- next = new TreeMap<>();
- next.put(e.getKey(), newRanges);
- }
- }
- return next;
- });
+ return new Loader(this, redundantBefore, maxDecidedRX, primaryTxnId, searchKeysOrRanges, testKind, minTxnId, maxTxnId, findAsDep);
}
}
- public static class Loader extends Summary.Loader
+ public static class Loader extends SummaryLoader
{
private final Manager manager;
- private final MaxDecidedRX maxDecidedRX;
- private final TxnId primaryTxnId;
- @Nullable
- private final TxnId minDecidedId;
- public Loader(Manager manager, TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKinds, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep, @Nullable MaxDecidedRX maxDecidedRX)
+ public Loader(Manager manager, RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, Kinds testKinds, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
{
- super(primaryTxnId, searchKeysOrRanges, redundantBefore, testKinds, minTxnId, maxTxnId, findAsDep);
+ super(redundantBefore, maxDecidedRX, primaryTxnId, searchKeysOrRanges, testKinds, minTxnId, maxTxnId, findAsDep);
this.manager = manager;
- this.maxDecidedRX = maxDecidedRX;
- this.primaryTxnId = primaryTxnId;
- this.minDecidedId = MaxDecidedRX.minDecidedDependencyId(maxDecidedRX, searchKeysOrRanges, primaryTxnId);
}
public void intersects(Consumer<TxnId> forEach)
{
- // TODO (expected): use the ranges we find to filter results by MaxDecidedRX (don't just consume the TxnId)
switch (searchKeysOrRanges.domain())
{
case Range:
@@ -440,38 +381,11 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
for (Unseekable key : searchKeysOrRanges)
manager.searcher.search(manager.commandStore.id(), (TokenKey) key, minTxnId, maxTxnId, minDecidedId).consume(forEach);
}
-
- NavigableMap<TxnId, Ranges> transitive = manager.transitive.get();
- if (!transitive.isEmpty())
- {
- for (Map.Entry<TxnId, Ranges> e : transitive.tailMap(minTxnId, true).entrySet())
- {
- if (e.getValue().intersects(searchKeysOrRanges))
- forEach.accept(e.getKey());
- }
- }
- }
-
- boolean isRelevant(TxnIdInterval txnIdInterval)
- {
- if (maxDecidedRX == null)
- return true;
-
- if (!isMaybeRelevant(txnIdInterval.txnId))
- return false;
-
- TxnId minRelevantId = MaxDecidedRX.minDecidedDependencyId(maxDecidedRX, Ranges.of(txnIdInterval), primaryTxnId);
- return isRelevant(minRelevantId, primaryTxnId);
- }
-
- private boolean isRelevant(@Nullable TxnId minRelevantId, TxnId txnId)
- {
- return minRelevantId == null || minRelevantId.compareTo(txnId) <= 0;
}
- boolean isMaybeRelevant(TxnId txnId)
+ boolean isMaybeRelevant(TxnIdInterval txnIdInterval)
{
- return isRelevant(minDecidedId, txnId);
+ return isMaybeRelevant(txnIdInterval.txnId, null, Ranges.of(txnIdInterval));
}
public void forEachInCache(Unseekables<?> keysOrRanges, Consumer<Summary> forEach, AccordCommandStore.Caches caches)
@@ -484,10 +398,9 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
for (RoutingKey key : (AbstractUnseekableKeys)keysOrRanges)
{
IntervalBTree.accumulate(manager.cachedRangeTxnsByRange(), KEY_COMPARATORS, key, (f, s, i, c) -> {
- TxnIdInterval interval = (TxnIdInterval)i;
- if (isRelevant(interval))
+ if (isMaybeRelevant(i))
{
- TxnId txnId = ((TxnIdInterval)i).txnId;
+ TxnId txnId = i.txnId;
Summary summary = ifRelevant(c.getUnsafe(txnId));
if (summary != null)
f.accept(summary);
@@ -502,7 +415,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
for (Range range : (AbstractRanges)keysOrRanges)
{
IntervalBTree.accumulate(manager.cachedRangeTxnsByRange(), COMPARATORS, new TxnIdInterval(range.start(), range.end(), TxnId.NONE), (f, s, i, c) -> {
- if (isRelevant(i))
+ if (isMaybeRelevant(i))
{
TxnId txnId = i.txnId;
AccordCacheEntry<TxnId, Command> entry = c.getUnsafe(txnId);
@@ -540,15 +453,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
return ifRelevant(cmd);
}
- Ranges ranges = manager.transitive.get().get(txnId);
- if (ranges == null)
- return null;
-
- ranges = ranges.intersecting(searchKeysOrRanges);
- if (ranges.isEmpty())
- return null;
-
- return new Summary(txnId, txnId, NOT_DIRECTLY_WITNESSED, ranges, null, null);
+ return null;
}
public Summary ifRelevant(AccordCacheEntry<TxnId, Command> state)
@@ -581,15 +486,10 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
return ifRelevant(command);
}
- public Summary ifRelevant(Command cmd)
- {
- return ifRelevant(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
- }
-
public Summary ifRelevant(Command.Minimal cmd)
{
Invariants.require(findAsDep == null);
- return ifRelevant(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
+ return ifRelevant(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.durability, cmd.participants, null);
}
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
index 9ebeba295f..b43b08fc20 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -253,7 +253,8 @@ public class AccordAgent implements Agent
long mostRecentStart = Math.max(command.txnId().hlc(), promisedHlc);
long waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS);
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
- Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros);
+ if (mostRecentStart > nowMicros + SECONDS.toMicros(1L))
+ logger.warn("max({},{})>{}", command.txnId(), command.promised(), nowMicros);
long startTime = mostRecentStart + waitMicros;
if (startTime < nowMicros)
startTime = nowMicros + waitMicros/2;
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 9c9835e4a8..945b8d4537 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
@@ -23,9 +23,12 @@ import com.google.common.base.Throwables;
import accord.api.RoutingKey;
import accord.local.CommandStores;
import accord.local.LoadKeys;
+import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.cfk.CommandsForKey;
import accord.primitives.Ranges;
+import accord.primitives.Routable;
+import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.async.AsyncChains;
import org.apache.cassandra.distributed.Cluster;
@@ -129,7 +132,8 @@ 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())), LoadKeys.SYNC, READ_WRITE, "Test");
+ TxnId syntheticTxnId = new TxnId(TxnId.MAX_EPOCH, 0, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, new Node.Id(1));
+ PreLoadContext ctx = PreLoadContext.contextFor(syntheticTxnId, Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), LoadKeys.SYNC, READ_WRITE, "Test");
CommandStores stores = accord.node().commandStores();
for (int storeId : stores.ids())
{
diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java
index c4dc9f10fc..658fda0c4d 100644
--- a/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java
+++ b/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java
@@ -63,6 +63,13 @@ public class AccordHarrySimulationTest extends HarrySimulatorTest
return schedulers;
}
+ @Override
+ public void test() throws Exception
+ {
+ seed = "0xfa3d51da237d56e5";
+ super.test();
+ }
+
protected ConsistencyLevel validateQueryConsistency()
{
return ConsistencyLevel.QUORUM;
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 c567dd69c8..5129bff029 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
@@ -656,7 +656,6 @@ public class CommandsForKeySerializerTest
@Override public AsyncChain<Void> build(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { return null; }
@Override public <T> AsyncChain<T> build(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public void shutdown() { }
- @Override protected void registerTransitive(SafeCommandStore safeStore, RangeDeps deps){ }
@Override public <T> AsyncChain<T> build(Callable<T> task) { throw new UnsupportedOperationException(); }
@Override public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next) { throw new UnsupportedOperationException(); }
@Override public void onFailedBootstrap(int attempts, String phase, Ranges ranges, Runnable retry, Throwable failure) { throw new UnsupportedOperationException(); }
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 02/03: Improve ProgressLog: - Schedule a fallback timeout for active requests to ensure progress with lost callbacks - Clear active task for a txnId when new RunInvoker is registered - Consult DurableBefore prior to invoking recovery - Support user invoked reset of a command, clearing any active work and requeueing it - (Testing): Treat progress log for RX as recurring tasks for burn test termination Also: - Add TxnOps RECOVER, FETCH, RESET_PROGRESS_LOG
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit f02020f06dec8b44469ad9666a346d7d739bcf76
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Jul 30 19:21:09 2025 +0100
Improve ProgressLog:
- Schedule a fallback timeout for active requests to ensure progress with lost callbacks
- Clear active task for a txnId when new RunInvoker is registered
- Consult DurableBefore prior to invoking recovery
- Support user invoked reset of a command, clearing any active work and requeueing it
- (Testing): Treat progress log for RX as recurring tasks for burn test termination
Also:
- Add TxnOps RECOVER, FETCH, RESET_PROGRESS_LOG
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20815
---
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 4 +-
.../cassandra/config/DatabaseDescriptor.java | 22 +++++-
.../cassandra/db/virtual/AccordDebugKeyspace.java | 92 +++++++++++++++++++++-
.../service/accord/AccordCommandStore.java | 4 +-
5 files changed, 116 insertions(+), 8 deletions(-)
diff --git a/modules/accord b/modules/accord
index 05b2f2415c..20df7e5e5a 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit 05b2f2415c3e7c109158208824fbc804dfe43f71
+Subproject commit 20df7e5e5ad7bae442ff6fe4ee716eb7104336c1
diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java
index c34d3fab49..94ff4e946a 100644
--- a/src/java/org/apache/cassandra/config/AccordSpec.java
+++ b/src/java/org/apache/cassandra/config/AccordSpec.java
@@ -121,7 +121,9 @@ public class AccordSpec
public volatile OptionaldPositiveInt max_queued_loads = OptionaldPositiveInt.UNDEFINED;
public volatile OptionaldPositiveInt max_queued_range_loads = OptionaldPositiveInt.UNDEFINED;
- public volatile OptionaldPositiveInt max_progress_log_concurrency = OptionaldPositiveInt.UNDEFINED;
+
+ public volatile OptionaldPositiveInt progress_log_concurrency = OptionaldPositiveInt.UNDEFINED;
+ public DurationSpec.IntMillisecondsBound progress_log_query_fallback_timeout = new DurationSpec.IntMillisecondsBound("1m");
public DataStorageSpec.LongMebibytesBound cache_size = null;
public DataStorageSpec.LongMebibytesBound working_set_size = null;
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 4f4dcf720b..f38dc0e9d5 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -67,6 +67,7 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.impl.progresslog.DefaultProgressLog;
import com.googlecode.concurrenttrees.common.Iterables;
import org.apache.cassandra.audit.AuditLogOptions;
import org.apache.cassandra.auth.AllowAllInternodeAuthenticator;
@@ -189,6 +190,7 @@ public class DatabaseDescriptor
private static final int MAX_NUM_TOKENS = 1536;
private static Config conf;
+ private static DefaultProgressLog.Config accordProgressLogConfig;
/**
* Request timeouts can not be less than below defined value (see CASSANDRA-9375)
@@ -561,6 +563,8 @@ public class DatabaseDescriptor
applyGuardrails();
+ applyAccordProgressLog();
+
applyStartupChecks();
}
@@ -1297,6 +1301,20 @@ public class DatabaseDescriptor
}
}
+ private static void applyAccordProgressLog()
+ {
+ try
+ {
+ accordProgressLogConfig = new DefaultProgressLog.Config();
+ accordProgressLogConfig.concurrency = conf.accord.progress_log_concurrency.or(32);
+ accordProgressLogConfig.maxActiveRunTime = conf.accord.progress_log_query_fallback_timeout.toDuration();
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new ConfigurationException("Invalid accord progress log configuration: " + e.getMessage(), e);
+ }
+ }
+
public static StartupChecksOptions getStartupChecksOptions()
{
return startupChecksOptions;
@@ -5394,9 +5412,9 @@ public class DatabaseDescriptor
return conf.accord.max_queued_range_loads.or(Math.max(4, getAccordConcurrentOps() / 4));
}
- public static int getAccordProgressLogMaxConcurrency()
+ public static DefaultProgressLog.Config getAccordProgressLogConfig()
{
- return conf.accord.max_progress_log_concurrency.or(64);
+ return accordProgressLogConfig;
}
public static boolean getAccordCacheShrinkingOn()
diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
index ae72d01d36..fa3101741e 100644
--- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
@@ -32,9 +32,12 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
import com.google.common.collect.BoundType;
import com.google.common.collect.Range;
import com.google.common.collect.Sets;
@@ -43,6 +46,10 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.api.TraceEventType;
+import accord.coordinate.FetchData;
+import accord.coordinate.FetchRoute;
+import accord.coordinate.MaybeRecover;
+import accord.coordinate.RecoverWithRoute;
import accord.impl.CommandChange;
import accord.impl.progresslog.DefaultProgressLog;
import accord.impl.progresslog.TxnStateKind;
@@ -50,11 +57,13 @@ import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
+import accord.local.CommandStores.LatentStoreSelector;
import accord.local.Commands;
import accord.local.DurableBefore;
import accord.local.LoadKeys;
import accord.local.LoadKeysFor;
import accord.local.MaxConflicts;
+import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.RejectBefore;
import accord.local.SafeCommand;
@@ -63,14 +72,21 @@ import accord.local.StoreParticipants;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.local.durability.ShardDurability;
+import accord.primitives.FullRoute;
+import accord.primitives.Known;
import accord.primitives.Participants;
+import accord.primitives.ProgressToken;
+import accord.primitives.Route;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
+import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
+import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
@@ -104,6 +120,8 @@ import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.LocalizeString;
+import static accord.api.TraceEventType.RECOVER;
+import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
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;
@@ -1201,7 +1219,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
public static final class TxnOpsTable extends AbstractMutableVirtualTable implements AbstractVirtualTable.DataSet
{
// TODO (expected): test each of these operations
- enum Op { ERASE_VESTIGIAL, INVALIDATE, TRY_EXECUTE, FORCE_APPLY, FORCE_UPDATE }
+ enum Op { ERASE_VESTIGIAL, INVALIDATE, TRY_EXECUTE, FORCE_APPLY, FORCE_UPDATE, RECOVER, FETCH, RESET_PROGRESS_LOG }
private TxnOpsTable()
{
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_OPS,
@@ -1209,7 +1227,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
"CREATE TABLE %s (\n" +
" txn_id text,\n" +
" command_store_id int,\n" +
- " op text,\n" +
+ " op text," +
" PRIMARY KEY (txn_id, command_store_id)" +
')', UTF8Type.instance));
}
@@ -1279,6 +1297,76 @@ public class AccordDebugKeyspace extends VirtualKeyspace
return Commands.applyChain(safeStore, (Command.Executed) command);
});
break;
+ case FETCH:
+ runWithRoute(txnId, commandStoreId, command -> {
+ Timestamp executeAt = command.executeAtIfKnown();
+ return (route, result) -> fetch(txnId, executeAt, route, result);
+ });
+ break;
+ case RECOVER:
+ runWithRoute(txnId, commandStoreId, command -> (route, result) -> {
+ recover(txnId, route, result);
+ });
+ break;
+ case RESET_PROGRESS_LOG:
+ run(txnId, commandStoreId, safeStore -> {
+ ((DefaultProgressLog)safeStore.progressLog()).requeue(safeStore, TxnStateKind.Waiting, txnId);
+ ((DefaultProgressLog)safeStore.progressLog()).requeue(safeStore, TxnStateKind.Home, txnId);
+ return AsyncChains.success(null);
+ });
+ }
+ }
+
+ private void runWithRoute(TxnId txnId, int commandStoreId, Function<Command, BiConsumer<Route<?>, AsyncResult.Settable<Void>>> apply)
+ {
+ run(txnId, commandStoreId, safeStore -> {
+ SafeCommand safeCommand = safeStore.unsafeGet(txnId);
+ Command command = safeCommand.current();
+ if (command == null)
+ throw new InvalidRequestException(txnId + " not known");
+ Node node = AccordService.instance().node();
+ AsyncResult.Settable<Void> result = new AsyncResults.SettableResult<>();
+ BiConsumer<Route<?>, AsyncResult.Settable<Void>> consumer = apply.apply(command);
+ if (command.route() == null)
+ {
+ FetchRoute.fetchRoute(node, txnId, command.maxContactable(), LatentStoreSelector.standard(), (success, fail) -> {
+ if (fail != null) result.setFailure(fail);
+ else consumer.accept(success, result);
+ });
+ }
+ else
+ {
+ consumer.accept(command.route(), result);
+ }
+ return result;
+ });
+ }
+
+ private void fetch(TxnId txnId, Timestamp executeAtIfKnown, Route<?> route, AsyncResult.Settable<Void> result)
+ {
+ Node node = AccordService.instance().node();
+ FetchData.fetchSpecific(Known.Apply, node, txnId, executeAtIfKnown, route, route.withHomeKey(), LatentStoreSelector.standard(), (success, fail) -> {
+ if (fail != null) result.setFailure(fail);
+ else result.setSuccess(null);
+ });
+ }
+
+ private void recover(TxnId txnId, @Nullable Route<?> route, AsyncResult.Settable<Void> result)
+ {
+ Node node = AccordService.instance().node();
+ if (Route.isFullRoute(route))
+ {
+ RecoverWithRoute.recover(node, node.someSequentialExecutor(), txnId, NotKnownToBeInvalid, (FullRoute<?>) route, null, LatentStoreSelector.standard(), (success, fail) -> {
+ if (fail != null) result.setFailure(fail);
+ else result.setSuccess(null);
+ }, node.agent().trace(txnId, RECOVER));
+ }
+ else
+ {
+ MaybeRecover.maybeRecover(node, txnId, NotKnownToBeInvalid, route, ProgressToken.NONE, LatentStoreSelector.standard(), (success, fail) -> {
+ if (fail != null) result.setFailure(fail);
+ else result.setSuccess(null);
+ });
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index f337f8f738..972a7488e0 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -181,8 +181,8 @@ public class AccordCommandStore extends CommandStore
this.journal = journal;
this.rangeSearcher = RangeSearcher.extractRangeSearcher(journal);
this.sharedExecutor = sharedExecutor;
- if (this.progressLog() instanceof DefaultProgressLog)
- ((DefaultProgressLog)this.progressLog).setMaxConcurrency(DatabaseDescriptor.getAccordProgressLogMaxConcurrency());
+ if (this.progressLog instanceof DefaultProgressLog)
+ ((DefaultProgressLog)this.progressLog).unsafeSetConfig(DatabaseDescriptor.getAccordProgressLogConfig());
final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKey;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org