You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/01/18 00:02:07 UTC
(cassandra) 01/01: 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 siz [...]
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 954a6e1af89bfc6b556f89e6f34c8013cd11645e
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Jan 8 09:58:12 2025 +0000
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 +-
.../cassandra/service/accord/AccordJournal.java | 4 +-
.../service/accord/AccordMessageSink.java | 29 ++--
.../service/accord/AccordObjectSizes.java | 79 ++++++---
.../cassandra/service/accord/AccordService.java | 9 +-
.../cassandra/service/accord/AccordTopology.java | 17 +-
.../service/accord/CommandsForRanges.java | 4 +-
.../cassandra/service/accord/api/AccordAgent.java | 9 +
.../fastpath/ParameterizedFastPathStrategy.java | 8 +-
.../accord/fastpath/SimpleFastPathStrategy.java | 2 +-
.../service/accord/interop/AccordInteropApply.java | 15 +-
.../accord/interop/AccordInteropCommit.java | 74 ---------
.../accord/interop/AccordInteropExecution.java | 15 +-
.../accord/interop/AccordInteropPersist.java | 20 ++-
.../service/accord/interop/AccordInteropRead.java | 20 +--
.../accord/interop/AccordInteropReadCallback.java | 3 +-
.../accord/interop/AccordInteropReadRepair.java | 16 +-
.../interop/AccordInteropStableThenRead.java | 181 +++++++++++++++++++++
.../accord/serializers/AcceptSerializers.java | 34 ++--
.../accord/serializers/ApplySerializers.java | 10 +-
.../accord/serializers/AwaitSerializer.java | 8 +-
.../accord/serializers/CheckStatusSerializers.java | 2 +-
.../accord/serializers/CommandSerializers.java | 18 +-
.../accord/serializers/CommitSerializers.java | 46 ++----
.../accord/serializers/FetchSerializers.java | 4 +-
.../accord/serializers/PreacceptSerializers.java | 36 ++--
.../accord/serializers/ReadDataSerializers.java | 90 ++++++++--
.../accord/serializers/RecoverySerializers.java | 10 +-
.../accord/serializers/TopologySerializers.java | 19 ++-
.../tcm/transformations/AccordMarkStale.java | 2 +-
.../test/accord/AccordInteroperabilityTest.java | 2 +-
.../test/accord/AccordSimpleFastPathTest.java | 2 +-
.../simulator/test/AccordHarrySimulationTest.java | 2 +-
.../db/virtual/AccordDebugKeyspaceTest.java | 9 +
.../service/accord/AccordCommandStoreTest.java | 12 +-
.../service/accord/AccordCommandTest.java | 17 +-
.../accord/AccordFastPathCoordinatorTest.java | 12 +-
.../service/accord/AccordJournalOrderTest.java | 6 +-
.../service/accord/AccordKeyspaceTest.java | 11 +-
.../cassandra/service/accord/AccordTestUtils.java | 49 ++----
.../service/accord/AccordTopologyTest.java | 12 +-
.../accord/SimulatedAccordCommandStore.java | 13 +-
.../SimulatedAccordCommandStoreTestBase.java | 2 +-
.../service/accord/SimulatedAccordTaskTest.java | 2 +-
.../serializers/CommandsForKeySerializerTest.java | 93 +++++++----
.../apache/cassandra/utils/AccordGenerators.java | 84 +++++++---
48 files changed, 698 insertions(+), 425 deletions(-)
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 90b583235c..c740e69db9 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 c7e40c03de..1847231299 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;
@@ -314,7 +314,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 ),
@@ -329,6 +329,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 ),
@@ -354,7 +355,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<MessageType> values;
@@ -122,7 +120,7 @@ public class AccordMessageSink implements MessageSink
private final Map<Verb, Set<Verb>> overrideReplyVerbs = ImmutableMap.<Verb, Set<Verb>>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());
+ }
- 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() {};
+ return builder;
+ }
- 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 7dcddf8629..f101d67e47 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<accord.primitives.Range, Shard>
{
- private Shard createOrReuse(boolean pendingRemoval, accord.primitives.Range range, SortedArrayList<Id> nodes, Set<Id> fastPathElectorate, Set<Id> joining)
+ private Shard createOrReuse(boolean pendingRemoval, accord.primitives.Range range, SortedArrayList<Id> nodes, SortedArrayList<Id> fastPath, Set<Id> 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<Id> fastPath = strategyFor(metadata).calculateFastPath(nodes, unavailable, dcMap);
+ SortedArrayList<Id> electorate = strategyFor(metadata).calculateFastPath(nodes, unavailable, dcMap);
List<Shard> 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 29cdcb5195..c803852522 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<Timestamp, Summary> implements CommandSummaries.Snapshot
@@ -188,7 +188,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> 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<TxnId, Command> 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<fpSize; i++)
array[i] = sorters.get(i).id;
Arrays.sort(array);
- SortedArrayList<Node.Id> fastPath = new SortedArrayList<>(array);
- Invariants.checkState(fastPath.size() >= slowQuorum);
- return fastPath;
+ SortedArrayList<Node.Id> 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<Node.Id> 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<AccordInteropCommit> 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..5223024426 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;
@@ -50,6 +49,7 @@ import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
+import accord.primitives.Writes;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
@@ -99,6 +99,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 +235,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 +383,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 +393,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<Data> result;
@@ -406,7 +405,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 +419,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<? super Result, Throwable> 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<AccordInteropRead> requestSerializer = new ReadDataSerializer<AccordInteropRead>()
+ public static final IVersionedSerializer<AccordInteropRead> 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<T> implements Callback<ReadReply
wrapped.onFailure(endpoint, requestFailure);
}
- public void onCallbackFailure(Node.Id from, Throwable failure)
+ public boolean onCallbackFailure(Node.Id from, Throwable failure)
{
wrapped.onFailure(endpoint, RequestFailure.UNKNOWN);
+ return true;
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
index c714c7857d..fdf8e3daf3 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
@@ -65,7 +65,7 @@ public class AccordInteropReadRepair extends ReadData
public void serialize(AccordInteropReadRepair repair, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(repair.txnId, out, version);
- KeySerializers.participants.serialize(repair.readScope, out, version);
+ KeySerializers.participants.serialize(repair.scope, out, version);
out.writeUnsignedVInt(repair.executeAtEpoch);
Mutation.serializer.serialize(repair.mutation, out, version);
}
@@ -74,17 +74,17 @@ public class AccordInteropReadRepair extends ReadData
public AccordInteropReadRepair 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();
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<ReadReply> 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<AccordInteropStableThenRead> 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<Accept> request = new TxnRequestSerializer.WithUnsyncedSerializer<>()
{
+ final IVersionedSerializer<Accept.Kind> 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<Accept.Invalidate> invalidate = new IVersionedSerializer<Accept.Invalidate>()
+ public static final IVersionedSerializer<Accept.NotAccept> 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..90ae01a41e 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
@@ -27,6 +27,7 @@ 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.primitives.Writes;
import accord.utils.Invariants;
@@ -35,6 +36,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 +71,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 +87,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 +100,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<Commit.Kind> kind = new EnumSerializer<>(Commit.Kind.class);
+ public static final IVersionedSerializer<Commit.Kind> kind = new EnumSerializer<>(Commit.Kind.class);
- public abstract static class CommitSerializer<C extends Commit, R extends ReadData> extends TxnRequestSerializer.WithUnsyncedSerializer<C>
+ public static final CommitSerializer request = new CommitSerializer();
+ public static class CommitSerializer extends TxnRequestSerializer.WithUnsyncedSerializer<Commit>
{
- private final IVersionedSerializer<ReadData> read;
-
- public CommitSerializer(Class<R> klass, IVersionedSerializer<R> 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<Commit> request = new CommitSerializer<Commit, ReadData>(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<Commit.Invalidate> 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<ReadReply> 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> waitUntilApplied = new ReadDataSerializer<WaitUntilApplied>()
{
@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> 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<Node.Id> nodes = CollectionSerializers.deserializeSortedArrayList(in, version, nodeId, Node.Id[]::new);
- Set<Node.Id> fastPathElectorate = CollectionSerializers.deserializeSet(in, version, nodeId);
- Set<Node.Id> joining = CollectionSerializers.deserializeSet(in, version, nodeId);
- return new Shard(range, nodes, fastPathElectorate, joining);
+ SortedArrayList<Node.Id> notInFastPath = CollectionSerializers.deserializeSortedArrayList(in, version, nodeId, Node.Id[]::new);
+ SortedArrayList<Node.Id> 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<Node.Id> 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<Verb> 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<Writes, Result> 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<TxnId, AsyncResult<PreAccept.PreAcceptOk>> 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);
-
- mutable.setParticipants(StoreParticipants.all(route));
- mutable.durability(NotDurable);
-
- return mutable;
+ builder.partialTxn(partialTxn);
+ if (saveStatus.known.deps().hasPreAcceptedOrProposedOrDecidedDeps())
+ builder.partialDeps(partialDeps);
+
+ 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());
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org