From 3e16efa49083c397d4c64f3f6c18e9f09928c3e6 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Wed, 8 Jan 2025 09:58:12 +0000 Subject: [PATCH] Protocol optimisations: - Privileged coordinator. If the coordinator is a replica we can reduce our quorum sizes by including the coordinator's vote. - with deps: if we include coordinator's preaccept deps we can reliably reduce quorum size by 1, at the expense of recovery sometimes requiring additional phases and waiting for future txns - with only vote: if we only include the vote we can avoid any additional recovery phases or waiting for future txns, but can reduce our quorum size for only some configurations - Medium path. If t=t0 at a simple majority we can take just two rounds. - with additional phase: on recovery, earlier txns must wait for medium path to be disabled (or committed) so we do not accidentally recover a transaction that won't be witnessed - with unstable deps: on slow path commit, deps not found in accept as committed as unstable and do not affect recovery decisions for earlier transactions Also improve: - Recovery of await conditions simply recalculates the rejects flag rather than restarting to ensure faster forward progress - refactor Command hierarchy - tweak: don't save Writes for non-write transactions Also fix: - isOutOfRange when invoked from callback for some txn < Committed - handle loadingPruned that is pre-bootstrap on update to bootstrappedAt - journal replay can overwrite in memory state loaded for a command not yet replayed - subtle ordering bug in LatestDeps.mergeProposal - fix occasional class initialisation order bug - handle race condition on learning of topology patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-20222 --- modules/accord | 2 +- .../db/compaction/CompactionIterator.java | 2 +- src/java/org/apache/cassandra/net/Verb.java | 7 +- .../service/accord/AccordJournal.java | 4 +- .../service/accord/AccordMessageSink.java | 29 ++- .../service/accord/AccordObjectSizes.java | 79 +++++--- .../service/accord/AccordService.java | 9 +- .../service/accord/AccordTopology.java | 17 +- .../service/accord/CommandsForRanges.java | 4 +- .../service/accord/api/AccordAgent.java | 9 + .../ParameterizedFastPathStrategy.java | 8 +- .../fastpath/SimpleFastPathStrategy.java | 2 +- .../accord/interop/AccordInteropApply.java | 15 +- .../accord/interop/AccordInteropCommit.java | 74 ------- .../interop/AccordInteropExecution.java | 14 +- .../accord/interop/AccordInteropPersist.java | 20 +- .../accord/interop/AccordInteropRead.java | 20 +- .../interop/AccordInteropReadCallback.java | 3 +- .../interop/AccordInteropReadRepair.java | 16 +- .../interop/AccordInteropStableThenRead.java | 181 ++++++++++++++++++ .../accord/serializers/AcceptSerializers.java | 34 ++-- .../accord/serializers/ApplySerializers.java | 9 +- .../accord/serializers/AwaitSerializer.java | 8 +- .../serializers/CheckStatusSerializers.java | 2 +- .../serializers/CommandSerializers.java | 18 +- .../accord/serializers/CommitSerializers.java | 46 +---- .../accord/serializers/FetchSerializers.java | 4 +- .../serializers/PreacceptSerializers.java | 36 ++-- .../serializers/ReadDataSerializers.java | 90 +++++++-- .../serializers/RecoverySerializers.java | 10 +- .../serializers/TopologySerializers.java | 19 +- .../tcm/transformations/AccordMarkStale.java | 2 +- .../accord/AccordInteroperabilityTest.java | 2 +- .../test/accord/AccordSimpleFastPathTest.java | 2 +- .../test/AccordHarrySimulationTest.java | 2 +- .../db/virtual/AccordDebugKeyspaceTest.java | 9 + .../accord/AccordCommandStoreTest.java | 12 +- .../service/accord/AccordCommandTest.java | 17 +- .../accord/AccordFastPathCoordinatorTest.java | 12 +- .../accord/AccordJournalOrderTest.java | 6 +- .../service/accord/AccordKeyspaceTest.java | 11 +- .../service/accord/AccordTestUtils.java | 49 ++--- .../service/accord/AccordTopologyTest.java | 12 +- .../accord/SimulatedAccordCommandStore.java | 13 +- .../SimulatedAccordCommandStoreTestBase.java | 2 +- .../accord/SimulatedAccordTaskTest.java | 2 +- .../CommandsForKeySerializerTest.java | 93 ++++++--- .../cassandra/utils/AccordGenerators.java | 82 +++++--- 48 files changed, 695 insertions(+), 424 deletions(-) delete mode 100644 src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java create mode 100644 src/java/org/apache/cassandra/service/accord/interop/AccordInteropStableThenRead.java diff --git a/modules/accord b/modules/accord index c80ec466f5..f9ed591f1b 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c80ec466f5dd633d4fcbaedc5674f183446def34 +Subproject commit f9ed591f1b91115351b871640b6318f11dd208af diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index c9c84f4efa..9c7df6d19f 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -1025,7 +1025,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte RedundantBefore redundantBefore = redundantBefores.get(key.commandStoreId); DurableBefore durableBefore = durableBefores.get(key.commandStoreId); - Cleanup cleanup = commandBuilder.shouldCleanup(agent, redundantBefore, durableBefore); + Cleanup cleanup = commandBuilder.shouldCleanup(agent, redundantBefore, durableBefore, true); if (cleanup == ERASE) return PartitionUpdate.fullPartitionDelete(metadata(), partition.partitionKey(), maxSeenTimestamp, nowInSec).unfilteredIterator(); diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index ec8a37b956..7f86490f26 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -84,9 +84,9 @@ import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification; import org.apache.cassandra.service.accord.FetchTopology; import org.apache.cassandra.service.accord.FetchMinEpoch; import org.apache.cassandra.service.accord.interop.AccordInteropApply; -import org.apache.cassandra.service.accord.interop.AccordInteropCommit; import org.apache.cassandra.service.accord.interop.AccordInteropRead; import org.apache.cassandra.service.accord.interop.AccordInteropReadRepair; +import org.apache.cassandra.service.accord.interop.AccordInteropStableThenRead; import org.apache.cassandra.service.accord.serializers.AcceptSerializers; import org.apache.cassandra.service.accord.serializers.ApplySerializers; import org.apache.cassandra.service.accord.serializers.BeginInvalidationSerializers; @@ -316,7 +316,7 @@ public enum Verb ACCORD_PRE_ACCEPT_REQ (121, P2, writeTimeout, IMMEDIATE, () -> PreacceptSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_PRE_ACCEPT_RSP ), ACCORD_ACCEPT_RSP (122, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.reply, AccordService::responseHandlerOrNoop ), ACCORD_ACCEPT_REQ (123, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ), - ACCORD_ACCEPT_INVALIDATE_REQ (124, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.invalidate, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ), + ACCORD_NOT_ACCEPT_REQ (124, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.notAccept, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ), ACCORD_READ_RSP (125, P2, readTimeout, IMMEDIATE, () -> ReadDataSerializers.reply, AccordService::responseHandlerOrNoop ), ACCORD_READ_REQ (126, P2, readTimeout, IMMEDIATE, () -> ReadDataSerializers.readData, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ), ACCORD_COMMIT_REQ (127, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ), @@ -331,6 +331,7 @@ public enum Verb ACCORD_AWAIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.request, AccordService::requestHandlerOrNoop, ACCORD_AWAIT_RSP ), ACCORD_AWAIT_ASYNC_RSP_REQ (137, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.asyncReply, AccordService::requestHandlerOrNoop ), ACCORD_WAIT_UNTIL_APPLIED_REQ (138, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ), + ACCORD_STABLE_THEN_READ_REQ (139, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.stableThenRead, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ), ACCORD_INFORM_DURABLE_REQ (140, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ), ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.reply, AccordService::responseHandlerOrNoop ), ACCORD_CHECK_STATUS_REQ (142, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ), @@ -356,7 +357,7 @@ public enum Verb ACCORD_INTEROP_READ_RSP (155, P2, writeTimeout, IMMEDIATE, () -> AccordInteropRead.replySerializer, AccordService::responseHandlerOrNoop), ACCORD_INTEROP_READ_REQ (156, P2, writeTimeout, IMMEDIATE, () -> AccordInteropRead.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP), - ACCORD_INTEROP_COMMIT_REQ (157, P2, writeTimeout, IMMEDIATE, () -> AccordInteropCommit.serializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP), + ACCORD_INTEROP_STABLE_THEN_READ_REQ(157, P2, writeTimeout, IMMEDIATE, () -> AccordInteropStableThenRead.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP), ACCORD_INTEROP_READ_REPAIR_RSP (158, P2, writeTimeout, IMMEDIATE, () -> AccordInteropReadRepair.replySerializer, AccordService::responseHandlerOrNoop), ACCORD_INTEROP_READ_REPAIR_REQ (159, P2, writeTimeout, IMMEDIATE, () -> AccordInteropReadRepair.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_REPAIR_RSP), ACCORD_INTEROP_APPLY_REQ (160, P2, writeTimeout, IMMEDIATE, () -> AccordInteropApply.serializer, AccordService::requestHandlerOrNoop, ACCORD_APPLY_RSP), diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index b3a05ed011..d171a4c021 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -209,7 +209,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore) { Builder builder = load(commandStoreId, txnId); - Cleanup cleanup = builder.shouldCleanup(agent, redundantBefore, durableBefore); + Cleanup cleanup = builder.shouldCleanup(agent, redundantBefore, durableBefore, false); switch (cleanup) { case EXPUNGE_PARTIAL: @@ -227,7 +227,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable if (builder.isEmpty()) return null; - Cleanup cleanup = builder.shouldCleanup(node.agent(), redundantBefore, durableBefore); + Cleanup cleanup = builder.shouldCleanup(node.agent(), redundantBefore, durableBefore, false); switch (cleanup) { case EXPUNGE_PARTIAL: diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java index afa2cfcc6c..897d6ac3b7 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java +++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java @@ -39,7 +39,6 @@ import accord.impl.RequestCallbacks; import accord.local.AgentExecutor; import accord.local.Node; import accord.messages.Callback; -import accord.messages.Commit; import accord.messages.MessageType; import accord.messages.ReadData; import accord.messages.Reply; @@ -72,14 +71,13 @@ public class AccordMessageSink implements MessageSink public static final class AccordMessageType extends MessageType { - public static final AccordMessageType INTEROP_READ_REQ = remote("INTEROP_READ_REQ", false); - public static final AccordMessageType INTEROP_READ_RSP = remote("INTEROP_READ_RSP", false); - public static final AccordMessageType INTEROP_READ_REPAIR_REQ = remote("INTEROP_READ_REPAIR_REQ", false); - public static final AccordMessageType INTEROP_READ_REPAIR_RSP = remote("INTEROP_READ_REPAIR_RSP", false); - public static final AccordMessageType INTEROP_COMMIT_MINIMAL_REQ = remote("INTEROP_COMMIT_MINIMAL_REQ", true ); - public static final AccordMessageType INTEROP_COMMIT_MAXIMAL_REQ = remote("INTEROP_COMMIT_MAXIMAL_REQ", true ); - public static final AccordMessageType INTEROP_APPLY_MINIMAL_REQ = remote("INTEROP_APPLY_MINIMAL_REQ", true ); - public static final AccordMessageType INTEROP_APPLY_MAXIMAL_REQ = remote("INTEROP_APPLY_MAXIMAL_REQ", true ); + public static final AccordMessageType INTEROP_READ_REQ = remote("INTEROP_READ_REQ", false); + public static final AccordMessageType INTEROP_READ_RSP = remote("INTEROP_READ_RSP", false); + public static final AccordMessageType INTEROP_STABLE_THEN_READ_REQ = remote("INTEROP_STABLE_THEN_READ_REQ", false); + public static final AccordMessageType INTEROP_READ_REPAIR_REQ = remote("INTEROP_READ_REPAIR_REQ", false); + public static final AccordMessageType INTEROP_READ_REPAIR_RSP = remote("INTEROP_READ_REPAIR_RSP", false); + public static final AccordMessageType INTEROP_APPLY_MINIMAL_REQ = remote("INTEROP_APPLY_MINIMAL_REQ", true ); + public static final AccordMessageType INTEROP_APPLY_MAXIMAL_REQ = remote("INTEROP_APPLY_MAXIMAL_REQ", true ); public static final List values; @@ -122,7 +120,7 @@ public class AccordMessageSink implements MessageSink private final Map> overrideReplyVerbs = ImmutableMap.>builder() // read takes Result | Nack .put(Verb.ACCORD_FETCH_DATA_REQ, EnumSet.of(Verb.ACCORD_FETCH_DATA_RSP, Verb.ACCORD_READ_RSP /* nack */)) - .put(Verb.ACCORD_INTEROP_COMMIT_REQ, EnumSet.of(Verb.ACCORD_INTEROP_READ_RSP, Verb.ACCORD_READ_RSP)) + .put(Verb.ACCORD_INTEROP_STABLE_THEN_READ_REQ, EnumSet.of(Verb.ACCORD_INTEROP_READ_RSP, Verb.ACCORD_READ_RSP)) .put(Verb.ACCORD_INTEROP_READ_REPAIR_REQ, EnumSet.of(Verb.ACCORD_INTEROP_READ_REPAIR_RSP, Verb.ACCORD_READ_RSP)) .build(); @@ -134,7 +132,7 @@ public class AccordMessageSink implements MessageSink builder.put(MessageType.PRE_ACCEPT_RSP, Verb.ACCORD_PRE_ACCEPT_RSP); builder.put(MessageType.ACCEPT_REQ, Verb.ACCORD_ACCEPT_REQ); builder.put(MessageType.ACCEPT_RSP, Verb.ACCORD_ACCEPT_RSP); - builder.put(MessageType.ACCEPT_INVALIDATE_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ); + builder.put(MessageType.NOT_ACCEPT_REQ, Verb.ACCORD_NOT_ACCEPT_REQ); builder.put(MessageType.CALCULATE_DEPS_REQ, Verb.ACCORD_CALCULATE_DEPS_REQ); builder.put(MessageType.CALCULATE_DEPS_RSP, Verb.ACCORD_CALCULATE_DEPS_RSP); builder.put(MessageType.GET_LATEST_DEPS_REQ, Verb.ACCORD_GET_LATEST_DEPS_REQ); @@ -151,6 +149,7 @@ public class AccordMessageSink implements MessageSink builder.put(MessageType.APPLY_MAXIMAL_REQ, Verb.ACCORD_APPLY_REQ); builder.put(MessageType.APPLY_RSP, Verb.ACCORD_APPLY_RSP); builder.put(MessageType.READ_REQ, Verb.ACCORD_READ_REQ); + builder.put(MessageType.STABLE_THEN_READ_REQ, Verb.ACCORD_STABLE_THEN_READ_REQ); builder.put(MessageType.READ_EPHEMERAL_REQ, Verb.ACCORD_READ_REQ); builder.put(MessageType.READ_RSP, Verb.ACCORD_READ_RSP); builder.put(MessageType.BEGIN_RECOVER_REQ, Verb.ACCORD_BEGIN_RECOVER_REQ); @@ -173,10 +172,9 @@ public class AccordMessageSink implements MessageSink builder.put(MessageType.QUERY_DURABLE_BEFORE_RSP, Verb.ACCORD_QUERY_DURABLE_BEFORE_RSP); builder.put(AccordMessageType.INTEROP_READ_REQ, Verb.ACCORD_INTEROP_READ_REQ); builder.put(AccordMessageType.INTEROP_READ_RSP, Verb.ACCORD_INTEROP_READ_RSP); + builder.put(AccordMessageType.INTEROP_STABLE_THEN_READ_REQ, Verb.ACCORD_INTEROP_STABLE_THEN_READ_REQ); builder.put(AccordMessageType.INTEROP_READ_REPAIR_REQ, Verb.ACCORD_INTEROP_READ_REPAIR_REQ); builder.put(AccordMessageType.INTEROP_READ_REPAIR_RSP, Verb.ACCORD_INTEROP_READ_REPAIR_RSP); - builder.put(AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ, Verb.ACCORD_INTEROP_COMMIT_REQ); - builder.put(AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ, Verb.ACCORD_INTEROP_COMMIT_REQ); builder.put(AccordMessageType.INTEROP_APPLY_MINIMAL_REQ, Verb.ACCORD_INTEROP_APPLY_REQ); builder.put(AccordMessageType.INTEROP_APPLY_MAXIMAL_REQ, Verb.ACCORD_INTEROP_APPLY_REQ); mapping = builder.build(); @@ -285,11 +283,8 @@ public class AccordMessageSink implements MessageSink switch (verb) { - case ACCORD_COMMIT_REQ: - if (((Commit)request).readData == null) - break; - case ACCORD_READ_REQ: + case ACCORD_STABLE_THEN_READ_REQ: if (slowRead == null || isRangeBarrier(request)) break; diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index da811ce83a..53059a0075 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -26,7 +26,7 @@ import accord.api.Result; import accord.api.RoutingKey; import accord.local.Command; import accord.local.Command.WaitingOn; -import accord.local.CommonAttributes; +import accord.local.ICommand; import accord.local.Node; import accord.local.StoreParticipants; import accord.local.cfk.CommandsForKey; @@ -61,9 +61,9 @@ import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.accord.api.AccordRoutingKey; import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey; import org.apache.cassandra.service.accord.api.PartitionKey; -import org.apache.cassandra.service.accord.serializers.ResultSerializers; import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer; import org.apache.cassandra.service.accord.txn.AccordUpdate; +import org.apache.cassandra.service.accord.txn.TxnData; import org.apache.cassandra.service.accord.txn.TxnQuery; import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.service.accord.txn.TxnResult; @@ -271,53 +271,86 @@ public class AccordObjectSizes private static class CommandEmptySizes { private final static TokenKey EMPTY_KEY = new TokenKey(EMPTY_ID, null); - private final static TxnId EMPTY_TXNID = new TxnId(42, 42, Kind.Read, Domain.Key, new Node.Id(42)); + private final static TxnId EMPTY_TXNID = new TxnId(42, 42, 0, Kind.Read, Domain.Key, new Node.Id(42)); - private static CommonAttributes attrs(boolean hasDeps, boolean hasTxn, boolean executes) + private static ICommand attrs(boolean hasDeps, boolean hasTxn, boolean executes) { FullKeyRoute route = new FullKeyRoute(EMPTY_KEY, new RoutingKey[]{ EMPTY_KEY }); - CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(EMPTY_TXNID) - .setParticipants(StoreParticipants.empty(EMPTY_TXNID, route, !executes)); - attrs.durability(Status.Durability.NotDurable); + ICommand.Builder builder = new ICommand.Builder(EMPTY_TXNID) + .setParticipants(StoreParticipants.empty(EMPTY_TXNID, route, !executes)) + .durability(Status.Durability.NotDurable) + .executeAt(EMPTY_TXNID) + .promised(Ballot.ZERO); if (hasDeps) - attrs.partialDeps(PartialDeps.NONE); + builder.partialDeps(PartialDeps.NONE); if (hasTxn) - attrs.partialTxn(new PartialTxn.InMemory(null, null, null, null, null)); + builder.partialTxn(new PartialTxn.InMemory(null, null, null, null, null)); - return attrs; + if (executes) + { + builder.waitingOn(WaitingOn.empty(Domain.Key)); + builder.result(new TxnData()); + } + + return builder; } - private static final Writes EMPTY_WRITES = new Writes(EMPTY_TXNID, EMPTY_TXNID, Keys.EMPTY, (key, safeStore, txnId, executeAt, store, txn) -> null); - private static final Result EMPTY_RESULT = new Result() {}; - - final static long NOT_DEFINED = measure(Command.SerializerSupport.notDefined(attrs(false, false, false), Ballot.ZERO)); - final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true, false), EMPTY_TXNID, Ballot.ZERO));; - final static long ACCEPTED = measure(Command.SerializerSupport.accepted(attrs(true, false, false), SaveStatus.Accepted, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO)); - final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true, false), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null)); - final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.empty(Domain.Key), EMPTY_WRITES, ResultSerializers.APPLIED)); - final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null)); - final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID))); + final static long NOT_DEFINED = measure(Command.NotDefined.notDefined(attrs(false, false, false))); + final static long PREACCEPTED = measure(Command.PreAccepted.preaccepted(attrs(false, true, false), SaveStatus.PreAccepted)); + final static long NOTACCEPTED = measure(Command.NotAcceptedWithoutDefinition.notAccepted(attrs(false, false, false), SaveStatus.NotAccepted)); + final static long ACCEPTED = measure(Command.Accepted.accepted(attrs(true, false, false), SaveStatus.AcceptedMedium)); + final static long COMMITTED = measure(Command.Committed.committed(attrs(true, true, false), SaveStatus.Committed)); + final static long EXECUTED = measure(Command.Executed.executed(attrs(true, true, true), SaveStatus.Applied)); + final static long TRUNCATED = measure(Command.Truncated.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null)); + final static long INVALIDATED = measure(Command.Truncated.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID))); private static long emptySize(Command command) { - switch (command.status()) + switch (command.saveStatus()) { + case Uninitialised: case NotDefined: return NOT_DEFINED; case PreAccepted: + case PreAcceptedWithDeps: + case PreAcceptedWithVote: return PREACCEPTED; + case NotAccepted: + case PreNotAccepted: case AcceptedInvalidate: - case Accepted: + return NOTACCEPTED; + case NotAcceptedWithDefinition: + case NotAcceptedWithDefAndVote: + case NotAcceptedWithDefAndDeps: + case PreNotAcceptedWithDefinition: + case PreNotAcceptedWithDefAndVote: + case PreNotAcceptedWithDefAndDeps: + case AcceptedInvalidateWithDefinition: + case AcceptedMedium: + case AcceptedMediumWithDefinition: + case AcceptedSlow: + case AcceptedSlowWithDefinition: case PreCommitted: + case PreCommittedWithDeps: + case PreCommittedWithFixedDeps: + case PreCommittedWithDefinition: + case PreCommittedWithDefAndDeps: + case PreCommittedWithDefAndFixedDeps: return ACCEPTED; case Committed: + case ReadyToExecute: case Stable: return COMMITTED; case PreApplied: + case Applying: case Applied: return EXECUTED; - case Truncated: + case TruncatedApply: + case TruncatedApplyWithDeps: + case TruncatedApplyWithOutcome: + case ErasedOrVestigial: + case Erased: return TRUNCATED; case Invalidated: return INVALIDATED; diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index af668547d6..cb1acc1a81 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -1077,7 +1077,7 @@ public class AccordService implements IAccordService, Shutdownable private static CommandStoreTxnBlockedGraph.TxnState populate(CommandStoreTxnBlockedGraph.Builder state, Command cmd) { CommandStoreTxnBlockedGraph.Builder.TxnBuilder cmdTxnState = state.txn(cmd.txnId(), cmd.executeAt(), cmd.saveStatus()); - if (!cmd.hasBeen(Status.Applied) && cmd.isCommitted()) + if (!cmd.hasBeen(Status.Applied) && cmd.hasBeen(Status.Stable)) { // check blocking state Command.WaitingOn waitingOn = cmd.asCommitted().waitingOn(); @@ -1419,9 +1419,6 @@ public class AccordService implements IAccordService, Shutdownable case Redundant: tryFailure(new ExecuteSyncPoint.SyncPointErased()); return; - case Invalid: - tryFailure(new Invalidated(exclusiveSyncPoint.syncId, exclusiveSyncPoint.route.homeKey())); - return; } } else @@ -1461,9 +1458,9 @@ public class AccordService implements IAccordService, Shutdownable } @Override - public void onCallbackFailure(Node.Id from, Throwable failure) + public boolean onCallbackFailure(Node.Id from, Throwable failure) { - tryFailure(failure); + return tryFailure(failure); } } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopology.java b/src/java/org/apache/cassandra/service/accord/AccordTopology.java index fc4e582ea9..6763d6090d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTopology.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTopology.java @@ -66,17 +66,18 @@ public class AccordTopology private static class ShardLookup extends HashMap { - private Shard createOrReuse(boolean pendingRemoval, accord.primitives.Range range, SortedArrayList nodes, Set fastPathElectorate, Set joining) + private Shard createOrReuse(boolean pendingRemoval, accord.primitives.Range range, SortedArrayList nodes, SortedArrayList fastPath, Set joining) { Shard prev = get(range); if (prev != null && prev.pendingRemoval == pendingRemoval - && Objects.equals(prev.nodes, nodes) - && Objects.equals(prev.fastPathElectorate, fastPathElectorate) - && Objects.equals(prev.joining, joining)) + && prev.nodes.equals(nodes) + && prev.fastPathElectorateSize == fastPath.size() + && prev.nodes.without(prev.notInFastPath).equals(fastPath) + && joining.size() == prev.joining.size() && prev.joining.containsAll(joining)) return prev; - return new Shard(range, nodes, fastPathElectorate, joining, pendingRemoval); + return Shard.create(range, nodes, fastPath, joining, pendingRemoval); } } @@ -112,11 +113,11 @@ public class AccordTopology .reduce(Ranges.EMPTY, Ranges::with) .mergeTouching(); - SortedArrayList fastPath = strategyFor(metadata).calculateFastPath(nodes, unavailable, dcMap); + SortedArrayList electorate = strategyFor(metadata).calculateFastPath(nodes, unavailable, dcMap); List shards = new ArrayList<>(ranges.size()); for (accord.primitives.Range range : ranges) - shards.add(lookup.createOrReuse(metadata.params.pendingDrop, range, nodes, fastPath, pending)); + shards.add(lookup.createOrReuse(metadata.params.pendingDrop, range, nodes, electorate, pending)); return shards; } @@ -267,7 +268,7 @@ public class AccordTopology return builder.build(); } - public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements, + public static Topology createAccordTopology(Epoch epoch, DistributedSchema schema, DataPlacements placements, Directory directory, AccordFastPath accordFastPath, ShardLookup lookup, AccordStaleReplicas staleReplicas) { diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index 295a58ca11..d2a2d8bbcb 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -43,7 +43,7 @@ import org.agrona.collections.ObjectHashSet; import org.apache.cassandra.index.accord.RoutesSearcher; import org.apache.cassandra.service.accord.api.AccordRoutingKey; -import static accord.local.CommandSummaries.SummaryStatus.NOT_ACCEPTED; +import static accord.local.CommandSummaries.SummaryStatus.NOT_DIRECTLY_WITNESSED; // TODO (required): move to accord-core, merge with existing logic there public class CommandsForRanges extends TreeMap implements CommandSummaries.Snapshot @@ -188,7 +188,7 @@ public class CommandsForRanges extends TreeMap implements Co if (ranges.isEmpty()) return null; - return new Summary(txnId, txnId, NOT_ACCEPTED, ranges, null, null); + return new Summary(txnId, txnId, NOT_DIRECTLY_WITNESSED, ranges, null, null); } public Summary ifRelevant(AccordCacheEntry state) 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 80508675d3..3398a6d9f8 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -38,6 +38,7 @@ import accord.primitives.Keys; import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.Seekables; +import accord.primitives.Status; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.Txn.Kind; @@ -57,6 +58,7 @@ import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.JVMStabilityInspector; import static accord.primitives.Routable.Domain.Key; +import static accord.primitives.Txn.Kind.Write; import static accord.utils.Invariants.illegalState; import static java.util.concurrent.TimeUnit.MICROSECONDS; import static java.util.concurrent.TimeUnit.MILLISECONDS; @@ -241,6 +243,13 @@ public class AccordAgent implements Agent return units.convert((1L << Math.min(retryCount, 4)), SECONDS); } + @Override + public long localExpiresAt(TxnId txnId, Status.Phase phase, TimeUnit unit) + { + // TODO (expected): make this configurable + return txnId.is(Write) ? DatabaseDescriptor.getWriteRpcTimeout(unit) : DatabaseDescriptor.getReadRpcTimeout(unit); + } + @Override public long expiresAt(ReplyContext replyContext, TimeUnit unit) { diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java index 13eebb8ce8..bbc938516c 100644 --- a/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java +++ b/src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java @@ -220,16 +220,16 @@ public class ParameterizedFastPathStrategy implements FastPathStrategy sorters.sort(Comparator.naturalOrder()); - int slowQuorum = Shard.slowPathQuorumSize(nodes.size()); + int slowQuorum = Shard.slowQuorumSize(nodes.size()); int fpSize = Math.max(size, slowQuorum); Node.Id[] array = new Node.Id[fpSize]; for (int i=0; i fastPath = new SortedArrayList<>(array); - Invariants.checkState(fastPath.size() >= slowQuorum); - return fastPath; + SortedArrayList electorate = new SortedArrayList<>(array); + Invariants.checkState(electorate.size() >= slowQuorum); + return electorate; } private static ConfigurationException cfe(String fmt, Object... args) diff --git a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java index 8a278faaa2..c69f0d840e 100644 --- a/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java +++ b/src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java @@ -59,7 +59,7 @@ public class SimpleFastPathStrategy implements FastPathStrategy Node.Id[] array = new Node.Id[nodes.size() - discarded]; System.arraycopy(tmp, 0, array, 0, nodes.size() - discarded); SortedArrayList fastPath = new SortedArrayList<>(array); - Invariants.checkState(fastPath.size() >= Shard.slowPathQuorumSize(nodes.size())); + Invariants.checkState(fastPath.size() >= Shard.slowQuorumSize(nodes.size())); return fastPath; } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java index 6acf5cfc18..8f4e51116a 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java @@ -41,6 +41,7 @@ import accord.primitives.TxnId; import accord.primitives.Unseekables; import accord.primitives.Writes; import accord.topology.Topologies; +import accord.utils.UnhandledEnum; import org.agrona.collections.Int2ObjectHashMap; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.io.IVersionedSerializer; @@ -119,10 +120,14 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL default: throw new AssertionError(); case NotDefined: case PreAccepted: - case Accepted: + case PreNotAccepted: + case NotAccepted: case AcceptedInvalidate: + case AcceptedMedium: + case AcceptedSlow: case PreCommitted: case Committed: + case Stable: case PreApplied: LocalListeners.Registered listener = safeStore.register(txnId, this); synchronized (this) @@ -239,14 +244,18 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL Command command = safeCommand.current(); switch (command.status()) { - default: throw new AssertionError(); + default: throw new UnhandledEnum(command.status()); case NotDefined: case PreAccepted: - case Accepted: + case PreNotAccepted: + case NotAccepted: case AcceptedInvalidate: + case AcceptedMedium: + case AcceptedSlow: case PreCommitted: case Committed: case PreApplied: + case Stable: return true; case Applied: diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java deleted file mode 100644 index 8b94515330..0000000000 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * 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.interop; - -import javax.annotation.Nonnull; -import javax.annotation.Nullable; - -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; -import accord.primitives.PartialTxn; -import accord.primitives.Route; -import accord.primitives.Timestamp; -import accord.primitives.Txn; -import accord.primitives.TxnId; -import accord.topology.Topologies; -import accord.topology.Topology; -import org.apache.cassandra.io.IVersionedSerializer; -import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType; -import org.apache.cassandra.service.accord.serializers.CommitSerializers.CommitSerializer; - -public class AccordInteropCommit extends Commit -{ - public static final IVersionedSerializer serializer = new CommitSerializer<>(AccordInteropRead.class, AccordInteropRead.requestSerializer) - { - @Override - protected AccordInteropCommit deserializeCommit(TxnId txnId, Route scope, long waitForEpoch, long minEpoch, Kind kind, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nullable ReadData read) - { - return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, minEpoch, ballot, executeAt, partialTxn, partialDeps, fullRoute, read); - } - }; - - public AccordInteropCommit(Kind kind, TxnId txnId, Route scope, long waitForEpoch, long minEpoch, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute fullRoute, @Nonnull ReadData readData) - { - super(kind, txnId, scope, waitForEpoch, minEpoch, 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, Ballot.ZERO, executeAt, deps, read); - } - - @Override - public MessageType type() - { - switch (kind) - { - 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 cf1a91e8be..8330f30f2c 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -41,7 +41,6 @@ import accord.local.Node; import accord.local.Node.Id; import accord.messages.Commit; import accord.messages.Commit.Kind; -import accord.messages.ReadTxnData; import accord.primitives.Ballot; import accord.primitives.Deps; import accord.primitives.FullRoute; @@ -99,6 +98,7 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.transport.Dispatcher; import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard; +import static accord.primitives.Txn.Kind.Write; import static accord.utils.Invariants.checkArgument; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteMetrics; @@ -234,10 +234,8 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen { Node.Id id = endpointMapper.mappedId(to); SinglePartitionReadCommand command = (SinglePartitionReadCommand) message.payload; - AccordInteropRead read = new AccordInteropRead(id, executes, txnId, readScope, executeAt.epoch(), command); // TODO (required): understand interop and whether StableFastPath is appropriate - AccordInteropCommit commit = new AccordInteropCommit(Kind.StableFastPath, id, coordinateTopology, allTopologies, - txnId, txn, route, executeAt, deps, read); + AccordInteropStableThenRead commit = new AccordInteropStableThenRead(id, allTopologies, txnId, Kind.StableFastPath, executeAt, txn, deps, route, command); node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this)); } @@ -384,7 +382,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen { for (Node.Id to : executeTopology.nodes()) if (!contacted.contains(endpointMapper.mappedEndpoint(to))) - node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); + node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps)); } public void start() @@ -394,7 +392,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen for (Node.Id to : allTopologies.nodes()) { if (!executeTopology.contains(to)) - node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); + node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps)); } } AsyncChain result; @@ -406,7 +404,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen CommandStore cs = node.commandStores().select(route.homeKey()); result.beginAsResult().withExecutor(cs).begin((data, failure) -> { if (failure == null) - ((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback); + ((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, txnId, txn, executeAt, deps, txnId.is(Write) ? txn.execute(txnId, executeAt, data) : null, txn.result(txnId, executeAt, data), callback); else callback.accept(null, failure); }); @@ -420,7 +418,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen // and can be extended similar to MessageType which allows additional types not from Accord to be added // This commit won't necessarily execute before the interop read repair message so there could be an insufficient which is fine for (Node.Id to : executeTopology.nodes()) - node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null)); + node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps)); repairUpdate.runBRR(AccordInteropExecution.this); return new TxnData(); }); 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 342dc40140..08d09f8960 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java @@ -49,6 +49,7 @@ public class AccordInteropPersist extends Persist { private static class CallbackHolder { + boolean isDone = false; private final ResponseTracker tracker; private final Result result; private final BiConsumer clientCallback; @@ -63,13 +64,18 @@ public class AccordInteropPersist extends Persist private void handleStatus(RequestStatus status) { + if (isDone) + return; + switch (status) { default: throw new IllegalStateException("Unhandled request status " + status); case Success: + isDone = true; clientCallback.accept(result, null); return; case Failed: + isDone = true; clientCallback.accept(null, failure); return; case NoChange: @@ -87,6 +93,16 @@ public class AccordInteropPersist extends Persist failure = Throwables.merge(failure, throwable); handleStatus(tracker.recordFailure(node)); } + + boolean recordCallbackFailure(Throwable throwable) + { + if (isDone) + return false; + isDone = true; + failure = Throwables.merge(failure, throwable); + clientCallback.accept(null, failure); + return true; + } } private final ConsistencyLevel consistencyLevel; @@ -143,8 +159,8 @@ public class AccordInteropPersist extends Persist } @Override - public void onCallbackFailure(Node.Id from, Throwable failure) + public boolean onCallbackFailure(Node.Id from, Throwable failure) { - callback.recordFailure(from, failure); + return callback.recordCallbackFailure(failure); } } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java index a6ab1831e8..9afa576063 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java @@ -56,13 +56,13 @@ import static accord.primitives.SaveStatus.ReadyToExecute; public class AccordInteropRead extends ReadData { - public static final IVersionedSerializer requestSerializer = new ReadDataSerializer() + public static final IVersionedSerializer requestSerializer = new ReadDataSerializer<>() { @Override public void serialize(AccordInteropRead read, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(read.txnId, out, version); - KeySerializers.participants.serialize(read.readScope, out, version); + KeySerializers.participants.serialize(read.scope, out, version); out.writeUnsignedVInt(read.executeAtEpoch); SinglePartitionReadCommand.serializer.serialize(read.command, out, version); } @@ -71,17 +71,17 @@ public class AccordInteropRead extends ReadData public AccordInteropRead deserialize(DataInputPlus in, int version) throws IOException { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); - Participants readScope = KeySerializers.participants.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); long executeAtEpoch = in.readUnsignedVInt(); SinglePartitionReadCommand command = (SinglePartitionReadCommand) SinglePartitionReadCommand.serializer.deserialize(in, version); - return new AccordInteropRead(txnId, readScope, executeAtEpoch, command); + return new AccordInteropRead(txnId, scope, executeAtEpoch, command); } @Override public long serializedSize(AccordInteropRead read, int version) { return CommandSerializers.txnId.serializedSize(read.txnId, version) - + KeySerializers.participants.serializedSize(read.readScope, version) + + KeySerializers.participants.serializedSize(read.scope, version) + TypeSizes.sizeofUnsignedVInt(read.executeAtEpoch) + SinglePartitionReadCommand.serializer.serializedSize(read.command, version); } @@ -148,17 +148,17 @@ public class AccordInteropRead extends ReadData private static final ExecuteOn EXECUTE_ON = new ExecuteOn(ReadyToExecute, PreApplied); - private final SinglePartitionReadCommand command; + final SinglePartitionReadCommand command; - public AccordInteropRead(Node.Id to, Topologies topologies, TxnId txnId, Participants readScope, long executeAtEpoch, SinglePartitionReadCommand command) + public AccordInteropRead(Node.Id to, Topologies topologies, TxnId txnId, Participants scope, long executeAtEpoch, SinglePartitionReadCommand command) { - super(to, topologies, txnId, readScope, executeAtEpoch); + super(to, topologies, txnId, scope, executeAtEpoch); this.command = command; } - public AccordInteropRead(TxnId txnId, Participants readScope, long executeAtEpoch, SinglePartitionReadCommand command) + public AccordInteropRead(TxnId txnId, Participants scope, long executeAtEpoch, SinglePartitionReadCommand command) { - super(txnId, readScope, executeAtEpoch); + super(txnId, scope, executeAtEpoch); this.command = command; } 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 89b6577b83..b2a9cc17b4 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java @@ -90,8 +90,9 @@ public abstract class AccordInteropReadCallback implements Callback readScope = KeySerializers.participants.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); long executeAtEpoch = in.readUnsignedVInt(); Mutation mutation = Mutation.serializer.deserialize(in, version); - return new AccordInteropReadRepair(txnId, readScope, executeAtEpoch, mutation); + return new AccordInteropReadRepair(txnId, scope, executeAtEpoch, mutation); } @Override public long serializedSize(AccordInteropReadRepair repair, int version) { return CommandSerializers.txnId.serializedSize(repair.txnId, version) - + KeySerializers.participants.serializedSize(repair.readScope, version) + + KeySerializers.participants.serializedSize(repair.scope, version) + TypeSizes.sizeofUnsignedVInt(repair.executeAtEpoch) + Mutation.serializer.serializedSize(repair.mutation, version); } @@ -120,16 +120,16 @@ public class AccordInteropReadRepair extends ReadData public static final IVersionedSerializer replySerializer = new ReadDataSerializers.ReplySerializer<>(noop_data_serializer); - public AccordInteropReadRepair(Node.Id to, Topologies topologies, TxnId txnId, Participants readScope, long executeAtEpoch, Mutation mutation) + public AccordInteropReadRepair(Node.Id to, Topologies topologies, TxnId txnId, Participants scope, long executeAtEpoch, Mutation mutation) { - super(to, topologies, txnId, readScope, executeAtEpoch); + super(to, topologies, txnId, scope, executeAtEpoch); this.mutation = mutation; } - public AccordInteropReadRepair(TxnId txnId, Participants readScope, long executeAtEpoch, Mutation mutation) + public AccordInteropReadRepair(TxnId txnId, Participants scope, long executeAtEpoch, Mutation mutation) { // TODO (review): remove followup read - Is there anything left to be done for this or can I remove it? - super(txnId, readScope, executeAtEpoch); + super(txnId, scope, executeAtEpoch); this.mutation = mutation; } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropStableThenRead.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropStableThenRead.java new file mode 100644 index 0000000000..2093a969ce --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropStableThenRead.java @@ -0,0 +1,181 @@ +/* + * 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.interop; + +import java.io.IOException; +import javax.annotation.Nullable; + +import accord.local.Commands; +import accord.local.Node; +import accord.local.SafeCommand; +import accord.local.SafeCommandStore; +import accord.local.StoreParticipants; +import accord.messages.Commit; +import accord.messages.MessageType; +import accord.primitives.Ballot; +import accord.primitives.Deps; +import accord.primitives.FullRoute; +import accord.primitives.PartialDeps; +import accord.primitives.PartialTxn; +import accord.primitives.Participants; +import accord.primitives.Route; +import accord.primitives.Timestamp; +import accord.primitives.Txn; +import accord.primitives.TxnId; +import accord.topology.Topologies; +import org.apache.cassandra.db.SinglePartitionReadCommand; +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType; +import org.apache.cassandra.service.accord.serializers.CommandSerializers; +import org.apache.cassandra.service.accord.serializers.CommitSerializers; +import org.apache.cassandra.service.accord.serializers.DepsSerializers; +import org.apache.cassandra.service.accord.serializers.KeySerializers; +import org.apache.cassandra.service.accord.serializers.ReadDataSerializers.ReadDataSerializer; + +import static accord.messages.Commit.WithDeps.HasDeps; +import static accord.messages.Commit.WithDeps.NoDeps; +import static accord.messages.Commit.WithTxn.HasTxn; +import static accord.messages.Commit.WithTxn.NoTxn; +import static accord.primitives.SaveStatus.PreApplied; +import static accord.primitives.SaveStatus.ReadyToExecute; + +public class AccordInteropStableThenRead extends AccordInteropRead +{ + // TODO (desired): duplicates a lot of StableThenReadSerializer + public static final IVersionedSerializer requestSerializer = new ReadDataSerializer<>() + { + @Override + public void serialize(AccordInteropStableThenRead read, DataOutputPlus out, int version) throws IOException + { + CommandSerializers.txnId.serialize(read.txnId, out, version); + KeySerializers.participants.serialize(read.scope, out, version); + CommitSerializers.kind.serialize(read.kind, out, version); + out.writeUnsignedVInt(read.minEpoch); + CommandSerializers.timestamp.serialize(read.executeAt, out, version); + if (read.kind.withTxn != NoTxn) + CommandSerializers.nullablePartialTxn.serialize(read.partialTxn, out, version); + if (read.kind.withDeps == HasDeps) + DepsSerializers.partialDeps.serialize(read.partialDeps, out, version); + if (read.kind.withTxn == HasTxn) + KeySerializers.fullRoute.serialize(read.route, out, version); + SinglePartitionReadCommand.serializer.serialize(read.command, out, version); + } + + @Override + public AccordInteropStableThenRead deserialize(DataInputPlus in, int version) throws IOException + { + TxnId txnId = CommandSerializers.txnId.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); + Commit.Kind kind = CommitSerializers.kind.deserialize(in, version); + long minEpoch = in.readUnsignedVInt(); + Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version); + PartialTxn partialTxn = kind.withTxn == NoTxn ? null : CommandSerializers.nullablePartialTxn.deserialize(in, version); + PartialDeps partialDeps = kind.withDeps == NoDeps ? null : DepsSerializers.partialDeps.deserialize(in, version); + FullRoute < ?> route = kind.withTxn == HasTxn ? KeySerializers.fullRoute.deserialize(in, version) : null; + SinglePartitionReadCommand command = (SinglePartitionReadCommand) SinglePartitionReadCommand.serializer.deserialize(in, version); + return new AccordInteropStableThenRead(txnId, scope, kind, minEpoch, executeAt, partialTxn, partialDeps, route, command); + } + + @Override + public long serializedSize(AccordInteropStableThenRead read, int version) + { + return CommandSerializers.txnId.serializedSize(read.txnId, version) + + KeySerializers.participants.serializedSize(read.scope, version) + + CommitSerializers.kind.serializedSize(read.kind, version) + + TypeSizes.sizeofUnsignedVInt(read.minEpoch) + + CommandSerializers.timestamp.serializedSize(read.executeAt, version) + + (read.kind.withTxn == NoTxn ? 0 : CommandSerializers.nullablePartialTxn.serializedSize(read.partialTxn, version)) + + (read.kind.withDeps != HasDeps ? 0 : DepsSerializers.partialDeps.serializedSize(read.partialDeps, version)) + + (read.kind.withTxn != HasTxn ? 0 : KeySerializers.fullRoute.serializedSize(read.route, version)) + + SinglePartitionReadCommand.serializer.serializedSize(read.command, version); + } + }; + + // TODO (required): why is this safe to execute at PreApplied? Document. + private static final ExecuteOn EXECUTE_ON = new ExecuteOn(ReadyToExecute, PreApplied); + + public final long minEpoch; + public final Commit.Kind kind; + public final Timestamp executeAt; + public final @Nullable PartialTxn partialTxn; + public final @Nullable PartialDeps partialDeps; + public final @Nullable FullRoute route; + + public AccordInteropStableThenRead(Node.Id to, Topologies topologies, TxnId txnId, Commit.Kind kind, Timestamp executeAt, Txn txn, Deps deps, FullRoute route, SinglePartitionReadCommand command) + { + super(to, topologies, txnId, route, executeAt.epoch(), command); + this.kind = kind; + this.minEpoch = topologies.oldestEpoch(); + this.executeAt = executeAt; + this.partialTxn = kind.withTxn.select(txn, scope, topologies, txnId, to); + this.partialDeps = kind.withDeps.select(deps, scope); + this.route = kind.withTxn.select(route); + } + + public AccordInteropStableThenRead(TxnId txnId, Participants scope, Commit.Kind kind, long minEpoch, Timestamp executeAt, @Nullable PartialTxn partialTxn, @Nullable PartialDeps partialDeps, @Nullable FullRoute route, SinglePartitionReadCommand command) + { + super(txnId, scope, executeAt.epoch(), command); + this.minEpoch = minEpoch; + this.kind = kind; + this.executeAt = executeAt; + this.partialTxn = partialTxn; + this.partialDeps = partialDeps; + this.route = route; + } + + @Override + public CommitOrReadNack apply(SafeCommandStore safeStore) + { + Route route = this.route == null ? (Route)scope : this.route; + StoreParticipants participants = StoreParticipants.execute(safeStore, route, txnId, minEpoch(), executeAtEpoch); + SafeCommand safeCommand = safeStore.get(txnId, participants); + Commands.commit(safeStore, safeCommand, participants, kind.saveStatus, Ballot.ZERO, txnId, route, partialTxn, executeAt, partialDeps, kind); + return super.apply(safeStore, safeCommand, participants); + } + + @Override + public ReadType kind() + { + return ReadType.stableThenRead; + } + + @Override + protected ExecuteOn executeOn() + { + return EXECUTE_ON; + } + + @Override + public MessageType type() + { + return AccordMessageType.INTEROP_STABLE_THEN_READ_REQ; + } + + @Override + public String toString() + { + return "AccordInteropStableThenRead{" + + "txnId=" + txnId + + "command=" + command + + '}'; + } +} diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java index 05c027e80f..bdaf24be8c 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java @@ -41,9 +41,12 @@ public class AcceptSerializers public static final IVersionedSerializer request = new TxnRequestSerializer.WithUnsyncedSerializer<>() { + final IVersionedSerializer kindSerializer = new EnumSerializer<>(Accept.Kind.class); + @Override public void serializeBody(Accept accept, DataOutputPlus out, int version) throws IOException { + kindSerializer.serialize(accept.kind, out, version); CommandSerializers.ballot.serialize(accept.ballot, out, version); CommandSerializers.timestamp.serialize(accept.executeAt, out, version); DepsSerializers.partialDeps.serialize(accept.partialDeps, out, version); @@ -53,6 +56,7 @@ public class AcceptSerializers public Accept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route scope, long waitForEpoch, long minEpoch) throws IOException { return create(txnId, scope, waitForEpoch, minEpoch, + kindSerializer.deserialize(in, version), CommandSerializers.ballot.deserialize(in, version), CommandSerializers.timestamp.deserialize(in, version), DepsSerializers.partialDeps.deserialize(in, version)); @@ -61,36 +65,40 @@ public class AcceptSerializers @Override public long serializedBodySize(Accept accept, int version) { - return CommandSerializers.ballot.serializedSize(accept.ballot, version) + return kindSerializer.serializedSize(accept.kind, version) + + CommandSerializers.ballot.serializedSize(accept.ballot, version) + CommandSerializers.timestamp.serializedSize(accept.executeAt, version) + DepsSerializers.partialDeps.serializedSize(accept.partialDeps, version); } }; - public static final IVersionedSerializer invalidate = new IVersionedSerializer() + public static final IVersionedSerializer notAccept = new IVersionedSerializer<>() { @Override - public void serialize(Accept.Invalidate invalidate, DataOutputPlus out, int version) throws IOException + public void serialize(Accept.NotAccept invalidate, DataOutputPlus out, int version) throws IOException { + CommandSerializers.status.serialize(invalidate.status, out, version); CommandSerializers.ballot.serialize(invalidate.ballot, out, version); CommandSerializers.txnId.serialize(invalidate.txnId, out, version); - KeySerializers.routingKey.serialize(invalidate.someKey, out, version); + KeySerializers.participants.serialize(invalidate.participants, out, version); } @Override - public Accept.Invalidate deserialize(DataInputPlus in, int version) throws IOException + public Accept.NotAccept deserialize(DataInputPlus in, int version) throws IOException { - return new Accept.Invalidate(CommandSerializers.ballot.deserialize(in, version), - CommandSerializers.txnId.deserialize(in, version), - KeySerializers.routingKey.deserialize(in, version)); + return new Accept.NotAccept(CommandSerializers.status.deserialize(in, version), + CommandSerializers.ballot.deserialize(in, version), + CommandSerializers.txnId.deserialize(in, version), + KeySerializers.participants.deserialize(in, version)); } @Override - public long serializedSize(Accept.Invalidate invalidate, int version) + public long serializedSize(Accept.NotAccept invalidate, int version) { - return CommandSerializers.ballot.serializedSize(invalidate.ballot, version) + return CommandSerializers.status.serializedSize(invalidate.status, version) + + CommandSerializers.ballot.serializedSize(invalidate.ballot, version) + CommandSerializers.txnId.serializedSize(invalidate.txnId, version) - + KeySerializers.routingKey.serializedSize(invalidate.someKey, version); + + KeySerializers.participants.serializedSize(invalidate.participants, version); } }; @@ -113,7 +121,7 @@ public class AcceptSerializers } else { - Invariants.checkState(reply == AcceptReply.ACCEPT_INVALIDATE); + Invariants.checkState(reply == AcceptReply.SUCCESS); out.writeByte(2); } break; @@ -141,7 +149,7 @@ public class AcceptSerializers case 1: return new AcceptReply(DepsSerializers.deps.deserialize(in, version)); case 2: - return AcceptReply.ACCEPT_INVALIDATE; + return AcceptReply.SUCCESS; case 3: return new AcceptReply(CommandSerializers.ballot.deserialize(in, version)); case 4: diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java index 9422738e3d..6189e9c099 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java @@ -35,6 +35,8 @@ import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; +import static accord.primitives.Txn.Kind.Write; + public class ApplySerializers { @@ -68,7 +70,8 @@ public class ApplySerializers DepsSerializers.partialDeps.serialize(apply.deps, out, version); CommandSerializers.nullablePartialTxn.serialize(apply.txn, out, version); KeySerializers.nullableFullRoute.serialize(apply.fullRoute, out, version); - CommandSerializers.writes.serialize(apply.writes, out, version); + if (apply.txnId.is(Write)) + CommandSerializers.writes.serialize(apply.writes, out, version); } protected abstract A deserializeApply(TxnId txnId, Route scope, long minEpoch, long waitForEpoch, Apply.Kind kind, @@ -83,7 +86,7 @@ public class ApplySerializers DepsSerializers.partialDeps.deserialize(in, version), CommandSerializers.nullablePartialTxn.deserialize(in, version), KeySerializers.nullableFullRoute.deserialize(in, version), - CommandSerializers.writes.deserialize(in, version), + (txnId.is(Write) ? CommandSerializers.writes.deserialize(in, version) : null), ResultSerializers.APPLIED); } @@ -96,7 +99,7 @@ public class ApplySerializers + DepsSerializers.partialDeps.serializedSize(apply.deps, version) + CommandSerializers.nullablePartialTxn.serializedSize(apply.txn, version) + KeySerializers.nullableFullRoute.serializedSize(apply.fullRoute, version) - + CommandSerializers.writes.serializedSize(apply.writes, version); + + (apply.txnId.is(Write) ? CommandSerializers.writes.serializedSize(apply.writes, version) : 0); } } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java index 0006a068ec..a93b737568 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java @@ -44,7 +44,7 @@ public class AwaitSerializer { CommandSerializers.txnId.serialize(await.txnId, out, version); KeySerializers.participants.serialize(await.scope, out, version); - out.writeByte(await.blockedUntil.ordinal()); + out.writeByte((await.blockedUntil.ordinal() << 1) | (await.notifyProgressLog ? 1 : 0)); out.writeUnsignedVInt(await.maxAwaitEpoch - await.txnId.epoch()); out.writeUnsignedVInt(await.maxAwaitEpoch - await.minAwaitEpoch); out.writeUnsignedVInt32(await.callbackId + 1); @@ -56,12 +56,14 @@ public class AwaitSerializer { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); Participants scope = KeySerializers.participants.deserialize(in, version); - BlockedUntil blockedUntil = BlockedUntil.forOrdinal(in.readByte()); + int blockedAndNotify = in.readByte(); + BlockedUntil blockedUntil = BlockedUntil.forOrdinal(blockedAndNotify >>> 1); + boolean notifyProgressLog = (blockedAndNotify & 1) == 1; long maxAwaitEpoch = in.readUnsignedVInt() + txnId.epoch(); long minAwaitEpoch = maxAwaitEpoch - in.readUnsignedVInt(); int callbackId = in.readUnsignedVInt32() - 1; Invariants.checkState(callbackId >= -1); - return Await.SerializerSupport.create(txnId, scope, blockedUntil, minAwaitEpoch, maxAwaitEpoch, callbackId); + return Await.SerializerSupport.create(txnId, scope, blockedUntil, notifyProgressLog, minAwaitEpoch, maxAwaitEpoch, callbackId); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java index 7c44b60835..d81927893d 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java @@ -223,7 +223,7 @@ public class CheckStatusSerializers Writes writes = CommandSerializers.nullableWrites.deserialize(in, version); Result result = null; - if (maxKnowledgeStatus.known.outcome.isOrWasApply()) + if (maxKnowledgeStatus.known.outcome().isOrWasApply()) result = ResultSerializers.APPLIED; return createOk(map, maxKnowledgeStatus, maxStatus, maxPromised, maxAcceptedOrCommitted, acceptedOrCommitted, executeAt, diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java index 46348a2a8b..32bf98a65f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -436,31 +436,19 @@ public class CommandSerializers @Override public void serialize(Known known, DataOutputPlus out, int version) throws IOException { - knownRoute.serialize(known.route, out, version); - definition.serialize(known.definition, out, version); - knownExecuteAt.serialize(known.executeAt, out, version); - knownDeps.serialize(known.deps, out, version); - outcome.serialize(known.outcome, out, version); + out.writeUnsignedVInt32(known.encoded); } @Override public Known deserialize(DataInputPlus in, int version) throws IOException { - return new Known(knownRoute.deserialize(in, version), - definition.deserialize(in, version), - knownExecuteAt.deserialize(in, version), - knownDeps.deserialize(in, version), - outcome.deserialize(in, version)); + return new Known(in.readUnsignedVInt32()); } @Override public long serializedSize(Known known, int 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); + return TypeSizes.sizeofUnsignedVInt(known.encoded); } }; 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 451b26fdfc..51555c1618 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java @@ -19,10 +19,8 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; -import javax.annotation.Nullable; import accord.messages.Commit; -import accord.messages.ReadData; import accord.primitives.Ballot; import accord.primitives.FullRoute; import accord.primitives.PartialDeps; @@ -35,7 +33,6 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.utils.CastingSerializer; import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; @@ -43,19 +40,13 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi public class CommitSerializers { - private static final IVersionedSerializer kind = new EnumSerializer<>(Commit.Kind.class); + public static final IVersionedSerializer kind = new EnumSerializer<>(Commit.Kind.class); - public abstract static class CommitSerializer extends TxnRequestSerializer.WithUnsyncedSerializer + public static final CommitSerializer request = new CommitSerializer(); + public static class CommitSerializer extends TxnRequestSerializer.WithUnsyncedSerializer { - private final IVersionedSerializer read; - - public CommitSerializer(Class klass, IVersionedSerializer read) - { - this.read = new CastingSerializer<>(klass, read); - } - @Override - public void serializeBody(C msg, DataOutputPlus out, int version) throws IOException + public void serializeBody(Commit msg, DataOutputPlus out, int version) throws IOException { kind.serialize(msg.kind, out, version); CommandSerializers.ballot.serialize(msg.ballot, out, version); @@ -63,49 +54,32 @@ public class CommitSerializers CommandSerializers.nullablePartialTxn.serialize(msg.partialTxn, out, version); DepsSerializers.partialDeps.serialize(msg.scope, msg.partialDeps, out, version); serializeNullable(msg.route, out, version, KeySerializers.fullRoute); - serializeNullable(msg.readData, out, version, read); } - protected abstract C deserializeCommit(TxnId txnId, Route scope, long waitForEpoch, long minEpoch, Commit.Kind kind, - Ballot ballot, Timestamp executeAt, - @Nullable PartialTxn partialTxn, PartialDeps partialDeps, - @Nullable FullRoute fullRoute, @Nullable ReadData read); - @Override - public C deserializeBody(DataInputPlus in, int version, TxnId txnId, Route scope, long waitForEpoch, long minEpoch) throws IOException + public Commit deserializeBody(DataInputPlus in, int version, TxnId txnId, Route scope, long waitForEpoch, long minEpoch) throws IOException { Commit.Kind kind = CommitSerializers.kind.deserialize(in, version); Ballot ballot = CommandSerializers.ballot.deserialize(in, version); Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version); - PartialTxn txn = CommandSerializers.nullablePartialTxn.deserialize(in, version); - PartialDeps deps = DepsSerializers.partialDeps.deserialize(scope, in, version); + PartialTxn partialTxn = CommandSerializers.nullablePartialTxn.deserialize(in, version); + PartialDeps partialDeps = DepsSerializers.partialDeps.deserialize(scope, in, version); FullRoute route = deserializeNullable(in, version, KeySerializers.fullRoute); - ReadData read = deserializeNullable(in, version, this.read); - return deserializeCommit(txnId, scope, waitForEpoch, minEpoch, kind, ballot, executeAt, txn, deps, route, read); + return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch, kind, ballot, executeAt, partialTxn, partialDeps, route); } @Override - public long serializedBodySize(C msg, int version) + public long serializedBodySize(Commit 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) + DepsSerializers.partialDeps.serializedSize(msg.scope, msg.partialDeps, version) - + serializedNullableSize(msg.route, version, KeySerializers.fullRoute) - + serializedNullableSize(msg.readData, version, read); + + serializedNullableSize(msg.route, version, KeySerializers.fullRoute); } } - public static final IVersionedSerializer request = new CommitSerializer(ReadData.class, ReadDataSerializers.readData) - { - @Override - protected Commit deserializeCommit(TxnId txnId, Route scope, long waitForEpoch, long minEpoch, 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, minEpoch, kind, ballot, executeAt, partialTxn, partialDeps, fullRoute, read); - } - }; - public static final IVersionedSerializer invalidate = new IVersionedSerializer<>() { @Override 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 34a0df8f6b..42861e1a47 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java @@ -48,7 +48,7 @@ public class FetchSerializers { out.writeUnsignedVInt(request.executeAtEpoch); CommandSerializers.txnId.serialize(request.txnId, out, version); - KeySerializers.ranges.serialize((Ranges) request.readScope, out, version); + KeySerializers.ranges.serialize((Ranges) request.scope, out, version); DepsSerializers.partialDeps.serialize(request.partialDeps, out, version); StreamingTxn.serializer.serialize(request.read, out, version); } @@ -68,7 +68,7 @@ public class FetchSerializers { return TypeSizes.sizeofUnsignedVInt(request.executeAtEpoch) + CommandSerializers.txnId.serializedSize(request.txnId, version) - + KeySerializers.ranges.serializedSize((Ranges) request.readScope, version) + + KeySerializers.ranges.serializedSize((Ranges) request.scope, version) + DepsSerializers.partialDeps.serializedSize(request.partialDeps, version) + StreamingTxn.serializer.serializedSize(request.read, version); } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java index 957eb79263..23cba8e903 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java @@ -25,6 +25,7 @@ import accord.messages.PreAccept; import accord.messages.PreAccept.PreAcceptOk; import accord.messages.PreAccept.PreAcceptReply; import accord.primitives.FullRoute; +import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.Route; import accord.primitives.TxnId; @@ -34,10 +35,6 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.service.accord.serializers.TxnRequestSerializer.WithUnsyncedSerializer; -import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; -import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; -import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize; - public class PreacceptSerializers { private PreacceptSerializers() {} @@ -47,27 +44,40 @@ public class PreacceptSerializers @Override public void serializeBody(PreAccept msg, DataOutputPlus out, int version) throws IOException { + int flags = (msg.partialDeps == null ? 0 : 1) + | (msg.route == null ? 0 : 2) + | (msg.hasCoordinatorVote ? 4 : 0) + | (msg.acceptEpoch == msg.minEpoch ? 0 : 8); + out.writeByte(flags); CommandSerializers.partialTxn.serialize(msg.partialTxn, out, version); - serializeNullable(msg.route, out, version, KeySerializers.fullRoute); - out.writeUnsignedVInt(msg.acceptEpoch - msg.minEpoch); + if (msg.partialDeps != null) + DepsSerializers.partialDeps.serialize(msg.partialDeps, out, version); + if (msg.route != null) + KeySerializers.fullRoute.serialize(msg.route, out, version); + if (msg.acceptEpoch != msg.minEpoch) + out.writeUnsignedVInt(msg.acceptEpoch - msg.minEpoch); } @Override public PreAccept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route scope, long waitForEpoch, long minEpoch) throws IOException { + byte flags = in.readByte(); PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version); - @Nullable FullRoute fullRoute = deserializeNullable(in, version, KeySerializers.fullRoute); - long acceptEpoch = in.readUnsignedVInt() + minEpoch; - return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch, - acceptEpoch, partialTxn, fullRoute); + @Nullable PartialDeps partialDeps = (flags & 1) == 0 ? null : DepsSerializers.partialDeps.deserialize(in, version); + @Nullable FullRoute fullRoute = (flags & 2) == 0 ? null : KeySerializers.fullRoute.deserialize(in, version); + boolean hasCoordinatorVote = (flags & 4) != 0; + long acceptEpoch = (flags & 8) == 0 ? minEpoch : in.readUnsignedVInt() + minEpoch; + return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch, acceptEpoch, partialTxn, partialDeps, hasCoordinatorVote, fullRoute); } @Override public long serializedBodySize(PreAccept msg, int version) { - return CommandSerializers.partialTxn.serializedSize(msg.partialTxn, version) - + serializedNullableSize(msg.route, version, KeySerializers.fullRoute) - + TypeSizes.sizeofUnsignedVInt(msg.acceptEpoch - msg.minEpoch); + return TypeSizes.BYTE_SIZE + + CommandSerializers.partialTxn.serializedSize(msg.partialTxn, version) + + (msg.partialDeps == null ? 0 : DepsSerializers.partialDeps.serializedSize(msg.partialDeps, version)) + + (msg.route == null ? 0 : KeySerializers.fullRoute.serializedSize(msg.route, version)) + + (msg.acceptEpoch == msg.minEpoch ? 0 : TypeSizes.sizeofUnsignedVInt(msg.acceptEpoch - msg.minEpoch)); } }; 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 9d85c63355..453ba69b1f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java @@ -22,6 +22,7 @@ import java.io.IOException; import accord.api.Data; import accord.messages.ApplyThenWaitUntilApplied; +import accord.messages.Commit; import accord.messages.ReadData; import accord.messages.ReadData.CommitOrReadNack; import accord.messages.ReadData.ReadOk; @@ -29,12 +30,14 @@ import accord.messages.ReadData.ReadReply; import accord.messages.ReadData.ReadType; import accord.messages.ReadEphemeralTxnData; import accord.messages.ReadTxnData; +import accord.messages.StableThenRead; import accord.messages.WaitUntilApplied; import accord.primitives.FullRoute; import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.Participants; import accord.primitives.Ranges; +import accord.primitives.Timestamp; import accord.primitives.TxnId; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; @@ -42,6 +45,10 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.service.accord.txn.TxnData; +import static accord.messages.Commit.WithDeps.HasDeps; +import static accord.messages.Commit.WithDeps.NoDeps; +import static accord.messages.Commit.WithTxn.HasTxn; +import static accord.messages.Commit.WithTxn.NoTxn; import static org.apache.cassandra.db.TypeSizes.sizeof; import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; @@ -79,13 +86,13 @@ public class ReadDataSerializers public void serialize(ApplyThenWaitUntilApplied msg, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(msg.txnId, out, version); - KeySerializers.participants.serialize(msg.readScope, out, version); + KeySerializers.participants.serialize(msg.scope, out, version); out.writeUnsignedVInt(msg.minEpoch()); CommandSerializers.timestamp.serialize(msg.executeAt, out, version); KeySerializers.fullRoute.serialize(msg.route, out, version); CommandSerializers.partialTxn.serialize(msg.txn, out, version); DepsSerializers.partialDeps.serialize(msg.deps, out, version); - CommandSerializers.writes.serialize(msg.writes, out, version); + CommandSerializers.nullableWrites.serialize(msg.writes, out, version); } @Override @@ -99,7 +106,7 @@ public class ReadDataSerializers KeySerializers.fullRoute.deserialize(in, version), CommandSerializers.partialTxn.deserialize(in, version), DepsSerializers.partialDeps.deserialize(in, version), - CommandSerializers.writes.deserialize(in, version), + CommandSerializers.nullableWrites.deserialize(in, version), ResultSerializers.APPLIED); } @@ -107,13 +114,13 @@ public class ReadDataSerializers public long serializedSize(ApplyThenWaitUntilApplied msg, int version) { return CommandSerializers.txnId.serializedSize(msg.txnId, version) - + KeySerializers.participants.serializedSize(msg.readScope, version) + + KeySerializers.participants.serializedSize(msg.scope, version) + TypeSizes.sizeofUnsignedVInt(msg.minEpoch()) + CommandSerializers.timestamp.serializedSize(msg.executeAt, version) + KeySerializers.fullRoute.serializedSize(msg.route, version) + CommandSerializers.partialTxn.serializedSize(msg.txn, version) + DepsSerializers.partialDeps.serializedSize(msg.deps, version) - + CommandSerializers.writes.serializedSize(msg.writes, version); + + CommandSerializers.nullableWrites.serializedSize(msg.writes, version); } } @@ -123,7 +130,7 @@ public class ReadDataSerializers public void serialize(ReadTxnData read, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(read.txnId, out, version); - KeySerializers.participants.serialize(read.readScope, out, version); + KeySerializers.participants.serialize(read.scope, out, version); out.writeUnsignedVInt(read.executeAtEpoch); } @@ -131,16 +138,16 @@ public class ReadDataSerializers public ReadTxnData deserialize(DataInputPlus in, int version) throws IOException { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); - Participants readScope = KeySerializers.participants.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); long executeAtEpoch = in.readUnsignedVInt(); - return ReadTxnData.SerializerSupport.create(txnId, readScope, executeAtEpoch); + return ReadTxnData.SerializerSupport.create(txnId, scope, executeAtEpoch); } @Override public long serializedSize(ReadTxnData read, int version) { return CommandSerializers.txnId.serializedSize(read.txnId, version) - + KeySerializers.participants.serializedSize(read.readScope, version) + + KeySerializers.participants.serializedSize(read.scope, version) + TypeSizes.sizeofUnsignedVInt(read.executeAtEpoch); } }; @@ -151,7 +158,7 @@ public class ReadDataSerializers public void serialize(ReadEphemeralTxnData read, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(read.txnId, out, version); - KeySerializers.participants.serialize(read.readScope, out, version); + KeySerializers.participants.serialize(read.scope, out, version); out.writeUnsignedVInt(read.executeAtEpoch); CommandSerializers.partialTxn.serialize(read.partialTxn(), out, version); DepsSerializers.partialDeps.serialize(read.partialDeps(), out, version); @@ -162,19 +169,19 @@ public class ReadDataSerializers public ReadEphemeralTxnData deserialize(DataInputPlus in, int version) throws IOException { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); - Participants readScope = KeySerializers.participants.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); long executeAtEpoch = in.readUnsignedVInt(); PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version); PartialDeps partialDeps = DepsSerializers.partialDeps.deserialize(in, version); FullRoute route = KeySerializers.fullRoute.deserialize(in, version); - return ReadEphemeralTxnData.SerializerSupport.create(txnId, readScope, executeAtEpoch, partialTxn, partialDeps, route); + return ReadEphemeralTxnData.SerializerSupport.create(txnId, scope, executeAtEpoch, partialTxn, partialDeps, route); } @Override public long serializedSize(ReadEphemeralTxnData read, int version) { return CommandSerializers.txnId.serializedSize(read.txnId, version) - + KeySerializers.participants.serializedSize(read.readScope, version) + + KeySerializers.participants.serializedSize(read.scope, version) + TypeSizes.sizeofUnsignedVInt(read.executeAtEpoch) + CommandSerializers.partialTxn.serializedSize(read.partialTxn(), version) + DepsSerializers.partialDeps.serializedSize(read.partialDeps(), version) @@ -274,14 +281,14 @@ public class ReadDataSerializers public static final IVersionedSerializer reply = new ReplySerializer<>(TxnData.nullableSerializer); - // TODO (consider): duplicates ReadTxnData ser/de logic; conside deduplicating if another instance of this is added + // TODO (desired): duplicates ReadTxnData ser/de logic; conside deduplicating if another instance of this is added public static final ReadDataSerializer waitUntilApplied = new ReadDataSerializer() { @Override public void serialize(WaitUntilApplied waitUntilApplied, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(waitUntilApplied.txnId, out, version); - KeySerializers.participants.serialize(waitUntilApplied.readScope, out, version); + KeySerializers.participants.serialize(waitUntilApplied.scope, out, version); out.writeUnsignedVInt(waitUntilApplied.minEpoch()); out.writeUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch()); } @@ -290,19 +297,66 @@ public class ReadDataSerializers public WaitUntilApplied deserialize(DataInputPlus in, int version) throws IOException { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); - Participants readScope = KeySerializers.participants.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); long minEpoch = in.readUnsignedVInt(); long executeAtEpoch = minEpoch + in.readUnsignedVInt(); - return WaitUntilApplied.SerializerSupport.create(txnId, readScope, minEpoch, executeAtEpoch); + return WaitUntilApplied.SerializerSupport.create(txnId, scope, minEpoch, executeAtEpoch); } @Override public long serializedSize(WaitUntilApplied waitUntilApplied, int version) { return CommandSerializers.txnId.serializedSize(waitUntilApplied.txnId, version) - + KeySerializers.participants.serializedSize(waitUntilApplied.readScope, version) + + KeySerializers.participants.serializedSize(waitUntilApplied.scope, version) + TypeSizes.sizeofUnsignedVInt(waitUntilApplied.minEpoch()) + TypeSizes.sizeofUnsignedVInt(waitUntilApplied.executeAtEpoch - waitUntilApplied.minEpoch()); } }; + + // TODO (desired): duplicates a lot of Commit serializer + public static final ReadDataSerializer stableThenRead = new ReadDataSerializer<>() + { + @Override + public void serialize(StableThenRead read, DataOutputPlus out, int version) throws IOException + { + CommandSerializers.txnId.serialize(read.txnId, out, version); + KeySerializers.participants.serialize(read.scope, out, version); + CommitSerializers.kind.serialize(read.kind, out, version); + out.writeUnsignedVInt(read.minEpoch); + CommandSerializers.timestamp.serialize(read.executeAt, out, version); + if (read.kind.withTxn != NoTxn) + CommandSerializers.nullablePartialTxn.serialize(read.partialTxn, out, version); + if (read.kind.withDeps == HasDeps) + DepsSerializers.partialDeps.serialize(read.partialDeps, out, version); + if (read.kind.withTxn == HasTxn) + KeySerializers.fullRoute.serialize(read.route, out, version); + } + + @Override + public StableThenRead deserialize(DataInputPlus in, int version) throws IOException + { + TxnId txnId = CommandSerializers.txnId.deserialize(in, version); + Participants scope = KeySerializers.participants.deserialize(in, version); + Commit.Kind kind = CommitSerializers.kind.deserialize(in, version); + long minEpoch = in.readUnsignedVInt(); + Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version); + PartialTxn partialTxn = kind.withTxn == NoTxn ? null : CommandSerializers.nullablePartialTxn.deserialize(in, version); + PartialDeps partialDeps = kind.withDeps == NoDeps ? null : DepsSerializers.partialDeps.deserialize(in, version); + FullRoute < ?> route = kind.withTxn == HasTxn ? KeySerializers.fullRoute.deserialize(in, version) : null; + return StableThenRead.SerializerSupport.create(txnId, scope, kind, minEpoch, executeAt, partialTxn, partialDeps, route); + } + + @Override + public long serializedSize(StableThenRead read, int version) + { + return CommandSerializers.txnId.serializedSize(read.txnId, version) + + KeySerializers.participants.serializedSize(read.scope, version) + + CommitSerializers.kind.serializedSize(read.kind, version) + + TypeSizes.sizeofUnsignedVInt(read.minEpoch) + + CommandSerializers.timestamp.serializedSize(read.executeAt, version) + + (read.kind.withTxn == NoTxn ? 0 : CommandSerializers.nullablePartialTxn.serializedSize(read.partialTxn, version)) + + (read.kind.withDeps != HasDeps ? 0 : DepsSerializers.partialDeps.serializedSize(read.partialDeps, version)) + + (read.kind.withTxn != HasTxn ? 0 : KeySerializers.fullRoute.serializedSize(read.route, version)); + } + }; } 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 3c439b6e96..c4625e7903 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -100,6 +100,8 @@ public class RecoverySerializers latestDeps.serialize(recoverOk.deps, out, version); DepsSerializers.deps.serialize(recoverOk.earlierWait, out, version); DepsSerializers.deps.serialize(recoverOk.earlierNoWait, out, version); + DepsSerializers.deps.serialize(recoverOk.laterWait, out, version); + DepsSerializers.deps.serialize(recoverOk.laterNoWait, out, version); out.writeBoolean(recoverOk.selfAcceptsFastPath); out.writeBoolean(recoverOk.supersedingRejects); CommandSerializers.nullableWrites.serialize(recoverOk.writes, out, version); @@ -118,9 +120,9 @@ public class RecoverySerializers return new RecoverNack(kind, supersededBy); } - RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierWait, Deps earlierNoWait, boolean acceptsFastPath, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) + RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierWait, Deps earlierNoWait, Deps laterWait, Deps laterNoWait, boolean acceptsFastPath, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) { - return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierWait, earlierNoWait, acceptsFastPath, rejectsFastPath, writes, result); + return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierWait, earlierNoWait, laterWait, laterNoWait, acceptsFastPath, rejectsFastPath, writes, result); } @Override @@ -144,6 +146,8 @@ public class RecoverySerializers latestDeps.deserialize(in, version), DepsSerializers.deps.deserialize(in, version), DepsSerializers.deps.deserialize(in, version), + DepsSerializers.deps.deserialize(in, version), + DepsSerializers.deps.deserialize(in, version), in.readBoolean(), in.readBoolean(), CommandSerializers.nullableWrites.deserialize(in, version), @@ -166,6 +170,8 @@ public class RecoverySerializers size += latestDeps.serializedSize(recoverOk.deps, version); size += DepsSerializers.deps.serializedSize(recoverOk.earlierWait, version); size += DepsSerializers.deps.serializedSize(recoverOk.earlierNoWait, version); + size += DepsSerializers.deps.serializedSize(recoverOk.laterWait, version); + size += DepsSerializers.deps.serializedSize(recoverOk.laterNoWait, version); size += TypeSizes.sizeof(recoverOk.selfAcceptsFastPath); size += TypeSizes.sizeof(recoverOk.supersedingRejects); size += CommandSerializers.nullableWrites.serializedSize(recoverOk.writes, version); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java index 865fb77b31..2e91542f1a 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java @@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.Set; import accord.local.Node; import accord.primitives.Range; @@ -137,9 +136,9 @@ public class TopologySerializers { range.serialize(shard.range, out, version); CollectionSerializers.serializeList(shard.nodes, out, version, nodeId); - CollectionSerializers.serializeCollection(shard.fastPathElectorate, out, version, nodeId); - CollectionSerializers.serializeCollection(shard.joining, out, version, nodeId); - + CollectionSerializers.serializeList(shard.notInFastPath, out, version, nodeId); + CollectionSerializers.serializeList(shard.joining, out, version, nodeId); + out.writeBoolean(shard.pendingRemoval); } @Override @@ -147,9 +146,10 @@ public class TopologySerializers { Range range = ShardSerializer.this.range.deserialize(in, version); SortedArrayList nodes = CollectionSerializers.deserializeSortedArrayList(in, version, nodeId, Node.Id[]::new); - Set fastPathElectorate = CollectionSerializers.deserializeSet(in, version, nodeId); - Set joining = CollectionSerializers.deserializeSet(in, version, nodeId); - return new Shard(range, nodes, fastPathElectorate, joining); + SortedArrayList notInFastPath = CollectionSerializers.deserializeSortedArrayList(in, version, nodeId, Node.Id[]::new); + SortedArrayList joining = CollectionSerializers.deserializeSortedArrayList(in, version, nodeId, Node.Id[]::new); + boolean pendingRemoval = in.readBoolean(); + return Shard.SerializerSupport.create(range, nodes, notInFastPath, joining, pendingRemoval); } @Override @@ -157,8 +157,9 @@ public class TopologySerializers { long size = range.serializedSize(shard.range, version); size += CollectionSerializers.serializedListSize(shard.nodes, version, nodeId); - size += CollectionSerializers.serializedCollectionSize(shard.fastPathElectorate, version, nodeId); - size += CollectionSerializers.serializedCollectionSize(shard.joining, version, nodeId); + size += CollectionSerializers.serializedListSize(shard.notInFastPath, version, nodeId); + size += CollectionSerializers.serializedListSize(shard.joining, version, nodeId); + size += TypeSizes.BOOL_SIZE; return size; } }; diff --git a/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java index 89501a2a19..33f68cb65d 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java +++ b/src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java @@ -85,7 +85,7 @@ public class AccordMarkStale implements Transformation // We're trying to mark a node in this shard stale... if (!Collections.disjoint(shard.nodes(), accordIds)) { - int quorumSize = Shard.slowPathQuorumSize(shard.nodes().size()); + int quorumSize = Shard.slowQuorumSize(shard.nodes().size()); Set nonStaleNodes = new HashSet<>(shard.nodes()); nonStaleNodes.removeAll(accordIds); nonStaleNodes.removeAll(prev.accordStaleReplicas.ids()); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordInteroperabilityTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordInteroperabilityTest.java index 7deb529191..f485d4d089 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordInteroperabilityTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordInteroperabilityTest.java @@ -152,7 +152,7 @@ public class AccordInteroperabilityTest extends AccordTestBase cluster -> { SHARED_CLUSTER.setMessageSink(new MessageCountingSink(SHARED_CLUSTER)); cluster.coordinator(1).execute("SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 0", org.apache.cassandra.distributed.api.ConsistencyLevel.SERIAL); - assertEquals(2, messageCounts.get(Verb.ACCORD_INTEROP_COMMIT_REQ).get()); + assertEquals(2, messageCounts.get(Verb.ACCORD_INTEROP_STABLE_THEN_READ_REQ).get()); assertEquals(2, messageCounts.get(Verb.ACCORD_INTEROP_READ_RSP).get()); }); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java index abebaeb49d..ba720be22d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java @@ -110,7 +110,7 @@ public class AccordSimpleFastPathTest extends TestBaseImpl AccordConfigurationService configService = ((AccordService) AccordService.instance()).configurationService(); Topology topology = configService.getTopologyForEpoch(epoch); Assert.assertFalse(topology.shards().isEmpty()); - topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.fastPathElectorate)); + topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath))); return cm.epoch.getEpoch(); })).max(Comparator.naturalOrder()).get(); 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 75b19b5b2d..96ede9a8c8 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java @@ -43,7 +43,7 @@ public class AccordHarrySimulationTest extends HarrySimulatorTest { Set extremelyLossy = new HashSet<>(Arrays.asList(Verb.ACCORD_SIMPLE_RSP, Verb.ACCORD_PRE_ACCEPT_RSP, Verb.ACCORD_PRE_ACCEPT_REQ, - Verb.ACCORD_ACCEPT_RSP, Verb.ACCORD_ACCEPT_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ, + Verb.ACCORD_ACCEPT_RSP, Verb.ACCORD_ACCEPT_REQ, Verb.ACCORD_NOT_ACCEPT_REQ, Verb.ACCORD_READ_RSP, Verb.ACCORD_READ_REQ, Verb.ACCORD_COMMIT_REQ, Verb.ACCORD_COMMIT_INVALIDATE_REQ, Verb.ACCORD_APPLY_RSP, Verb.ACCORD_APPLY_REQ, Verb.ACCORD_BEGIN_RECOVER_RSP, Verb.ACCORD_BEGIN_RECOVER_REQ, Verb.ACCORD_BEGIN_INVALIDATE_RSP)); diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 1a95b5158a..7698fe764e 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -31,6 +31,7 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.api.ProtocolModifiers; import accord.messages.TxnRequest; import accord.primitives.Routable; import accord.primitives.SaveStatus; @@ -52,6 +53,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.concurrent.Condition; import org.awaitility.Awaitility; +import static accord.primitives.TxnId.FastPath.UNOPTIMISED; import static org.apache.cassandra.Util.spinUntilSuccess; import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn; @@ -133,6 +135,8 @@ public class AccordDebugKeyspaceTest extends CQLTester @Test public void blocked() throws ExecutionException, InterruptedException { + ProtocolModifiers.Toggles.setPermitLocalExecution(false); + ProtocolModifiers.Toggles.setPermittedFastPaths(new TxnId.FastPaths(UNOPTIMISED)); AccordMsgFilter filter = new AccordMsgFilter(); MessagingService.instance().outboundSink.add(filter); try @@ -225,15 +229,20 @@ public class AccordDebugKeyspaceTest extends CQLTester preAccept.signalAll(); return true; case ACCORD_COMMIT_REQ: + case ACCORD_STABLE_THEN_READ_REQ: commit.signalAll(); return true; case ACCORD_PRE_ACCEPT_REQ: + case ACCORD_ACCEPT_REQ: + case ACCORD_ACCEPT_RSP: case ACCORD_CHECK_STATUS_REQ: case ACCORD_CHECK_STATUS_RSP: case ACCORD_READ_RSP: case ACCORD_AWAIT_REQ: case ACCORD_AWAIT_RSP: case ACCORD_AWAIT_ASYNC_RSP_REQ: + case ACCORD_FETCH_MIN_EPOCH_REQ: + case ACCORD_FETCH_MIN_EPOCH_RSP: return true; default: // many code paths don't log the error... diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java index a24f2c876d..505b8e8a0d 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java @@ -32,7 +32,6 @@ import org.slf4j.LoggerFactory; import accord.api.Key; import accord.api.Result; import accord.local.Command; -import accord.local.CommonAttributes; import accord.local.StoreParticipants; import accord.local.cfk.CommandsForKey; import accord.primitives.Ballot; @@ -121,24 +120,19 @@ public class AccordCommandStoreTest dependencies = builder.build(); } - CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId); PartialTxn txn = createPartialTxn(0); Route route = RoutingKeys.of(key.toUnseekable()).toRoute(key.toUnseekable()); - attrs.partialTxn(txn); - attrs.setParticipants(StoreParticipants.all(route)); - attrs.durability(Majority); - attrs.partialTxn(txn); Ballot promised = ballot(1, clock.incrementAndGet(), 1); Ballot accepted = ballot(1, clock.incrementAndGet(), 1); Timestamp executeAt = timestamp(1, clock.incrementAndGet(), 1); - attrs.partialDeps(dependencies); SimpleBitSet waitingOnApply = new SimpleBitSet(3); waitingOnApply.set(1); Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps, dependencies.directKeyDeps, new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2)); Pair result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt); - Command expected = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted, - waitingOn, result.left, ResultSerializers.APPLIED); + Command expected = Command.Executed.executed(txnId, SaveStatus.Applied, Majority, StoreParticipants.all(route), + promised, executeAt, txn, dependencies, accepted, + waitingOn, result.left, ResultSerializers.APPLIED); AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null)); safeCommand.set(expected); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index 4084bc1594..9baf5f28c9 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java @@ -34,6 +34,7 @@ import accord.local.Node; import accord.local.PreLoadContext; import accord.local.SafeCommand; import accord.local.SafeCommandStore; +import accord.primitives.KeyDeps; import accord.primitives.Status; import accord.messages.Accept; import accord.messages.Commit; @@ -55,6 +56,8 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.api.PartitionKey; import org.apache.cassandra.utils.ByteBufferUtil; +import static accord.api.ProtocolModifiers.Toggles.filterDuplicateDependenciesFromAcceptReply; +import static accord.messages.Accept.Kind.SLOW; import static accord.utils.async.AsyncChains.getUninterruptibly; import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse; import static org.apache.cassandra.service.accord.AccordTestUtils.createAccordCommandStore; @@ -102,7 +105,7 @@ public class AccordCommandTest FullRoute fullRoute = txn.keys().toRoute(homeKey); Route route = fullRoute.slice(fullRange(txn)); PartialTxn partialTxn = txn.intersecting(route, true); - PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, fullRoute); + PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, null, false, fullRoute); // Check preaccept getUninterruptibly(commandStore.execute(preAccept, safeStore -> { @@ -141,13 +144,13 @@ public class AccordCommandTest builder.add(key.toUnseekable(), txnId2); deps = builder.build(); } - Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, Ballot.ZERO, executeAt, deps); + Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, SLOW, Ballot.ZERO, executeAt, deps); getUninterruptibly(commandStore.execute(accept, safeStore -> { Command before = safeStore.ifInitialised(txnId).current(); Accept.AcceptReply reply = accept.apply(safeStore); Assert.assertTrue(reply.isOk()); - Assert.assertEquals(deps.keyDeps, reply.deps.keyDeps); + Assert.assertEquals(filterDuplicateDependenciesFromAcceptReply() ? KeyDeps.NONE : deps.keyDeps, reply.deps.keyDeps); Command after = safeStore.ifInitialised(txnId).current(); AccordTestUtils.appendCommandsBlocking(commandStore, before, after); })); @@ -155,7 +158,7 @@ public class AccordCommandTest getUninterruptibly(commandStore.execute(accept, safeStore -> { Command before = safeStore.ifInitialised(txnId).current(); Assert.assertEquals(executeAt, before.executeAt()); - Assert.assertEquals(Status.Accepted, before.status()); + Assert.assertEquals(Status.AcceptedSlow, before.status()); Assert.assertEquals(deps, before.partialDeps()); CommandsForKey cfk = safeStore.get(key(1).toUnseekable()).current(); @@ -165,7 +168,7 @@ public class AccordCommandTest })); // check commit - Commit commit = Commit.SerializerSupport.create(txnId, route, 1, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute, null); + Commit commit = Commit.SerializerSupport.create(txnId, route, 1, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute); getUninterruptibly(commandStore.execute(commit, commit::apply)); getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(txnId, Keys.of(key).toParticipants(), KeyHistory.SYNC), safeStore -> { @@ -194,7 +197,7 @@ public class AccordCommandTest FullRoute fullRoute = txn.keys().toRoute(homeKey); Route route = fullRoute.slice(fullRange(txn)); PartialTxn partialTxn = txn.intersecting(route, true); - PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, fullRoute); + PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, null, false, fullRoute); getUninterruptibly(commandStore.execute(preAccept1, safeStore -> { persistDiff(commandStore, safeStore, txnId1, route, () -> { @@ -204,7 +207,7 @@ public class AccordCommandTest // second preaccept should identify txnId1 as a dependency TxnId txnId2 = txnId(1, clock.incrementAndGet(), 1); - PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, 1, partialTxn, fullRoute); + PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, 1, partialTxn, null, false, fullRoute); getUninterruptibly(commandStore.execute(preAccept2, safeStore -> { persistDiff(commandStore, safeStore, txnId2, route, () -> { PreAccept.PreAcceptReply reply = preAccept2.apply(safeStore); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java index 4bec25b4d8..f9ba3b1cfc 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java @@ -143,8 +143,8 @@ public class AccordFastPathCoordinatorTest public void simpleAlive() { Topology topology = new Topology(1, - new Shard(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), - new Shard(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); + Shard.create(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), + Shard.create(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); InstrumentedFastPathCoordinator coordinator = new InstrumentedFastPathCoordinator(id(0)); coordinator.updatePeers(topology); @@ -174,8 +174,8 @@ public class AccordFastPathCoordinatorTest public void simpleDead() { Topology topology = new Topology(1, - new Shard(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), - new Shard(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); + Shard.create(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), + Shard.create(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); InstrumentedFastPathCoordinator coordinator = new InstrumentedFastPathCoordinator(id(0)); coordinator.updatePeers(topology); Assert.assertTrue(coordinator.capturedUpdates.isEmpty()); @@ -236,8 +236,8 @@ public class AccordFastPathCoordinatorTest public void peerShutdownTest() { Topology topology = new Topology(1, - new Shard(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), - new Shard(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); + Shard.create(AccordTopology.minRange(TABLE_1, token(0)), idList(0, 1, 2), idSet(0, 1, 2)), + Shard.create(AccordTopology.maxRange(TABLE_1, token(0)), idList(3, 4, 5), idSet(3, 4, 5))); InstrumentedFastPathCoordinator coordinator = new InstrumentedFastPathCoordinator(id(0)); coordinator.currentMetadata(EMPTY.transformer().withFastPathStatusSince(id(1), Status.SHUTDOWN, 1, 1).build().metadata); coordinator.updatePeers(topology); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java index 3c655133fa..45ff263646 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java @@ -28,8 +28,10 @@ import org.junit.Test; import accord.api.Journal; import accord.local.Command; -import accord.local.CommonAttributes; +import accord.local.StoreParticipants; import accord.primitives.Ballot; +import accord.primitives.SaveStatus; +import accord.primitives.Status; import accord.primitives.TxnId; import accord.utils.AccordGens; import accord.utils.RandomSource; @@ -80,7 +82,7 @@ public class AccordJournalOrderTest TxnId txnId = randomSource.nextBoolean() ? id1 : id2; JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, randomSource.nextInt(5)); res.compute(key, (k, prev) -> prev == null ? 1 : prev + 1); - Command command = Command.NotDefined.notDefined(new CommonAttributes.Mutable(txnId), Ballot.ZERO); + Command command = Command.NotDefined.notDefined(txnId, SaveStatus.NotDefined, Status.Durability.NotDurable, StoreParticipants.empty(txnId), Ballot.ZERO); accordJournal.saveCommand(key.commandStoreId, new Journal.CommandUpdate(null, command), () -> {}); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java index 3c17fef8b4..87388b877a 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java @@ -33,7 +33,6 @@ import org.junit.Test; import accord.api.RoutingKey; import accord.local.Command; -import accord.local.CommonAttributes; import accord.local.Node; import accord.local.StoreParticipants; import accord.primitives.Ballot; @@ -71,6 +70,7 @@ import org.assertj.core.api.Assertions; import org.mockito.Mockito; import org.mockito.stubbing.Answer; +import static accord.local.Command.Committed.committed; import static accord.utils.Property.qt; import static org.apache.cassandra.config.DatabaseDescriptor.setSelectedSSTableFormat; import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS; @@ -110,14 +110,11 @@ public class AccordKeyspaceTest extends CQLTester.InMemory StoreParticipants participants = StoreParticipants.all(route); Deps deps = new Deps(KeyDeps.none(((Keys) txn.keys()).toParticipants()), RangeDeps.NONE, KeyDeps.NONE); - CommonAttributes.Mutable common = new CommonAttributes.Mutable(id); - common.partialTxn(partialTxn); - common.setParticipants(participants); - common.partialDeps(deps.intersecting(scope)); - common.durability(Status.Durability.NotDurable); Command.WaitingOn waitingOn = null; - Command.Committed committed = Command.SerializerSupport.committed(common, SaveStatus.Committed, id, Ballot.ZERO, Ballot.ZERO, waitingOn); + Command.Committed committed = committed(id, SaveStatus.Committed, Status.Durability.NotDurable, + participants, Ballot.ZERO, id, partialTxn, deps.intersecting(scope), + Ballot.ZERO, waitingOn); AccordSafeCommand safeCommand = new AccordSafeCommand(AccordTestUtils.loaded(id, null)); safeCommand.set(committed); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 3f5b59f0dc..a516da4d8d 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -46,7 +46,6 @@ import accord.impl.DefaultLocalListeners.NotifySink.NoOpNotifySink; import accord.local.Command; import accord.local.CommandStore; import accord.local.CommandStores; -import accord.local.CommonAttributes; import accord.local.DurableBefore; import accord.local.Node; import accord.local.Node.Id; @@ -66,13 +65,13 @@ import accord.primitives.Routable; import accord.primitives.SaveStatus; import accord.primitives.Seekable; import accord.primitives.Seekables; -import accord.primitives.Status; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; import accord.primitives.Writes; import accord.topology.Shard; import accord.topology.Topology; +import accord.topology.TopologyManager; import accord.utils.SortedArrays.SortedArrayList; import accord.utils.async.AsyncChains; import org.apache.cassandra.ServerTestUtils; @@ -106,6 +105,10 @@ import org.apache.cassandra.utils.concurrent.Condition; import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static accord.primitives.Routable.Domain.Key; +import static accord.primitives.SaveStatus.NotDefined; +import static accord.primitives.SaveStatus.PreAccepted; +import static accord.primitives.Status.Durability.NotDurable; +import static accord.primitives.Txn.Kind.Write; import static accord.utils.async.AsyncChains.getUninterruptibly; import static java.lang.String.format; import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_LOCK; @@ -120,44 +123,24 @@ public class AccordTestUtils { public static Command notDefined(TxnId txnId, PartialTxn txn) { - CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId); - attrs.partialTxn(txn); - return Command.SerializerSupport.notDefined(attrs, Ballot.ZERO); + return Command.NotDefined.notDefined(txnId, NotDefined, NotDurable, StoreParticipants.empty(txnId), Ballot.ZERO); } public static Command preaccepted(TxnId txnId, PartialTxn txn, Timestamp executeAt) { - CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId); - attrs.partialTxn(txn); - attrs.setParticipants(StoreParticipants.all(route(txn))); - attrs.durability(Status.Durability.NotDurable); - return Command.SerializerSupport.preaccepted(attrs, executeAt, Ballot.ZERO); + return Command.PreAccepted.preaccepted(txnId, PreAccepted, NotDurable, StoreParticipants.all(route(txn)), Ballot.ZERO, executeAt, txn, null); } public static Command committed(TxnId txnId, PartialTxn txn, Timestamp executeAt) { - CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId).partialDeps(PartialDeps.NONE); - attrs.partialTxn(txn); - attrs.setParticipants(StoreParticipants.all(route(txn))); - return Command.SerializerSupport.committed(attrs, - SaveStatus.Committed, - executeAt, - Ballot.ZERO, - Ballot.ZERO, - null); + return Command.Committed.committed(txnId, SaveStatus.Committed, NotDurable, StoreParticipants.all(route(txn)), + Ballot.ZERO, executeAt, txn, PartialDeps.NONE, 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.setParticipants(StoreParticipants.all(route(txn))); - return Command.SerializerSupport.committed(attrs, - SaveStatus.Stable, - executeAt, - Ballot.ZERO, - Ballot.ZERO, - Command.WaitingOn.empty(txnId.domain())); + return Command.Committed.committed(txnId, SaveStatus.Stable, NotDurable, StoreParticipants.all(route(txn)), + Ballot.ZERO, executeAt, txn, PartialDeps.NONE, Ballot.ZERO, Command.WaitingOn.empty(txnId.domain())); } private static FullRoute route(PartialTxn txn) @@ -202,7 +185,7 @@ public class AccordTestUtils public static TxnId txnId(long epoch, long hlc, int node) { - return txnId(epoch, hlc, node, Txn.Kind.Write); + return txnId(epoch, hlc, node, Write); } public static TxnId txnId(long epoch, long hlc, int node, Txn.Kind kind) @@ -251,7 +234,7 @@ public class AccordTestUtils } }) .reduce(null, TxnData::merge); - return Pair.create(txn.execute(txnId, executeAt, readData), + return Pair.create(txnId.is(Write) ? txn.execute(txnId, executeAt, readData) : null, txn.query().compute(txnId, executeAt, txn.keys(), readData, txn.read(), txn.update())); } @@ -377,8 +360,8 @@ public class AccordTestUtils @Override public long now() {return now.getAsLong(); } @Override public Timestamp uniqueNow() { return uniqueNow(Timestamp.NONE); } @Override public Timestamp uniqueNow(Timestamp atLeast) { return Timestamp.fromValues(1, now.getAsLong(), node); } - @Override - public long elapsed(TimeUnit timeUnit) { return elapsed.applyAsLong(timeUnit); } + @Override public long elapsed(TimeUnit timeUnit) { return elapsed.applyAsLong(timeUnit); } + @Override public TopologyManager topology() { throw new UnsupportedOperationException(); } }; @@ -410,7 +393,7 @@ public class AccordTestUtils TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table); TokenRange range = TokenRange.fullRange(metadata.id); Node.Id node = new Id(1); - Topology topology = new Topology(1, new Shard(range, new SortedArrayList<>(new Id[] { node }), Sets.newHashSet(node), Collections.emptySet())); + Topology topology = new Topology(1, Shard.create(range, new SortedArrayList<>(new Id[] { node }), Sets.newHashSet(node), Collections.emptySet())); AccordCommandStore store = createAccordCommandStore(node, now, topology, loadExecutor, saveExecutor); store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).executor().cacheUnsafe().setCapacity(1 << 20)); return store; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTopologyTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTopologyTest.java index 5c895f1ff5..fd28cafb68 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTopologyTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTopologyTest.java @@ -84,7 +84,7 @@ public class AccordTopologyTest Topology topology = AccordTopology.createAccordTopology(metadata); Topology expected = new Topology(1, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); Assert.assertEquals(expected, topology); } @@ -99,7 +99,7 @@ public class AccordTopologyTest ClusterMetadata metadata = configureCluster(ranges, Keyspaces.of(keyspace)); Topology topology = AccordTopology.createAccordTopology(metadata); Topology expected = new Topology(1, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); Assert.assertEquals(expected, topology); } @@ -113,7 +113,7 @@ public class AccordTopologyTest ClusterMetadata metadata = configureCluster(ranges, Keyspaces.of(keyspace)); Topology topology = AccordTopology.createAccordTopology(metadata); Topology expected = new Topology(1, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); Assert.assertEquals(expected, topology); topology = AccordTopology.createAccordTopology(metadata.transformer().withFastPathStatusSince(new Id(1), AccordFastPath.Status.UNAVAILABLE, 1, 1).build().metadata); @@ -122,7 +122,7 @@ public class AccordTopologyTest fastPath.remove(new Node.Id(1)); expected = new Topology(2, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, fastPath)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, fastPath)); Assert.assertEquals(expected, topology); } @@ -138,7 +138,7 @@ public class AccordTopologyTest ClusterMetadata metadata = configureCluster(ranges, Keyspaces.of(keyspace)); Topology topology = AccordTopology.createAccordTopology(metadata); Topology expected = new Topology(1, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, NODE_SET)); Assert.assertEquals(expected, topology); metadata = metadata.transformer() @@ -151,7 +151,7 @@ public class AccordTopologyTest fastPath.remove(new Node.Id(1)); expected = new Topology(2, - new Shard(AccordTopology.fullRange(tableId), NODE_LIST, fastPath)); + Shard.create(AccordTopology.fullRange(tableId), NODE_LIST, fastPath)); Assert.assertEquals(expected, topology); } } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 786429bc6c..17117222ae 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -69,6 +69,7 @@ import accord.primitives.TxnId; import accord.primitives.Unseekables; import accord.topology.Topologies; import accord.topology.Topology; +import accord.topology.TopologyManager; import accord.utils.Gens; import accord.utils.RandomSource; import accord.utils.async.AsyncChains; @@ -184,6 +185,12 @@ public class SimulatedAccordCommandStore implements AutoCloseable throw new UnsupportedOperationException(); return now; } + + @Override + public TopologyManager topology() + { + throw new UnsupportedOperationException(); + } }; TestAgent.RethrowAgent agent = new TestAgent.RethrowAgent() @@ -260,7 +267,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable public TxnId nextTxnId(Txn.Kind kind, Routable.Domain domain) { - return new TxnId(storeService.epoch(), storeService.now(), kind, domain, nodeId); + return new TxnId(storeService.epoch(), storeService.now(), 0, kind, domain, nodeId); } public void maybeCacheEvict(Unseekables keysOrRanges) @@ -288,7 +295,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable if (TxnId.class.equals(keyType)) { Command command = (Command) state.getExclusive(); - if (command != null && command.known().definition.isKnown() + if (command != null && command.known().isDefinitionKnown() && (command.partialTxn().keys().intersects(keys) || ranges.intersects(command.partialTxn().keys())) && shouldEvict.getAsBoolean()) cache.tryEvict(state); @@ -393,7 +400,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable public Pair> enqueuePreAccept(Txn txn, FullRoute route) { TxnId txnId = nextTxnId(txn.kind(), txn.keys().domain()); - PreAccept preAccept = new PreAccept(nodeId, topologies, txnId, txn, route); + PreAccept preAccept = new PreAccept(nodeId, topologies, txnId, txn, null, false, route); return Pair.create(txnId, processAsync(preAccept, safe -> { var reply = preAccept.apply(safe); Assertions.assertThat(reply.isOk()).isTrue(); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index 091f01dd23..4b35621570 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -275,7 +275,7 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester .collect(Collectors.toMap(e -> e.getKey(), e -> new ArrayList(e.getValue()))); TxnId txnId = instance.nextTxnId(txn.kind(), txn.keys().domain()); - PreAccept preAccept = new PreAccept(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, txn, route); + PreAccept preAccept = new PreAccept(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, txn, null, false, route); var preAcceptAsync = instance.processAsync(preAccept, safe -> { var reply = preAccept.apply(safe); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java index c8d3f82052..953df9c206 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java @@ -118,7 +118,7 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase Action action = actionGen.next(rs); TxnId txnId = instance.nextTxnId(txn.kind(), txn.keys().domain()); FullRoute route = txnWithRoute.right; - PreAccept preAccept = new PreAccept(nodeId, instance.topologies, txnId, txn, route) { + PreAccept preAccept = new PreAccept(nodeId, instance.topologies, txnId, txn, null, false, route) { @Override public PreAcceptReply apply(SafeCommandStore safeStore) { 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 7d9ab80872..91d3a8b232 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -45,8 +45,7 @@ import org.junit.Test; import accord.api.Key; import accord.api.RoutingKey; import accord.local.Command; -import accord.local.CommonAttributes; -import accord.local.CommonAttributes.Mutable; +import accord.local.ICommand; import accord.local.Node; import accord.local.StoreParticipants; import accord.local.cfk.CommandsForKey; @@ -56,6 +55,7 @@ import accord.local.cfk.CommandsForKey.Unmanaged; import accord.local.cfk.Serialize; import accord.primitives.Ballot; import accord.primitives.KeyDeps; +import accord.primitives.Known; import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.RangeDeps; @@ -71,6 +71,7 @@ import accord.utils.Gen; import accord.utils.Gens; import accord.utils.RandomSource; import accord.utils.SortedArrays; +import accord.utils.UnhandledEnum; import org.agrona.collections.Int2ObjectHashMap; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.dht.Murmur3Partitioner; @@ -143,25 +144,38 @@ public class CommandsForKeySerializerTest this.isDurable = isDurable; } - CommonAttributes attributes() + ICommand.Builder builder() { - Mutable mutable = new Mutable(txnId); + ICommand.Builder builder = new ICommand.Builder(txnId); if (saveStatus.known.isDefinitionKnown()) - mutable.partialTxn(txn); + builder.partialTxn(txn); - mutable.setParticipants(StoreParticipants.all(txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null)))); - mutable.durability(isDurable ? Majority : NotDurable); - if (saveStatus.known.deps.hasProposedOrDecidedDeps()) + builder.setParticipants(StoreParticipants.all(txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null)))); + builder.durability(isDurable ? Majority : NotDurable); + if (saveStatus.known.deps().hasPreAcceptedOrProposedOrDecidedDeps()) { - try (KeyDeps.Builder builder = KeyDeps.builder();) + try (KeyDeps.Builder keyBuilder = KeyDeps.builder();) { for (TxnId id : deps) - builder.add(((Key)txn.keys().get(0)).toUnseekable(), id); - mutable.partialDeps(new PartialDeps(AccordTestUtils.fullRange(txn), builder.build(), RangeDeps.NONE, KeyDeps.NONE)); + keyBuilder.add(((Key)txn.keys().get(0)).toUnseekable(), id); + builder.partialDeps(new PartialDeps(AccordTestUtils.fullRange(txn), keyBuilder.build(), RangeDeps.NONE, KeyDeps.NONE)); } } - return mutable; + builder.executeAt(executeAt); + builder.promised(ballot); + builder.acceptedOrCommitted(ballot); + builder.durability(isDurable ? Majority : NotDurable); + if (saveStatus.compareTo(SaveStatus.Stable) >= 0 && !saveStatus.hasBeen(Status.Truncated)) + builder.waitingOn(Command.WaitingOn.empty(txnId.domain())); + + if (saveStatus.known.outcome() == Known.Outcome.Apply) + { + if (txnId.is(Txn.Kind.Write)) + builder.writes(new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true))); + builder.result(new TxnData()); + } + return builder; } Command toCommand() @@ -171,34 +185,50 @@ public class CommandsForKeySerializerTest default: throw new AssertionError("Unhandled saveStatus: " + saveStatus); case Uninitialised: case NotDefined: - return Command.SerializerSupport.notDefined(attributes(), Ballot.ZERO); + return Command.NotDefined.notDefined(builder(), Ballot.ZERO); case PreAccepted: - return Command.SerializerSupport.preaccepted(attributes(), executeAt, Ballot.ZERO); + case PreAcceptedWithVote: + case PreAcceptedWithDeps: + return Command.PreAccepted.preaccepted(builder(), saveStatus); case AcceptedInvalidate: - return Command.SerializerSupport.acceptedInvalidateWithoutDefinition(attributes(), ballot, ballot); - case Accepted: - case AcceptedWithDefinition: + case PreNotAccepted: + case NotAccepted: + return Command.NotAcceptedWithoutDefinition.notAccepted(builder(), saveStatus); + case AcceptedMedium: + case AcceptedMediumWithDefinition: + case AcceptedMediumWithDefAndVote: + case AcceptedSlow: + case AcceptedSlowWithDefinition: + case AcceptedSlowWithDefAndVote: case AcceptedInvalidateWithDefinition: + case PreNotAcceptedWithDefinition: + case PreNotAcceptedWithDefAndVote: + case PreNotAcceptedWithDefAndDeps: + case NotAcceptedWithDefinition: + case NotAcceptedWithDefAndVote: + case NotAcceptedWithDefAndDeps: case PreCommittedWithDefinition: - case PreCommittedWithDefinitionAndAcceptedDeps: - case PreCommittedWithAcceptedDeps: + case PreCommittedWithDefAndDeps: + case PreCommittedWithDefAndFixedDeps: + case PreCommittedWithDeps: + case PreCommittedWithFixedDeps: case PreCommitted: - return Command.SerializerSupport.accepted(attributes(), saveStatus, executeAt, ballot, ballot); + return Command.Accepted.accepted(builder(), saveStatus); case Committed: - return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, null); + return Command.Committed.committed(builder(), saveStatus); case Stable: case ReadyToExecute: - return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.empty(txnId.domain())); + return Command.Committed.committed(builder(), saveStatus); case PreApplied: case Applying: case Applied: - return Command.SerializerSupport.executed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.empty(txnId.domain()), new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true)), new TxnData()); + return Command.Executed.executed(builder(), saveStatus); case Invalidated: - return Command.SerializerSupport.invalidated(txnId, attributes().participants()); + return Command.Truncated.invalidated(txnId, builder().participants()); } } @@ -242,14 +272,14 @@ public class CommandsForKeySerializerTest TxnId txnId = txnIdSupplier.get(); SaveStatus saveStatus = saveStatusSupplier.get(); Timestamp executeAt = txnId; - if (!txnId.kind().awaitsOnlyDeps() && saveStatus.known.executeAt != ExecuteAtErased && saveStatus.known.executeAt != ExecuteAtUnknown) + if (!txnId.kind().awaitsOnlyDeps() && !saveStatus.known.is(ExecuteAtErased) && !saveStatus.known.is(ExecuteAtUnknown)) executeAt = timestampSupplier.apply(txnId); boolean isDurable = false; Ballot ballot; switch (saveStatus.status) { - default: throw new AssertionError(); + default: throw new UnhandledEnum(saveStatus.status); case NotDefined: case PreAccepted: case Invalidated: @@ -259,8 +289,11 @@ public class CommandsForKeySerializerTest case PreApplied: case Applied: isDurable = source.nextBoolean(); + case PreNotAccepted: + case NotAccepted: case AcceptedInvalidate: - case Accepted: + case AcceptedMedium: + case AcceptedSlow: case PreCommitted: case Committed: case Stable: @@ -272,10 +305,10 @@ public class CommandsForKeySerializerTest Arrays.sort(cmds, Comparator.comparing(o -> o.txnId)); for (int i = 0 ; i < txnIdCount ; ++i) { - if (!cmds[i].saveStatus.known.deps.hasProposedOrDecidedDeps()) + if (!cmds[i].saveStatus.known.deps().hasProposedOrDecidedDeps()) continue; - Timestamp knownBefore = cmds[i].saveStatus.known.deps.hasCommittedOrDecidedDeps() ? cmds[i].executeAt : cmds[i].txnId; + Timestamp knownBefore = cmds[i].saveStatus.known.deps().hasCommittedOrDecidedDeps() ? cmds[i].executeAt : cmds[i].txnId; int limit = SortedArrays.binarySearch(cmds, 0, cmds.length, knownBefore, (a, b) -> a.compareTo(b.txnId), FAST); if (limit < 0) limit = -1 - limit; @@ -573,7 +606,7 @@ public class CommandsForKeySerializerTest TokenKey pk = new TokenKey(TableId.fromString("1b255f4d-ef25-40a6-0000-000000000009"), token); TxnId txnId = TxnId.fromValues(11,34052499,2,1); CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, - new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, true, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) }, + new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_WITHOUT_DEPS, true, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) }, 0, CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, NO_BOUNDS_INFO); ByteBuffer buffer = Serialize.toBytesWithoutKey(expected); diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index c551d4a023..026d2c26f8 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -33,7 +33,7 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSortedMap; import accord.local.Command; -import accord.local.CommonAttributes; +import accord.local.ICommand; import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.local.StoreParticipants; @@ -49,6 +49,7 @@ import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.SaveStatus; import accord.primitives.Seekables; +import accord.primitives.Status; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; @@ -75,6 +76,7 @@ import org.quicktheories.impl.JavaRandom; import static accord.local.CommandStores.RangesForEpoch; import static accord.primitives.Status.Durability.NotDurable; +import static accord.primitives.Txn.Kind.Write; import static org.apache.cassandra.service.accord.AccordTestUtils.TABLE_ID1; import static org.apache.cassandra.service.accord.AccordTestUtils.createPartialTxn; @@ -212,18 +214,30 @@ public class AccordGenerators this.keysOrRanges = txn.keys(); } - private CommonAttributes attributes(SaveStatus saveStatus) + private ICommand attributes(SaveStatus saveStatus) { - CommonAttributes.Mutable mutable = new CommonAttributes.Mutable(txnId); + ICommand.Builder builder = new ICommand.Builder(txnId); if (saveStatus.known.isDefinitionKnown()) - mutable.partialTxn(partialTxn); - if (saveStatus.known.deps.hasProposedOrDecidedDeps()) - mutable.partialDeps(partialDeps); + builder.partialTxn(partialTxn); + if (saveStatus.known.deps().hasPreAcceptedOrProposedOrDecidedDeps()) + builder.partialDeps(partialDeps); - mutable.setParticipants(StoreParticipants.all(route)); - mutable.durability(NotDurable); - - return mutable; + builder.setParticipants(StoreParticipants.all(route)); + builder.durability(NotDurable); + if (saveStatus.compareTo(SaveStatus.PreAccepted) >= 0) + builder.executeAt(executeAt); + builder.promised(promised); + if (saveStatus.status.compareTo(Status.PreAccepted) > 0) + builder.acceptedOrCommitted(accepted); + if (saveStatus.compareTo(SaveStatus.Stable) >= 0 && !saveStatus.hasBeen(Status.Truncated)) + builder.waitingOn(waitingOn); + if (saveStatus.hasBeen(Status.PreApplied) && !saveStatus.hasBeen(Status.Truncated)) + { + if (txnId.is(Write)) + builder.writes(new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true))); + builder.result(new TxnData()); + } + return builder; } public Command build(SaveStatus saveStatus) @@ -235,45 +249,61 @@ public class AccordGenerators throw new IllegalArgumentException("TruncatedApplyWithDeps is not a valid state for a Command to be in, its for FetchData"); case Uninitialised: case NotDefined: - return Command.SerializerSupport.notDefined(attributes(saveStatus), Ballot.ZERO); + return Command.NotDefined.notDefined(attributes(saveStatus), Ballot.ZERO); case PreAccepted: - return Command.SerializerSupport.preaccepted(attributes(saveStatus), executeAt, Ballot.ZERO); + case PreAcceptedWithVote: + case PreAcceptedWithDeps: + return Command.PreAccepted.preaccepted(attributes(saveStatus), saveStatus); + case PreNotAccepted: + case PreNotAcceptedWithDefinition: + case PreNotAcceptedWithDefAndDeps: + case PreNotAcceptedWithDefAndVote: + case NotAccepted: + case NotAcceptedWithDefinition: + case NotAcceptedWithDefAndDeps: + case NotAcceptedWithDefAndVote: case AcceptedInvalidate: - return Command.AcceptedInvalidateWithoutDefinition.acceptedInvalidate(attributes(saveStatus), promised, Ballot.ZERO); + return Command.NotAcceptedWithoutDefinition.acceptedInvalidate(attributes(saveStatus)); - case Accepted: - case AcceptedWithDefinition: + case AcceptedMedium: + case AcceptedMediumWithDefinition: + case AcceptedMediumWithDefAndVote: case AcceptedInvalidateWithDefinition: + case AcceptedSlow: + case AcceptedSlowWithDefinition: + case AcceptedSlowWithDefAndVote: case PreCommittedWithDefinition: - case PreCommittedWithDefinitionAndAcceptedDeps: - case PreCommittedWithAcceptedDeps: + case PreCommittedWithDeps: + case PreCommittedWithFixedDeps: + case PreCommittedWithDefAndDeps: + case PreCommittedWithDefAndFixedDeps: case PreCommitted: - return Command.SerializerSupport.accepted(attributes(saveStatus), saveStatus, executeAt, promised, accepted); + return Command.Accepted.accepted(attributes(saveStatus), saveStatus); case Committed: - return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, null); + return Command.Committed.committed(attributes(saveStatus), saveStatus); case Stable: case ReadyToExecute: - return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, waitingOn); + return Command.Committed.committed(attributes(saveStatus), saveStatus); case PreApplied: case Applying: case Applied: - return Command.SerializerSupport.executed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, waitingOn, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData()); + return Command.Executed.executed(attributes(saveStatus), saveStatus); case TruncatedApply: - if (txnId.kind().awaitsOnlyDeps()) return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null, txnId); - else return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null); + if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null, txnId); + else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null); case TruncatedApplyWithOutcome: - if (txnId.kind().awaitsOnlyDeps()) return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData(), txnId); - else return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData()); + if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges,new TxnWrite(Collections.emptyList(), true)) : null, new TxnData(), txnId); + else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)) : null, new TxnData()); case Erased: case ErasedOrVestigial: case Invalidated: - return Command.SerializerSupport.invalidated(txnId, attributes(saveStatus).participants()); + return Command.Truncated.invalidated(txnId, attributes(saveStatus).participants()); } } }