From 8de3dfc711113a694f9ffe1bebc2438585ada2f2 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Wed, 17 Jan 2024 17:08:21 +0000 Subject: [PATCH] CASSANDRA-18365: Protocol fixes --- modules/accord | 2 +- .../locator/AbstractReplicationStrategy.java | 3 - .../cassandra/metrics/AccordMetrics.java | 10 +- .../service/accord/AccordJournal.java | 76 ++++++++--- .../service/accord/AccordKeyspace.java | 11 +- .../service/accord/AccordMessageSink.java | 5 +- .../service/accord/AccordObjectSizes.java | 7 +- .../service/accord/api/AccordRoutingKey.java | 7 ++ .../accord/interop/AccordInteropCommit.java | 15 +-- .../interop/AccordInteropExecution.java | 25 ++-- .../accord/interop/AccordInteropPersist.java | 4 +- .../interop/AccordInteropReadCallback.java | 4 +- .../serializers/CheckStatusSerializers.java | 23 ++-- .../serializers/CommandSerializers.java | 30 ++--- .../accord/serializers/CommitSerializers.java | 32 ++--- .../accord/serializers/DepsSerializer.java | 1 + .../accord/serializers/FetchSerializers.java | 15 ++- .../serializers/ReadDataSerializers.java | 6 +- .../serializers/RecoverySerializers.java | 92 ++++++++++++-- .../serializers/SmallEnumSerializer.java | 118 ++++++++++++++++++ .../test/accord/AccordMetricsTest.java | 4 +- .../selection/SelectionColumnMappingTest.java | 1 - .../entities/FrozenCollectionsTest.java | 1 - .../cql3/validation/entities/JsonTest.java | 1 - .../validation/entities/UserTypesTest.java | 2 - .../operations/SelectLimitTest.java | 2 - .../CompactionAccordIteratorsTest.java | 8 +- .../service/accord/AccordCommandTest.java | 2 +- .../service/accord/AccordKeyspaceTest.java | 7 +- .../service/accord/AccordMessageSinkTest.java | 4 +- .../service/accord/AccordTestUtils.java | 15 ++- .../service/accord/CommandsForRangesTest.java | 2 +- .../accord/async/AsyncOperationTest.java | 83 ++++++++++-- .../predefined/PredefinedOperation.java | 1 - 34 files changed, 460 insertions(+), 159 deletions(-) create mode 100644 src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java diff --git a/modules/accord b/modules/accord index c524b6d3de..3789c5bfec 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c524b6d3de3923ccb6314715bd987f3b891348ab +Subproject commit 3789c5bfec50eb96157c0a55af77f78ee0cac804 diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 0870902681..04cfac933c 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -20,9 +20,6 @@ package org.apache.cassandra.locator; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; -import java.util.Collection; -import java.util.Collections; -import java.util.Map; import java.util.function.Supplier; import java.util.*; diff --git a/src/java/org/apache/cassandra/metrics/AccordMetrics.java b/src/java/org/apache/cassandra/metrics/AccordMetrics.java index 5601b9fa40..4dc053ccee 100644 --- a/src/java/org/apache/cassandra/metrics/AccordMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordMetrics.java @@ -41,7 +41,7 @@ public class AccordMetrics public final static AccordMetrics readMetrics = new AccordMetrics("ro"); public final static AccordMetrics writeMetrics = new AccordMetrics("rw"); - public static final String COMMIT_LATENCY = "CommitLatency"; + public static final String STABLE_LATENCY = "StableLatency"; public static final String EXECUTE_LATENCY = "ExecuteLatency"; public static final String APPLY_LATENCY = "ApplyLatency"; public static final String APPLY_DURATION = "ApplyDuration"; @@ -63,7 +63,7 @@ public class AccordMetrics /** * The time between start on the coordinator and commit on this replica. */ - public final Timer commitLatency; + public final Timer stableLatency; /** * The time between start on the coordinator and execution on this replica. @@ -135,7 +135,7 @@ public class AccordMetrics private AccordMetrics(String scope) { DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope); - commitLatency = Metrics.timer(replica.createMetricName(COMMIT_LATENCY)); + stableLatency = Metrics.timer(replica.createMetricName(STABLE_LATENCY)); executeLatency = Metrics.timer(replica.createMetricName(EXECUTE_LATENCY)); applyLatency = Metrics.timer(replica.createMetricName(APPLY_LATENCY)); applyDuration = Metrics.timer(replica.createMetricName(APPLY_DURATION)); @@ -204,14 +204,14 @@ public class AccordMetrics } @Override - public void onCommitted(Command cmd) + public void onStable(Command cmd) { long now = AccordService.uniqueNow(); AccordMetrics metrics = forTransaction(cmd.txnId()); if (metrics != null) { long trxTimestamp = cmd.txnId().hlc(); - metrics.commitLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS); + metrics.stableLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index ee1f79625a..28bf2c2cc5 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -110,12 +110,12 @@ import static accord.messages.MessageType.BEGIN_INVALIDATE_REQ; import static accord.messages.MessageType.BEGIN_RECOVER_REQ; import static accord.messages.MessageType.COMMIT_INVALIDATE_REQ; import static accord.messages.MessageType.COMMIT_MAXIMAL_REQ; -import static accord.messages.MessageType.COMMIT_MINIMAL_REQ; +import static accord.messages.MessageType.COMMIT_SLOW_PATH_REQ; import static accord.messages.MessageType.INFORM_DURABLE_REQ; import static accord.messages.MessageType.INFORM_OF_TXN_REQ; import static accord.messages.MessageType.PRE_ACCEPT_REQ; import static accord.messages.MessageType.PROPAGATE_APPLY_MSG; -import static accord.messages.MessageType.PROPAGATE_COMMIT_MSG; +import static accord.messages.MessageType.PROPAGATE_STABLE_MSG; import static accord.messages.MessageType.PROPAGATE_OTHER_MSG; import static accord.messages.MessageType.PROPAGATE_PRE_ACCEPT_MSG; import static accord.messages.MessageType.SET_GLOBALLY_DURABLE_REQ; @@ -124,6 +124,9 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Daemon.NON_DAEMON; import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Interrupts.SYNCHRONIZED; import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.SAFE; +import static accord.messages.MessageType.STABLE_FAST_PATH_REQ; +import static accord.messages.MessageType.STABLE_MAXIMAL_REQ; +import static accord.messages.MessageType.STABLE_SLOW_PATH_REQ; import static org.apache.cassandra.db.TypeSizes.BYTE_SIZE; import static org.apache.cassandra.db.TypeSizes.INT_SIZE; import static org.apache.cassandra.db.TypeSizes.LONG_SIZE; @@ -829,8 +832,11 @@ public class AccordJournal implements Shutdownable PRE_ACCEPT (64, PRE_ACCEPT_REQ, PreacceptSerializers.request, TXN ), ACCEPT (65, ACCEPT_REQ, AcceptSerializers.request, TXN ), ACCEPT_INVALIDATE (66, ACCEPT_INVALIDATE_REQ, AcceptSerializers.invalidate, EPOCH), - COMMIT_MINIMAL (67, COMMIT_MINIMAL_REQ, CommitSerializers.request, TXN ), + COMMIT_SLOW_PATH (67, COMMIT_SLOW_PATH_REQ, CommitSerializers.request, TXN ), COMMIT_MAXIMAL (68, COMMIT_MAXIMAL_REQ, CommitSerializers.request, TXN ), + STABLE_FAST_PATH (87, STABLE_FAST_PATH_REQ, CommitSerializers.request, TXN ), + STABLE_SLOW_PATH (88, STABLE_SLOW_PATH_REQ, CommitSerializers.request, TXN ), + STABLE_MAXIMAL (89, STABLE_MAXIMAL_REQ, CommitSerializers.request, TXN ), COMMIT_INVALIDATE (69, COMMIT_INVALIDATE_REQ, CommitSerializers.invalidate, INVL ), APPLY_MINIMAL (70, APPLY_MINIMAL_REQ, ApplySerializers.request, TXN ), APPLY_MAXIMAL (71, APPLY_MAXIMAL_REQ, ApplySerializers.request, TXN ), @@ -845,15 +851,15 @@ public class AccordJournal implements Shutdownable /* Accord local messages */ PROPAGATE_PRE_ACCEPT (79, PROPAGATE_PRE_ACCEPT_MSG, FetchSerializers.propagate, LOCAL), - PROPAGATE_COMMIT (80, PROPAGATE_COMMIT_MSG, FetchSerializers.propagate, LOCAL), + PROPAGATE_STABLE (80, PROPAGATE_STABLE_MSG, FetchSerializers.propagate, LOCAL), PROPAGATE_APPLY (81, PROPAGATE_APPLY_MSG, FetchSerializers.propagate, LOCAL), PROPAGATE_OTHER (82, PROPAGATE_OTHER_MSG, FetchSerializers.propagate, LOCAL), /* C* interop messages */ - INTEROP_COMMIT_MINIMAL (83, INTEROP_COMMIT_MINIMAL_REQ, COMMIT_MINIMAL_REQ, AccordInteropCommit.serializer, TXN), - INTEROP_COMMIT_MAXIMAL (84, INTEROP_COMMIT_MAXIMAL_REQ, COMMIT_MAXIMAL_REQ, AccordInteropCommit.serializer, TXN), - INTEROP_APPLY_MINIMAL (85, INTEROP_APPLY_MINIMAL_REQ, APPLY_MINIMAL_REQ, AccordInteropApply.serializer, TXN), - INTEROP_APPLY_MAXIMAL (86, INTEROP_APPLY_MAXIMAL_REQ, APPLY_MAXIMAL_REQ, AccordInteropApply.serializer, TXN), + INTEROP_COMMIT (83, INTEROP_COMMIT_MINIMAL_REQ, STABLE_FAST_PATH_REQ, AccordInteropCommit.serializer, TXN), + INTEROP_COMMIT_MAXIMAL (84, INTEROP_COMMIT_MAXIMAL_REQ, STABLE_MAXIMAL_REQ, AccordInteropCommit.serializer, TXN), + INTEROP_APPLY_MINIMAL (85, INTEROP_APPLY_MINIMAL_REQ, APPLY_MINIMAL_REQ, AccordInteropApply.serializer, TXN), + INTEROP_APPLY_MAXIMAL (86, INTEROP_APPLY_MAXIMAL_REQ, APPLY_MAXIMAL_REQ, AccordInteropApply.serializer, TXN), ; final int id; @@ -1377,9 +1383,9 @@ public class AccordJournal implements Shutdownable } @Override - public Commit commitMinimal() + public Commit commitSlowPath() { - return readMessage(txnId, COMMIT_MINIMAL_REQ, Commit.class); + return readMessage(txnId, COMMIT_SLOW_PATH_REQ, Commit.class); } @Override @@ -1389,9 +1395,21 @@ public class AccordJournal implements Shutdownable } @Override - public Propagate propagateCommit() + public Commit stableFastPath() { - return readMessage(txnId, PROPAGATE_COMMIT_MSG, Propagate.class); + return readMessage(txnId, STABLE_FAST_PATH_REQ, Commit.class); + } + + @Override + public Commit stableMaximal() + { + return readMessage(txnId, STABLE_MAXIMAL_REQ, Commit.class); + } + + @Override + public Propagate propagateStable() + { + return readMessage(txnId, PROPAGATE_STABLE_MSG, Propagate.class); } @Override @@ -1470,11 +1488,11 @@ public class AccordJournal implements Shutdownable } @Override - public Commit commitMinimal() + public Commit commitSlowPath() { - logger.debug("Fetching {} message for {}", COMMIT_MINIMAL_REQ, txnId); - Commit commit = provider.commitMinimal(); - logger.debug("Fetched {} message for {}: {}", COMMIT_MINIMAL_REQ, txnId, commit); + logger.debug("Fetching {} message for {}", COMMIT_SLOW_PATH_REQ, txnId); + Commit commit = provider.commitSlowPath(); + logger.debug("Fetched {} message for {}: {}", COMMIT_SLOW_PATH_REQ, txnId, commit); return commit; } @@ -1488,11 +1506,29 @@ public class AccordJournal implements Shutdownable } @Override - public Propagate propagateCommit() + public Commit stableFastPath() { - logger.debug("Fetching {} message for {}", PROPAGATE_COMMIT_MSG, txnId); - Propagate propagate = provider.propagateCommit(); - logger.debug("Fetched {} message for {}: {}", PROPAGATE_COMMIT_MSG, txnId, propagate); + logger.debug("Fetching {} message for {}", STABLE_FAST_PATH_REQ, txnId); + Commit commit = provider.stableFastPath(); + logger.debug("Fetched {} message for {}: {}", STABLE_FAST_PATH_REQ, txnId, commit); + return commit; + } + + @Override + public Commit stableMaximal() + { + logger.debug("Fetching {} message for {}", STABLE_MAXIMAL_REQ, txnId); + Commit commit = provider.stableMaximal(); + logger.debug("Fetched {} message for {}: {}", STABLE_MAXIMAL_REQ, txnId, commit); + return commit; + } + + @Override + public Propagate propagateStable() + { + logger.debug("Fetching {} message for {}", PROPAGATE_STABLE_MSG, txnId); + Propagate propagate = provider.propagateStable(); + logger.debug("Fetched {} message for {}: {}", PROPAGATE_STABLE_MSG, txnId, propagate); return propagate; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index fc695f9263..ddcaed5a0d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -821,11 +821,11 @@ public class AccordKeyspace addEnumCellIfModified(CommandsColumns.status, Command::saveStatus, builder, timestampMicros, nowInSeconds, original, command); addCellIfModified(CommandsColumns.execute_at, Command::executeAt, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command); addCellIfModified(CommandsColumns.promised_ballot, Command::promised, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command); - addCellIfModified(CommandsColumns.accepted_ballot, Command::accepted, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command); + addCellIfModified(CommandsColumns.accepted_ballot, Command::acceptedOrCommitted, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command); // TODO review this is just to work around Truncated not being committed but having a status after committed // so status claims it is committed. - if (!command.isTruncated() && command.isCommitted()) + if (command.isStable() && !command.isTruncated()) { Command.Committed committed = command.asCommitted(); Command.Committed originalCommitted = original != null && original.isCommitted() ? original.asCommitted() : null; @@ -1284,8 +1284,11 @@ public class AccordKeyspace return (deps) -> { - if (bytes == null || !bytes.hasRemaining()) - return deps == null ? WaitingOn.EMPTY : WaitingOn.none(deps); + if (bytes == null) + return null; + + if (!bytes.hasRemaining()) + return WaitingOn.none(deps); try { diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java index 64f4ced5da..5efdb7c0f4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java +++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java @@ -124,8 +124,11 @@ public class AccordMessageSink implements MessageSink builder.put(MessageType.ACCEPT_INVALIDATE_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ); builder.put(MessageType.GET_DEPS_REQ, Verb.ACCORD_GET_DEPS_REQ); builder.put(MessageType.GET_DEPS_RSP, Verb.ACCORD_GET_DEPS_RSP); - builder.put(MessageType.COMMIT_MINIMAL_REQ, Verb.ACCORD_COMMIT_REQ); + builder.put(MessageType.COMMIT_SLOW_PATH_REQ, Verb.ACCORD_COMMIT_REQ); builder.put(MessageType.COMMIT_MAXIMAL_REQ, Verb.ACCORD_COMMIT_REQ); + builder.put(MessageType.STABLE_FAST_PATH_REQ, Verb.ACCORD_COMMIT_REQ); + builder.put(MessageType.STABLE_SLOW_PATH_REQ, Verb.ACCORD_COMMIT_REQ); + builder.put(MessageType.STABLE_MAXIMAL_REQ, Verb.ACCORD_COMMIT_REQ); builder.put(MessageType.COMMIT_INVALIDATE_REQ, Verb.ACCORD_COMMIT_INVALIDATE_REQ); builder.put(MessageType.APPLY_MINIMAL_REQ, Verb.ACCORD_APPLY_REQ); builder.put(MessageType.APPLY_MAXIMAL_REQ, Verb.ACCORD_APPLY_REQ); diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index 095781f46a..3f506302db 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -296,7 +296,7 @@ public class AccordObjectSizes final static long NOT_DEFINED = measure(Command.SerializerSupport.notDefined(attrs(false, false), Ballot.ZERO)); final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true), EMPTY_TXNID, null));; final static long ACCEPTED = measure(Command.SerializerSupport.accepted(attrs(true, false), SaveStatus.Accepted, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO)); - final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.EMPTY)); + final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null)); final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.EMPTY, EMPTY_WRITES, EMPTY_RESULT)); final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null)); final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, null)); @@ -314,6 +314,7 @@ public class AccordObjectSizes case PreCommitted: return ACCEPTED; case Committed: + case Stable: case ReadyToExecute: return COMMITTED; case PreApplied: @@ -346,13 +347,13 @@ public class AccordObjectSizes size += sizeNullable(command.executeAt(), AccordObjectSizes::timestamp); size += sizeNullable(command.partialTxn(), AccordObjectSizes::txn); size += sizeNullable(command.partialDeps(), AccordObjectSizes::dependencies); - size += sizeNullable(command.accepted(), AccordObjectSizes::timestamp); + size += sizeNullable(command.acceptedOrCommitted(), AccordObjectSizes::timestamp); size += sizeNullable(command.writes(), AccordObjectSizes::writes); if (command.result() instanceof TxnResult) size += sizeNullable(command.result(), AccordObjectSizes::results); - if (!(command instanceof Command.Committed)) + if (!(command instanceof Command.Committed && command.saveStatus().hasBeen(Status.Stable))) return size; Command.Committed committed = command.asCommitted(); diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java index 5b089267a1..acf4da1923 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java @@ -31,6 +31,7 @@ import accord.api.Key; import accord.api.RoutingKey; import accord.local.ShardDistributor; import accord.primitives.Range; +import accord.primitives.RangeFactory; import accord.primitives.Ranges; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.dht.IPartitioner; @@ -63,6 +64,12 @@ public abstract class AccordRoutingKey extends AccordRoutableKey implements Rout public abstract long estimatedSizeOnHeap(); public abstract AccordRoutingKey withTable(TableId table); + @Override + public RangeFactory rangeFactory() + { + return (s, e) -> new TokenRange((AccordRoutingKey) s, (AccordRoutingKey) e); + } + public SentinelKey asSentinelKey() { return (SentinelKey) this; diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java index e3051bf644..e92edb1ec9 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java @@ -25,6 +25,7 @@ import accord.local.Node; import accord.messages.Commit; import accord.messages.MessageType; import accord.messages.ReadData; +import accord.primitives.Ballot; import accord.primitives.Deps; import accord.primitives.FullRoute; import accord.primitives.PartialDeps; @@ -44,20 +45,20 @@ public class AccordInteropCommit extends Commit public static final IVersionedSerializer serializer = new CommitSerializer(AccordInteropRead.class, AccordInteropRead.requestSerializer) { @Override - protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Kind kind, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read) + protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read) { - return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, executeAt, partialTxn, partialDeps, fullRoute, read); + return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, ballot, executeAt, partialTxn, partialDeps, fullRoute, read); } }; - public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute scope, long waitForEpoch, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nonnull ReadData readData) + public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nonnull ReadData readData) { - super(kind, txnId, scope, waitForEpoch, executeAt, partialTxn, partialDeps, fullRoute, readData); + super(kind, txnId, scope, waitForEpoch, ballot, executeAt, partialTxn, partialDeps, fullRoute, readData); } public AccordInteropCommit(Kind kind, Node.Id to, Topology coordinateTopology, Topologies topologies, TxnId txnId, Txn txn, FullRoute route, Timestamp executeAt, Deps deps, AccordInteropRead read) { - super(kind, to, coordinateTopology, topologies, txnId, txn, route, executeAt, deps, (t, u, p) -> read); + super(kind, to, coordinateTopology, topologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, read); } @Override @@ -65,8 +66,8 @@ public class AccordInteropCommit extends Commit { switch (kind) { - case Minimal: return AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ; - case Maximal: return AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ; + case StableFastPath: return AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ; + case StableWithTxnAndDeps: return AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ; default: throw new IllegalStateException(); } } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index 9aa9179b38..489fdfba27 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -28,6 +28,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; +import accord.messages.ReadTxnData; +import accord.primitives.Ballot; import org.apache.cassandra.schema.TableId; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,7 +39,7 @@ import accord.api.Data; import accord.api.Result; import accord.coordinate.Execute; import accord.coordinate.Persist; -import accord.coordinate.TxnExecute; +import accord.coordinate.ExecuteTxn; import accord.local.AgentExecutor; import accord.local.CommandStore; import accord.local.Node; @@ -153,13 +155,13 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal } @Override - public Execute create(Node node, TxnId txnId, Txn txn, FullRoute route, Participants readScope, Timestamp executeAt, Deps deps, BiConsumer callback) + public Execute create(Node node, Topologies topologies, Path path, TxnId txnId, Txn txn, FullRoute route, Participants readScope, Timestamp executeAt, Deps deps, BiConsumer callback) { // Unrecoverable repair always needs to be run by AccordInteropExecution AccordUpdate.Kind updateKind = AccordUpdate.kind(txn.update()); ConsistencyLevel consistencyLevel = txn.read() instanceof TxnRead ? ((TxnRead) txn.read()).cassandraConsistencyLevel() : null; if (updateKind != AccordUpdate.Kind.UNRECOVERABLE_REPAIR && (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ONE || txn.read().keys().isEmpty())) - return TxnExecute.FACTORY.create(node, txnId, txn, route, readScope, executeAt, deps, callback); + return ExecuteTxn.FACTORY.create(node, topologies, path, txnId, txn, route, readScope, executeAt, deps, callback); return new AccordInteropExecution(node, txnId, txn, updateKind, route, readScope, executeAt, deps, callback, executor, consistencyLevel, endpointMapper); } } @@ -252,7 +254,8 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal Node.Id id = endpointMapper.mappedId(to); SinglePartitionReadCommand command = (SinglePartitionReadCommand) message.payload; AccordInteropRead read = new AccordInteropRead(id, executes, txnId, readScope, executeAt, command); - AccordInteropCommit commit = new AccordInteropCommit(Commit.Kind.Minimal, id, coordinateTopology, allTopologies, + // TODO (required): understand interop and whether StableFastPath is appropriate + AccordInteropCommit commit = new AccordInteropCommit(Kind.StableFastPath, id, coordinateTopology, allTopologies, txnId, txn, route, executeAt, deps, read); node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this)); } @@ -337,14 +340,14 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal for (int i = 0; i < digestRequests.size(); i++) contacted.add(digestRequests.endpoint(i)); if (readsCurrentlyUnderConstruction.decrementAndGet() == 0) - sendCommitsToUncontacted(); + sendStableToUncontacted(); } - private void sendCommitsToUncontacted() + private void sendStableToUncontacted() { for (Node.Id to : executeTopology.nodes()) if (!contacted.contains(endpointMapper.mappedEndpoint(to))) - node.send(to, new Commit(Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false)); + node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); } @Override @@ -355,7 +358,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal for (Node.Id to : allTopologies.nodes()) { if (!executeTopology.contains(to)) - node.send(to, new Commit(Commit.Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false)); + node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); } } AsyncChain result; @@ -367,7 +370,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal CommandStore cs = node.commandStores().select(route.homeKey()); result.beginAsResult().withExecutor(cs).begin((data, failure) -> { if (failure == null) - Persist.persist(node, executes, txnId, route, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback); + Persist.persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback); else callback.accept(null, failure); }); @@ -380,7 +383,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal // TODO (expected): We should send the read in the same message as the commit. This requires refactor ReadData.Kind so that it doesn't specify the ordinal encoding // and can be extended similar to MessageType which allows additional types not from Accord to be added for (Node.Id to : executeTopology.nodes()) - node.send(to, new Commit(Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false)); + node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); repairUpdate.runBRR(AccordInteropExecution.this); return new TxnData(); }); @@ -408,6 +411,6 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal @Override public void sendMaximalCommit(Id to) { - Commit.commitMaximal(node, to, txn, txnId, executeAt, route, deps, readScope); + Commit.stableMaximal(node, to, txn, txnId, executeAt, route, deps); } } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java index 7ef1581536..5144546959 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java @@ -23,7 +23,7 @@ import java.util.function.BiConsumer; import accord.api.Result; import accord.api.Update; import accord.coordinate.Persist; -import accord.coordinate.TxnPersist; +import accord.coordinate.PersistTxn; import accord.coordinate.tracking.AppliedTracker; import accord.coordinate.tracking.QuorumTracker; import accord.coordinate.tracking.RequestStatus; @@ -57,7 +57,7 @@ public class AccordInteropPersist extends Persist Update update = txn.update(); ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null; if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty()) - return TxnPersist.FACTORY.create(node, topologies, txnId, route, txn, executeAt, deps, writes, result); + return PersistTxn.FACTORY.create(node, topologies, txnId, route, txn, executeAt, deps, writes, result); return new AccordInteropPersist(node, topologies, txnId, route, txn, executeAt, deps, writes, result, consistencyLevel); } }; diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java index 6bf006b3a8..36dcedd09e 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java @@ -30,7 +30,7 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.RequestCallback; -import static accord.messages.ReadData.ReadNack.NotCommitted; +import static accord.messages.ReadData.CommitOrReadNack.Insufficient; public abstract class AccordInteropReadCallback implements Callback { @@ -63,7 +63,7 @@ public abstract class AccordInteropReadCallback implements Callback nullableWrites = NullableSerializer.wrap(writes); - public static final EnumSerializer route = new EnumSerializer<>(Status.KnownRoute.class); - public static final EnumSerializer definition = new EnumSerializer<>(Status.Definition.class); - public static final EnumSerializer knownExecuteAt = new EnumSerializer<>(Status.KnownExecuteAt.class); - public static final EnumSerializer knownDeps = new EnumSerializer<>(Status.KnownDeps.class); - public static final EnumSerializer outcome = new EnumSerializer<>(Status.Outcome.class); - public static final EnumSerializer invalidIfNot = new EnumSerializer<>(Infer.InvalidIfNot.class); - public static final EnumSerializer isPreempted = new EnumSerializer<>(Infer.IsPreempted.class); + public static final SmallEnumSerializer knownRoute = new SmallEnumSerializer<>(Status.KnownRoute.class); + public static final SmallEnumSerializer definition = new SmallEnumSerializer<>(Status.Definition.class); + public static final SmallEnumSerializer knownExecuteAt = new SmallEnumSerializer<>(Status.KnownExecuteAt.class); + public static final SmallEnumSerializer knownDeps = new SmallEnumSerializer<>(Status.KnownDeps.class); + public static final NullableSmallEnumSerializer nullableKnownDeps = new NullableSmallEnumSerializer<>(knownDeps); + public static final SmallEnumSerializer outcome = new SmallEnumSerializer<>(Status.Outcome.class); + public static final SmallEnumSerializer invalidIfNot = new SmallEnumSerializer<>(Infer.InvalidIfNot.class); + public static final SmallEnumSerializer isPreempted = new SmallEnumSerializer<>(Infer.IsPreempted.class); public static final IVersionedSerializer known = new IVersionedSerializer<>() { @Override public void serialize(Known known, DataOutputPlus out, int version) throws IOException { - route.serialize(known.route, out, version); + knownRoute.serialize(known.route, out, version); definition.serialize(known.definition, out, version); knownExecuteAt.serialize(known.executeAt, out, version); knownDeps.serialize(known.deps, out, version); @@ -264,7 +266,7 @@ public class CommandSerializers @Override public Known deserialize(DataInputPlus in, int version) throws IOException { - return new Known(route.deserialize(in, version), + return new Known(knownRoute.deserialize(in, version), definition.deserialize(in, version), knownExecuteAt.deserialize(in, version), knownDeps.deserialize(in, version), @@ -274,11 +276,11 @@ public class CommandSerializers @Override public long serializedSize(Known known, int version) { - return route.serializedSize(known.route, version) - + definition.serializedSize(known.definition, version) - + knownExecuteAt.serializedSize(known.executeAt, version) - + knownDeps.serializedSize(known.deps, version) - + outcome.serializedSize(known.outcome, version); + return knownRoute.serializedSize(known.route, version) + + definition.serializedSize(known.definition, version) + + knownExecuteAt.serializedSize(known.executeAt, version) + + knownDeps.serializedSize(known.deps, version) + + outcome.serializedSize(known.outcome, version); } }; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java index 23ea20b197..cd704d3db1 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java @@ -23,6 +23,7 @@ import javax.annotation.Nullable; import accord.messages.Commit; import accord.messages.ReadData; +import accord.primitives.Ballot; import accord.primitives.FullRoute; import accord.primitives.PartialDeps; import accord.primitives.PartialRoute; @@ -30,7 +31,6 @@ import accord.primitives.PartialTxn; import accord.primitives.Timestamp; import accord.primitives.TxnId; import accord.primitives.Unseekables; -import accord.utils.Invariants; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -43,25 +43,7 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi public class CommitSerializers { - private static final IVersionedSerializer kind = new IVersionedSerializer() - { - public void serialize(Commit.Kind kind, DataOutputPlus out, int version) throws IOException - { - Invariants.checkArgument(kind == Commit.Kind.Minimal || kind == Commit.Kind.Maximal); - out.writeBoolean(kind == Commit.Kind.Maximal); - - } - - public Commit.Kind deserialize(DataInputPlus in, int version) throws IOException - { - return in.readBoolean() ? Commit.Kind.Maximal : Commit.Kind.Minimal; - } - - public long serializedSize(Commit.Kind kind, int version) - { - return TypeSizes.BOOL_SIZE; - } - }; + private static final IVersionedSerializer kind = new EnumSerializer<>(Commit.Kind.class); public abstract static class CommitSerializer extends TxnRequestSerializer { @@ -76,6 +58,7 @@ public class CommitSerializers public void serializeBody(C msg, DataOutputPlus out, int version) throws IOException { kind.serialize(msg.kind, out, version); + CommandSerializers.ballot.serialize(msg.ballot, out, version); CommandSerializers.timestamp.serialize(msg.executeAt, out, version); CommandSerializers.nullablePartialTxn.serialize(msg.partialTxn, out, version); DepsSerializer.partialDeps.serialize(msg.partialDeps, out, version); @@ -83,7 +66,8 @@ public class CommitSerializers serializeNullable(msg.readData, out, version, read); } - protected abstract C deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Commit.Kind kind, Timestamp executeAt, + protected abstract C deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Commit.Kind kind, + Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read); @@ -92,6 +76,7 @@ public class CommitSerializers { return deserializeCommit(txnId, scope, waitForEpoch, kind.deserialize(in, version), + CommandSerializers.ballot.deserialize(in, version), CommandSerializers.timestamp.deserialize(in, version), CommandSerializers.nullablePartialTxn.deserialize(in, version), DepsSerializer.partialDeps.deserialize(in, version), @@ -104,6 +89,7 @@ public class CommitSerializers public long serializedBodySize(C msg, int version) { return kind.serializedSize(msg.kind, version) + + CommandSerializers.ballot.serializedSize(msg.ballot, version) + CommandSerializers.timestamp.serializedSize(msg.executeAt, version) + CommandSerializers.nullablePartialTxn.serializedSize(msg.partialTxn, version) + DepsSerializer.partialDeps.serializedSize(msg.partialDeps, version) @@ -115,9 +101,9 @@ public class CommitSerializers public static final IVersionedSerializer request = new CommitSerializer(ReadData.class, ReadDataSerializers.readData) { @Override - protected Commit deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Commit.Kind kind, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read) + protected Commit deserializeCommit(TxnId txnId, PartialRoute scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read) { - return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, executeAt, partialTxn, partialDeps, fullRoute, read); + return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, ballot, executeAt, partialTxn, partialDeps, fullRoute, read); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java index 3530e06936..9498bef0f2 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java @@ -51,6 +51,7 @@ public abstract class DepsSerializer implements IVersionedSerial return new Deps(keyDeps, rangeDeps); } }; + public static final IVersionedSerializer nullableDeps = NullableSerializer.wrap(deps); public static final DepsSerializer partialDeps = new DepsSerializer() { diff --git a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java index 61d715b798..61e60f3dbd 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java @@ -32,6 +32,7 @@ import accord.messages.CheckStatus; import accord.messages.Propagate; import accord.messages.ReadData; import accord.messages.ReadData.ReadReply; +import accord.primitives.Ballot; import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.Ranges; @@ -94,7 +95,7 @@ public class FetchSerializers public static final IVersionedSerializer reply = new IVersionedSerializer() { - final ReadData.ReadNack[] nacks = ReadData.ReadNack.values(); + final ReadData.CommitOrReadNack[] nacks = ReadData.CommitOrReadNack.values(); final IVersionedSerializer streamDataSerializer = new CastingSerializer<>(StreamData.class, StreamData.serializer); @Override @@ -102,7 +103,7 @@ public class FetchSerializers { if (!reply.isOk()) { - out.writeByte(1 + ((ReadData.ReadNack) reply).ordinal()); + out.writeByte(1 + ((ReadData.CommitOrReadNack) reply).ordinal()); return; } @@ -148,14 +149,15 @@ public class FetchSerializers KeySerializers.route.serialize(p.route, out, version); CommandSerializers.saveStatus.serialize(p.maxKnowledgeSaveStatus, out, version); CommandSerializers.saveStatus.serialize(p.maxSaveStatus, out, version); + CommandSerializers.ballot.serialize(p.ballot, out, version); CommandSerializers.durability.serialize(p.durability, out, version); KeySerializers.nullableRoutingKey.serialize(p.homeKey, out, version); KeySerializers.nullableRoutingKey.serialize(p.progressKey, out, version); CommandSerializers.known.serialize(p.achieved, out, version); CheckStatusSerializers.foundKnownMap.serialize(p.known, out, version); - out.writeBoolean(p.isTruncated); + out.writeBoolean(p.isShardTruncated); CommandSerializers.nullablePartialTxn.serialize(p.partialTxn, out, version); - DepsSerializer.nullablePartialDeps.serialize(p.committedDeps, out, version); + DepsSerializer.nullablePartialDeps.serialize(p.stableDeps, out, version); out.writeLong(p.toEpoch); CommandSerializers.nullableTimestamp.serialize(p.committedExecuteAt, out, version); CommandSerializers.nullableWrites.serialize(p.writes, out, version); @@ -168,6 +170,7 @@ public class FetchSerializers Route route = KeySerializers.route.deserialize(in, version); SaveStatus maxKnowledgeSaveStatus = CommandSerializers.saveStatus.deserialize(in, version); SaveStatus maxSaveStatus = CommandSerializers.saveStatus.deserialize(in, version); + Ballot ballot = CommandSerializers.ballot.deserialize(in, version); Durability durability = CommandSerializers.durability.deserialize(in, version); RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version); RoutingKey progressKey = KeySerializers.nullableRoutingKey.deserialize(in, version); @@ -197,6 +200,7 @@ public class FetchSerializers route, maxKnowledgeSaveStatus, maxSaveStatus, + ballot, durability, homeKey, progressKey, @@ -218,6 +222,7 @@ public class FetchSerializers + KeySerializers.route.serializedSize(p.route, version) + CommandSerializers.saveStatus.serializedSize(p.maxKnowledgeSaveStatus, version) + CommandSerializers.saveStatus.serializedSize(p.maxSaveStatus, version) + + CommandSerializers.ballot.serializedSize(p.ballot, version) + CommandSerializers.durability.serializedSize(p.durability, version) + KeySerializers.nullableRoutingKey.serializedSize(p.homeKey, version) + KeySerializers.nullableRoutingKey.serializedSize(p.progressKey, version) @@ -225,7 +230,7 @@ public class FetchSerializers + CheckStatusSerializers.foundKnownMap.serializedSize(p.known, version) + TypeSizes.BOOL_SIZE + CommandSerializers.nullablePartialTxn.serializedSize(p.partialTxn, version) - + DepsSerializer.nullablePartialDeps.serializedSize(p.committedDeps, version) + + DepsSerializer.nullablePartialDeps.serializedSize(p.stableDeps, version) + TypeSizes.sizeof(p.toEpoch) + CommandSerializers.nullableTimestamp.serializedSize(p.committedExecuteAt, version) + CommandSerializers.nullableWrites.serializedSize(p.writes, version) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java index 5afc451bb8..cfae34db4f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java @@ -23,7 +23,7 @@ import java.io.IOException; import accord.api.Data; import accord.messages.ApplyThenWaitUntilApplied; import accord.messages.ReadData; -import accord.messages.ReadData.ReadNack; +import accord.messages.ReadData.CommitOrReadNack; import accord.messages.ReadData.ReadOk; import accord.messages.ReadData.ReadReply; import accord.messages.ReadData.ReadType; @@ -172,7 +172,7 @@ public class ReadDataSerializers public static final class ReplySerializer implements IVersionedSerializer { // TODO (now): use something other than ordinal - final ReadNack[] nacks = ReadNack.values(); + final CommitOrReadNack[] nacks = CommitOrReadNack.values(); private final IVersionedSerializer dataSerializer; public ReplySerializer(IVersionedSerializer dataSerializer) @@ -185,7 +185,7 @@ public class ReadDataSerializers { if (!reply.isOk()) { - out.writeByte(1 + ((ReadNack) reply).ordinal()); + out.writeByte(1 + ((CommitOrReadNack) reply).ordinal()); return; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index adf60212cc..346d1c8bdf 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -23,6 +23,7 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; import accord.api.Result; +import accord.api.RoutingKey; import accord.local.Status; import accord.messages.BeginRecovery; import accord.messages.BeginRecovery.RecoverNack; @@ -31,7 +32,7 @@ import accord.messages.BeginRecovery.RecoverReply; import accord.primitives.Ballot; import accord.primitives.Deps; import accord.primitives.FullRoute; -import accord.primitives.PartialDeps; +import accord.primitives.LatestDeps; import accord.primitives.PartialRoute; import accord.primitives.PartialTxn; import accord.primitives.Timestamp; @@ -89,8 +90,7 @@ public class RecoverySerializers CommandSerializers.status.serialize(recoverOk.status, out, version); CommandSerializers.ballot.serialize(recoverOk.accepted, out, version); CommandSerializers.nullableTimestamp.serialize(recoverOk.executeAt, out, version); - DepsSerializer.partialDeps.serialize(recoverOk.deps, out, version); - DepsSerializer.nullablePartialDeps.serialize(recoverOk.acceptedDeps, out, version); + latestDeps.serialize(recoverOk.deps, out, version); DepsSerializer.deps.serialize(recoverOk.earlierCommittedWitness, out, version); DepsSerializer.deps.serialize(recoverOk.earlierAcceptedNoWitness, out, version); out.writeBoolean(recoverOk.rejectsFastPath); @@ -112,9 +112,9 @@ public class RecoverySerializers return new RecoverNack(supersededBy); } - RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull PartialDeps deps, PartialDeps acceptedDeps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) + RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) { - return new RecoverOk(txnId, status, accepted, executeAt, deps, acceptedDeps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result); + return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result); } @Override @@ -135,8 +135,7 @@ public class RecoverySerializers status, CommandSerializers.ballot.deserialize(in, version), CommandSerializers.nullableTimestamp.deserialize(in, version), - DepsSerializer.partialDeps.deserialize(in, version), - DepsSerializer.nullablePartialDeps.deserialize(in, version), + latestDeps.deserialize(in, version), DepsSerializer.deps.deserialize(in, version), DepsSerializer.deps.deserialize(in, version), in.readBoolean(), @@ -157,8 +156,7 @@ public class RecoverySerializers size += CommandSerializers.status.serializedSize(recoverOk.status, version); size += CommandSerializers.ballot.serializedSize(recoverOk.accepted, version); size += CommandSerializers.nullableTimestamp.serializedSize(recoverOk.executeAt, version); - size += DepsSerializer.partialDeps.serializedSize(recoverOk.deps, version); - size += DepsSerializer.nullablePartialDeps.serializedSize(recoverOk.acceptedDeps, version); + size += latestDeps.serializedSize(recoverOk.deps, version); size += DepsSerializer.deps.serializedSize(recoverOk.earlierCommittedWitness, version); size += DepsSerializer.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version); size += TypeSizes.sizeof(recoverOk.rejectsFastPath); @@ -173,4 +171,80 @@ public class RecoverySerializers + (reply.isOk() ? serializedOkSize((RecoverOk) reply, version) : serializedNackSize((RecoverNack) reply, version)); } }; + + public static final IVersionedSerializer latestDeps = new IVersionedSerializer() + { + @Override + public void serialize(LatestDeps t, DataOutputPlus out, int version) throws IOException + { + out.writeUnsignedVInt32(t.size()); + for (int i = 0 ; i < t.size() ; ++i) + { + RoutingKey start = t.startAt(i); + KeySerializers.routingKey.serialize(start, out, version); + LatestDeps.LatestEntry e = t.valueAt(i); + if (e == null) + { + CommandSerializers.nullableKnownDeps.serialize(null, out, version); + } + else + { + CommandSerializers.nullableKnownDeps.serialize(e.known, out, version); + CommandSerializers.ballot.serialize(e.ballot, out, version); + DepsSerializer.nullableDeps.serialize(e.coordinatedDeps, out, version); + DepsSerializer.nullableDeps.serialize(e.localDeps, out, version); + } + } + KeySerializers.routingKey.serialize(t.startAt(t.size()), out, version); + } + + @Override + public LatestDeps deserialize(DataInputPlus in, int version) throws IOException + { + int size = in.readUnsignedVInt32(); + RoutingKey[] starts = new RoutingKey[size + 1]; + LatestDeps.LatestEntry[] values = new LatestDeps.LatestEntry[size]; + for (int i = 0 ; i < size ; ++i) + { + starts[i] = KeySerializers.routingKey.deserialize(in, version); + Status.KnownDeps knownDeps = CommandSerializers.nullableKnownDeps.deserialize(in, version); + if (knownDeps == null) + continue; + + Ballot ballot = CommandSerializers.ballot.deserialize(in, version); + Deps coordinatedDeps = DepsSerializer.nullableDeps.deserialize(in, version); + Deps localDeps = DepsSerializer.nullableDeps.deserialize(in, version); + values[i] = new LatestDeps.LatestEntry(knownDeps, ballot, coordinatedDeps, localDeps); + } + starts[size] = KeySerializers.routingKey.deserialize(in, version); + + return LatestDeps.SerializerSupport.create(true, starts, values); + } + + @Override + public long serializedSize(LatestDeps t, int version) + { + long size = 0; + size += TypeSizes.sizeofUnsignedVInt(t.size()); + for (int i = 0 ; i < t.size() ; ++i) + { + RoutingKey start = t.startAt(i); + size += KeySerializers.routingKey.serializedSize(start, version); + LatestDeps.LatestEntry e = t.valueAt(i); + if (e == null) + { + size += CommandSerializers.nullableKnownDeps.serializedSize(null, version); + } + else + { + size += CommandSerializers.nullableKnownDeps.serializedSize(e.known, version); + size += CommandSerializers.ballot.serializedSize(e.ballot, version); + size += DepsSerializer.nullableDeps.serializedSize(e.coordinatedDeps, version); + size += DepsSerializer.nullableDeps.serializedSize(e.localDeps, version); + } + } + size += KeySerializers.routingKey.serializedSize(t.startAt(t.size()), version); + return size; + } + }; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java new file mode 100644 index 0000000000..2182d359a2 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.accord.serializers; + +import java.io.IOException; +import java.nio.ByteBuffer; + +import javax.annotation.Nullable; + +import accord.messages.SimpleReply; +import accord.utils.Invariants; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; + +public class SmallEnumSerializer> implements IVersionedSerializer +{ + public static final SmallEnumSerializer simpleReply = new SmallEnumSerializer<>(SimpleReply.class); + + // TODO: should use something other than ordinal for ser/deser + final E[] values; + + public SmallEnumSerializer(Class clazz) + { + this.values = clazz.getEnumConstants(); + Invariants.checkArgument(values.length < 255); // allow an extra 1 for nullable variant to ensure consistency + } + + public E forOrdinal(int ordinal) + { + return values[ordinal]; + } + + @Override + public void serialize(E t, DataOutputPlus out, int version) throws IOException + { + out.write(t.ordinal()); + } + + @Override + public E deserialize(DataInputPlus in, int version) throws IOException + { + return values[in.readByte()]; + } + + public ByteBuffer serialize(E e) + { + ByteBuffer out = ByteBuffer.allocate(1); + out.put((byte)e.ordinal()); + out.flip(); + return out; + } + + @Override + public long serializedSize(E t, int version) + { + return 1; + } + + public static class NullableSmallEnumSerializer> implements IVersionedSerializer + { + // TODO: should use something other than ordinal for ser/deser + final E[] values; + + public NullableSmallEnumSerializer(SmallEnumSerializer wrap) + { + this.values = wrap.values; + } + + public E forOrdinal(int ordinal) + { + return values[ordinal]; + } + + @Override + public void serialize(@Nullable E t, DataOutputPlus out, int version) throws IOException + { + out.write(t == null ? 0 : 1 + t.ordinal()); + } + + @Override + public E deserialize(DataInputPlus in, int version) throws IOException + { + int ordinal = in.readByte(); + return ordinal == 0 ? null : values[ordinal - 1]; + } + + public ByteBuffer serialize(E e) + { + ByteBuffer out = ByteBuffer.allocate(1); + out.put((byte)(e == null ? 0 : (1 + e.ordinal()))); + out.flip(); + return out; + } + + @Override + public long serializedSize(E t, int version) + { + return 1; + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java index 028bcfbf32..487fbf5f1d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java @@ -231,12 +231,12 @@ public class AccordMetricsTest extends AccordTestBase } } - private void assertReplicaMetrics(int node, String scope, long commits, long executions, long applications) + private void assertReplicaMetrics(int node, String scope, long stable, long executions, long applications) { DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_REPLICA, scope); Map metrics = diff(countingMetrics0).get(node); Function metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName()); - assertThat(metric.apply(AccordMetrics.COMMIT_LATENCY)).isEqualTo(commits); + assertThat(metric.apply(AccordMetrics.STABLE_LATENCY)).isEqualTo(stable); assertThat(metric.apply(AccordMetrics.EXECUTE_LATENCY)).isEqualTo(executions); assertThat(metric.apply(AccordMetrics.APPLY_LATENCY)).isEqualTo(applications); assertThat(metric.apply(AccordMetrics.APPLY_DURATION)).isEqualTo(applications); diff --git a/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java b/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java index 9f20aea4c0..58f4a3364e 100644 --- a/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java +++ b/test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java @@ -40,7 +40,6 @@ import org.apache.cassandra.service.QueryState; import org.apache.cassandra.utils.ByteBufferUtil; import static java.util.Arrays.asList; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java index 1bdd1ca19b..2d935eb978 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java @@ -42,7 +42,6 @@ import org.apache.cassandra.exceptions.SyntaxException; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; import static org.junit.Assert.assertEquals; public class FrozenCollectionsTest extends CQLTester diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java index d05391afcc..2976c014bf 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java @@ -39,7 +39,6 @@ import java.text.SimpleDateFormat; import java.util.*; import java.util.concurrent.*; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java index 530d55ba11..2d30f1946c 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java @@ -27,8 +27,6 @@ import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.service.StorageService; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; - public class UserTypesTest extends CQLTester { @BeforeClass diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java index 3cbc9d79c0..31ef09ec86 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java @@ -28,8 +28,6 @@ import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.service.StorageService; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; - public class SelectLimitTest extends CQLTester { // This method will be ran instead of the CQLTester#setUpClass diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index 85ef473ac4..7765b9c1ad 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -83,8 +83,6 @@ import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.AccordCommandStore; import org.apache.cassandra.service.accord.AccordKeyspace; -import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows; -import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns; import org.apache.cassandra.service.accord.AccordTestUtils; import org.apache.cassandra.service.accord.IAccordService; import org.apache.cassandra.utils.FBUtilities; @@ -461,7 +459,7 @@ public class CompactionAccordIteratorsTest TxnId[] txnIds = additionalCommand ? TXN_IDS : new TxnId[] {TXN_ID}; for (TxnId txnId : txnIds) { - Txn txn = txnId.rw().isWrite() ? AccordTestUtils.createWriteTxn(42) : AccordTestUtils.createTxn(42); + Txn txn = txnId.kind().isWrite() ? AccordTestUtils.createWriteTxn(42) : AccordTestUtils.createTxn(42); Seekable key = txn.keys().get(0); PartialDeps partialDeps = Deps.NONE.slice(AccordTestUtils.fullRange(txn)); PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true); @@ -482,9 +480,9 @@ public class CompactionAccordIteratorsTest flush(commandStore); getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> { Commit commit = - Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Minimal, txnId, partialTxn, partialDeps, route, null); + Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableFastPath, Ballot.ZERO, txnId, partialTxn, partialDeps, route, null); commandStore.appendToJournal(commit); - CheckedCommands.commit(safe, txnId, route, null, partialTxn, txnId, partialDeps); + CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, txnId, partialDeps); }).beginAsResult()); flush(commandStore); getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> { diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index 0bd628903a..06a55d24cb 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java @@ -157,7 +157,7 @@ public class AccordCommandTest })); // check commit - Commit commit = Commit.SerializerSupport.create(txnId, route, 1, Commit.Kind.Maximal, executeAt, partialTxn, deps, fullRoute, null); + Commit commit = Commit.SerializerSupport.create(txnId, route, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute, null); commandStore.appendToJournal(commit); getUninterruptibly(commandStore.execute(commit, commit::apply)); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java index 52aff302e3..0be7d69254 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java @@ -82,18 +82,19 @@ public class AccordKeyspaceTest extends CQLTester.InMemory common.route(route); common.partialDeps(deps.slice(scope)); common.durability(Status.Durability.NotDurable); - Command.WaitingOn waitingOn = Command.WaitingOn.none(deps.slice(scope)); + Command.WaitingOn waitingOn = null; Command.Committed committed = Command.SerializerSupport.committed(common, SaveStatus.Committed, id, Ballot.ZERO, Ballot.ZERO, waitingOn); AccordSafeCommand safeCommand = new AccordSafeCommand(AccordTestUtils.loaded(id, null)); safeCommand.set(committed); - Commit commit = Commit.SerializerSupport.create(id, route.slice(scope), 1, Commit.Kind.Maximal, id, partialTxn, partialDeps, route, null); + Commit commit = Commit.SerializerSupport.create(id, route.slice(scope), 1, Commit.Kind.StableFastPath, Ballot.ZERO, id, partialTxn, partialDeps, route, null); store.appendToJournal(commit); Mutation mutation = AccordKeyspace.getCommandMutation(store, safeCommand, 42); mutation.apply(); - Assertions.assertThat(AccordKeyspace.loadCommand(store, id)).isEqualTo(committed); + Command loaded = AccordKeyspace.loadCommand(store, id); + Assertions.assertThat(loaded).isEqualTo(committed); } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java index 93150d1054..82f56f8690 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java @@ -92,7 +92,7 @@ public class AccordMessageSinkTest checkRequestReplies(request, new AbstractFetchCoordinator.FetchResponse(null, null, id), - ReadData.ReadNack.NotCommitted); + ReadData.CommitOrReadNack.Insufficient); } @@ -103,7 +103,7 @@ public class AccordMessageSinkTest Request request = new ReadTxnData(node, topologies, txnId, topology.ranges(), txnId); checkRequestReplies(request, new ReadData.ReadOk(null, null), - ReadData.ReadNack.NotCommitted); + ReadData.CommitOrReadNack.Insufficient); } private static void checkRequestReplies(Request request, Reply... replies) diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index c27878f401..77911183de 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -128,6 +128,19 @@ public class AccordTestUtils executeAt, Ballot.ZERO, Ballot.ZERO, + null); + } + + public static Command stable(TxnId txnId, PartialTxn txn, Timestamp executeAt) + { + CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId).partialDeps(PartialDeps.NONE); + attrs.partialTxn(txn); + attrs.route(route(txn)); + return Command.SerializerSupport.committed(attrs, + SaveStatus.Stable, + executeAt, + Ballot.ZERO, + Ballot.ZERO, Command.WaitingOn.EMPTY); } @@ -189,7 +202,7 @@ public class AccordTestUtils @Override public void preaccepted(Command command, ProgressShard progressShard) {} @Override public void accepted(Command command, ProgressShard progressShard) {} @Override public void precommitted(Command command) {} - @Override public void committed(Command command, ProgressShard progressShard) {} + @Override public void stable(Command command, ProgressShard progressShard) {} @Override public void readyToExecute(Command command) {} @Override public void executed(Command command, ProgressShard progressShard) {} @Override public void clear(TxnId txnId) {} diff --git a/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java b/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java index e3c3ba4346..982ad8ca8f 100644 --- a/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java +++ b/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java @@ -84,7 +84,7 @@ public class CommandsForRangesTest assertThat(cfr.knownIds()).containsExactly(max); assertThat(cfr.maxRedundant()).isEqualTo(knownIds.size() == 1 ? null : knownIds.get(knownIds.size() - 2)); - cfr.prune(new TxnId(max.logicalNext(max.node), max.rw(), max.domain()), FULL_RANGE); + cfr.prune(new TxnId(max.logicalNext(max.node), max.kind(), max.domain()), FULL_RANGE); assertThat(cfr.knownIds()).isEmpty(); assertThat(cfr.maxRedundant()).isEqualTo(max); }); diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java index 66f796e4a7..ca695433f9 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java @@ -42,6 +42,7 @@ import accord.local.Command; import accord.local.PreLoadContext; import accord.local.SafeCommand; import accord.local.SafeCommandStore; +import accord.local.SaveStatus; import accord.messages.Accept; import accord.messages.Commit; import accord.messages.PreAccept; @@ -162,9 +163,9 @@ public class AsyncOperationTest } } - private static Command createCommittedAndPersist(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt) + private static Command createStableAndPersist(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt) { - Command command = AccordTestUtils.Commands.committed(txnId, createPartialTxn(0), executeAt); + Command command = AccordTestUtils.Commands.stable(txnId, createPartialTxn(0), executeAt); AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null)); safeCommand.set(command); @@ -173,7 +174,8 @@ public class AsyncOperationTest Commit.SerializerSupport.create(txnId, command.route().slice(AccordTestUtils.fullRange(command.partialTxn().keys())), txnId.epoch(), - Commit.Kind.Maximal, + Commit.Kind.StableWithTxnAndDeps, + Ballot.ZERO, executeAt, command.partialTxn(), command.partialDeps(), @@ -184,17 +186,64 @@ public class AsyncOperationTest return command; } - private static Command createCommittedAndPersist(AccordCommandStore commandStore, TxnId txnId) + private static Command createStableAndPersist(AccordCommandStore commandStore, TxnId txnId) { - return createCommittedAndPersist(commandStore, txnId, txnId); + return createStableAndPersist(commandStore, txnId, txnId); } - private static Command createCommittedUsingLifeCycle(AccordCommandStore commandStore, TxnId txnId) + private static Command createStableUsingFastLifeCycle(AccordCommandStore commandStore, TxnId txnId) { - return createCommittedUsingLifeCycle(commandStore, txnId, txnId); + return createStableUsingFastLifeCycle(commandStore, txnId, txnId); } - private static Command createCommittedUsingLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt) + private static Command createStableUsingFastLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt) + { + PartialTxn partialTxn = createPartialTxn(0); + RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable(); + FullRoute route = partialTxn.keys().toRoute(routingKey); + Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys()); + PartialRoute partialRoute = route.slice(ranges); + PartialDeps deps = PartialDeps.builder(ranges).build(); + + // create and write messages to the journal for loading to succeed + PreAccept preAccept = + PreAccept.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), txnId.epoch(), false, txnId.epoch(), partialTxn, route); + Commit stable = + Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableFastPath, Ballot.ZERO, executeAt, partialTxn, deps, route, null); + + commandStore.appendToJournal(preAccept); + commandStore.appendToJournal(stable); + + try + { + Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys()), safe -> { + CheckedCommands.preaccept(safe, txnId, partialTxn, route, null); + CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps); + return safe.ifInitialised(txnId).current(); + }).beginAsResult()); + + // clear cache + commandStore.executeBlocking(() -> { + long cacheSize = commandStore.capacity(); + commandStore.setCapacity(0); + commandStore.setCapacity(cacheSize); + commandStore.cache().awaitSaveResults(); + }); + + return command; + } + catch (ExecutionException e) + { + throw new AssertionError(e); + } + } + + private static Command createStableUsingSlowLifeCycle(AccordCommandStore commandStore, TxnId txnId) + { + return createStableUsingSlowLifeCycle(commandStore, txnId, txnId); + } + + private static Command createStableUsingSlowLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt) { PartialTxn partialTxn = createPartialTxn(0); RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable(); @@ -209,18 +258,22 @@ public class AsyncOperationTest Accept accept = Accept.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), txnId.epoch(), false, Ballot.ZERO, executeAt, partialTxn.keys(), deps); Commit commit = - Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Minimal, executeAt, partialTxn, deps, route, null); + Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Commit, Ballot.ZERO, executeAt, partialTxn, deps, route, null); + Commit stable = + Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableSlowPath, Ballot.ZERO, executeAt, partialTxn, deps, route, null); commandStore.appendToJournal(preAccept); commandStore.appendToJournal(accept); commandStore.appendToJournal(commit); + commandStore.appendToJournal(stable); try { Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys()), safe -> { CheckedCommands.preaccept(safe, txnId, partialTxn, route, null); CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps); - CheckedCommands.commit(safe, txnId, route, null, partialTxn, executeAt, deps); + CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps); + CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps); return safe.ifInitialised(txnId).current(); }).beginAsResult()); @@ -265,7 +318,7 @@ public class AsyncOperationTest TxnId txnId = txnId(1, clock.incrementAndGet(), 1); - createCommittedAndPersist(commandStore, txnId); + createStableAndPersist(commandStore, txnId); Consumer consumer = safeStore -> safeStore.ifInitialised(txnId).readyToExecute(); PreLoadContext ctx = contextFor(txnId); @@ -396,8 +449,12 @@ public class AsyncOperationTest private static void createCommand(AccordCommandStore commandStore, RandomSource rs, List ids) { // to simulate CommandsForKey not being found, use createCommittedAndPersist periodically as it does not update - if (rs.nextBoolean()) ids.forEach(id -> createCommittedAndPersist(commandStore, id)); - else ids.forEach(id -> createCommittedUsingLifeCycle(commandStore, id)); + switch (rs.nextInt(3)) + { + case 0: ids.forEach(id -> createStableAndPersist(commandStore, id)); break; + case 1: ids.forEach(id -> createStableUsingFastLifeCycle(commandStore, id)); break; + case 2: ids.forEach(id -> createStableUsingSlowLifeCycle(commandStore, id)); + } commandStore.unsafeClearCache(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java index bf969ad46b..630ee3aca5 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java @@ -28,7 +28,6 @@ import org.apache.cassandra.stress.generate.*; import org.apache.cassandra.stress.operations.PartitionOperation; import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; -import org.apache.cassandra.stress.settings.CqlVersion; import org.apache.cassandra.stress.settings.StressSettings; public abstract class PredefinedOperation extends PartitionOperation