natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Benedict Elliott Smith 2025-01-08 09:58:12 +00:00 committed by David Capwell
parent 13625288c6
commit 3e16efa490
48 changed files with 695 additions and 424 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit c80ec466f5dd633d4fcbaedc5674f183446def34
Subproject commit f9ed591f1b91115351b871640b6318f11dd208af

2
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -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();

7
src/java/org/apache/cassandra/net/Verb.java
View File

@ -84,9 +84,9 @@ import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification;
import org.apache.cassandra.service.accord.FetchTopology;
import org.apache.cassandra.service.accord.FetchMinEpoch;
import org.apache.cassandra.service.accord.interop.AccordInteropApply;
import org.apache.cassandra.service.accord.interop.AccordInteropCommit;
import org.apache.cassandra.service.accord.interop.AccordInteropRead;
import org.apache.cassandra.service.accord.interop.AccordInteropReadRepair;
import org.apache.cassandra.service.accord.interop.AccordInteropStableThenRead;
import org.apache.cassandra.service.accord.serializers.AcceptSerializers;
import org.apache.cassandra.service.accord.serializers.ApplySerializers;
import org.apache.cassandra.service.accord.serializers.BeginInvalidationSerializers;
@ -316,7 +316,7 @@ public enum Verb
ACCORD_PRE_ACCEPT_REQ (121, P2, writeTimeout, IMMEDIATE, () -> PreacceptSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_PRE_ACCEPT_RSP ),
ACCORD_ACCEPT_RSP (122, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.reply, AccordService::responseHandlerOrNoop ),
ACCORD_ACCEPT_REQ (123, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ),
ACCORD_ACCEPT_INVALIDATE_REQ (124, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.invalidate, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ),
ACCORD_NOT_ACCEPT_REQ (124, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.notAccept, AccordService::requestHandlerOrNoop, ACCORD_ACCEPT_RSP ),
ACCORD_READ_RSP (125, P2, readTimeout, IMMEDIATE, () -> ReadDataSerializers.reply, AccordService::responseHandlerOrNoop ),
ACCORD_READ_REQ (126, P2, readTimeout, IMMEDIATE, () -> ReadDataSerializers.readData, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_COMMIT_REQ (127, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
@ -331,6 +331,7 @@ public enum Verb
ACCORD_AWAIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.request, AccordService::requestHandlerOrNoop, ACCORD_AWAIT_RSP ),
ACCORD_AWAIT_ASYNC_RSP_REQ (137, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.asyncReply, AccordService::requestHandlerOrNoop ),
ACCORD_WAIT_UNTIL_APPLIED_REQ (138, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_STABLE_THEN_READ_REQ (139, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.stableThenRead, AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_INFORM_DURABLE_REQ (140, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.reply, AccordService::responseHandlerOrNoop ),
ACCORD_CHECK_STATUS_REQ (142, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.request, AccordService::requestHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ),
@ -356,7 +357,7 @@ public enum Verb
ACCORD_INTEROP_READ_RSP (155, P2, writeTimeout, IMMEDIATE, () -> AccordInteropRead.replySerializer, AccordService::responseHandlerOrNoop),
ACCORD_INTEROP_READ_REQ (156, P2, writeTimeout, IMMEDIATE, () -> AccordInteropRead.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP),
ACCORD_INTEROP_COMMIT_REQ (157, P2, writeTimeout, IMMEDIATE, () -> AccordInteropCommit.serializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP),
ACCORD_INTEROP_STABLE_THEN_READ_REQ(157, P2, writeTimeout, IMMEDIATE, () -> AccordInteropStableThenRead.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_RSP),
ACCORD_INTEROP_READ_REPAIR_RSP (158, P2, writeTimeout, IMMEDIATE, () -> AccordInteropReadRepair.replySerializer, AccordService::responseHandlerOrNoop),
ACCORD_INTEROP_READ_REPAIR_REQ (159, P2, writeTimeout, IMMEDIATE, () -> AccordInteropReadRepair.requestSerializer, AccordService::requestHandlerOrNoop, ACCORD_INTEROP_READ_REPAIR_RSP),
ACCORD_INTEROP_APPLY_REQ (160, P2, writeTimeout, IMMEDIATE, () -> AccordInteropApply.serializer, AccordService::requestHandlerOrNoop, ACCORD_APPLY_RSP),

4
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -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:

29
src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
View File

@ -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;

79
src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
View File

@ -26,7 +26,7 @@ import accord.api.Result;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.Command.WaitingOn;
import accord.local.CommonAttributes;
import accord.local.ICommand;
import accord.local.Node;
import accord.local.StoreParticipants;
import accord.local.cfk.CommandsForKey;
@ -61,9 +61,9 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.serializers.ResultSerializers;
import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer;
import org.apache.cassandra.service.accord.txn.AccordUpdate;
import org.apache.cassandra.service.accord.txn.TxnData;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.service.accord.txn.TxnResult;
@ -271,53 +271,86 @@ public class AccordObjectSizes
private static class CommandEmptySizes
{
private final static TokenKey EMPTY_KEY = new TokenKey(EMPTY_ID, null);
private final static TxnId EMPTY_TXNID = new TxnId(42, 42, Kind.Read, Domain.Key, new Node.Id(42));
private final static TxnId EMPTY_TXNID = new TxnId(42, 42, 0, Kind.Read, Domain.Key, new Node.Id(42));
private static CommonAttributes attrs(boolean hasDeps, boolean hasTxn, boolean executes)
private static ICommand attrs(boolean hasDeps, boolean hasTxn, boolean executes)
{
FullKeyRoute route = new FullKeyRoute(EMPTY_KEY, new RoutingKey[]{ EMPTY_KEY });
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(EMPTY_TXNID)
.setParticipants(StoreParticipants.empty(EMPTY_TXNID, route, !executes));
attrs.durability(Status.Durability.NotDurable);
ICommand.Builder builder = new ICommand.Builder(EMPTY_TXNID)
.setParticipants(StoreParticipants.empty(EMPTY_TXNID, route, !executes))
.durability(Status.Durability.NotDurable)
.executeAt(EMPTY_TXNID)
.promised(Ballot.ZERO);
if (hasDeps)
attrs.partialDeps(PartialDeps.NONE);
builder.partialDeps(PartialDeps.NONE);
if (hasTxn)
attrs.partialTxn(new PartialTxn.InMemory(null, null, null, null, null));
builder.partialTxn(new PartialTxn.InMemory(null, null, null, null, null));
return attrs;
if (executes)
{
builder.waitingOn(WaitingOn.empty(Domain.Key));
builder.result(new TxnData());
}
return builder;
}
private static final Writes EMPTY_WRITES = new Writes(EMPTY_TXNID, EMPTY_TXNID, Keys.EMPTY, (key, safeStore, txnId, executeAt, store, txn) -> null);
private static final Result EMPTY_RESULT = new Result() {};
final static long NOT_DEFINED = measure(Command.SerializerSupport.notDefined(attrs(false, false, false), Ballot.ZERO));
final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true, false), EMPTY_TXNID, Ballot.ZERO));;
final static long ACCEPTED = measure(Command.SerializerSupport.accepted(attrs(true, false, false), SaveStatus.Accepted, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO));
final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true, false), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null));
final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.empty(Domain.Key), EMPTY_WRITES, ResultSerializers.APPLIED));
final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID)));
final static long NOT_DEFINED = measure(Command.NotDefined.notDefined(attrs(false, false, false)));
final static long PREACCEPTED = measure(Command.PreAccepted.preaccepted(attrs(false, true, false), SaveStatus.PreAccepted));
final static long NOTACCEPTED = measure(Command.NotAcceptedWithoutDefinition.notAccepted(attrs(false, false, false), SaveStatus.NotAccepted));
final static long ACCEPTED = measure(Command.Accepted.accepted(attrs(true, false, false), SaveStatus.AcceptedMedium));
final static long COMMITTED = measure(Command.Committed.committed(attrs(true, true, false), SaveStatus.Committed));
final static long EXECUTED = measure(Command.Executed.executed(attrs(true, true, true), SaveStatus.Applied));
final static long TRUNCATED = measure(Command.Truncated.truncatedApply(attrs(false, false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
final static long INVALIDATED = measure(Command.Truncated.invalidated(EMPTY_TXNID, StoreParticipants.empty(EMPTY_TXNID)));
private static long emptySize(Command command)
{
switch (command.status())
switch (command.saveStatus())
{
case Uninitialised:
case NotDefined:
return NOT_DEFINED;
case PreAccepted:
case PreAcceptedWithDeps:
case PreAcceptedWithVote:
return PREACCEPTED;
case NotAccepted:
case PreNotAccepted:
case AcceptedInvalidate:
case Accepted:
return NOTACCEPTED;
case NotAcceptedWithDefinition:
case NotAcceptedWithDefAndVote:
case NotAcceptedWithDefAndDeps:
case PreNotAcceptedWithDefinition:
case PreNotAcceptedWithDefAndVote:
case PreNotAcceptedWithDefAndDeps:
case AcceptedInvalidateWithDefinition:
case AcceptedMedium:
case AcceptedMediumWithDefinition:
case AcceptedSlow:
case AcceptedSlowWithDefinition:
case PreCommitted:
case PreCommittedWithDeps:
case PreCommittedWithFixedDeps:
case PreCommittedWithDefinition:
case PreCommittedWithDefAndDeps:
case PreCommittedWithDefAndFixedDeps:
return ACCEPTED;
case Committed:
case ReadyToExecute:
case Stable:
return COMMITTED;
case PreApplied:
case Applying:
case Applied:
return EXECUTED;
case Truncated:
case TruncatedApply:
case TruncatedApplyWithDeps:
case TruncatedApplyWithOutcome:
case ErasedOrVestigial:
case Erased:
return TRUNCATED;
case Invalidated:
return INVALIDATED;

9
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -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);
}
}
}

17
src/java/org/apache/cassandra/service/accord/AccordTopology.java
View File

@ -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)
{

4
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
View File

@ -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)

9
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -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)
{

8
src/java/org/apache/cassandra/service/accord/fastpath/ParameterizedFastPathStrategy.java
View File

@ -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)

2
src/java/org/apache/cassandra/service/accord/fastpath/SimpleFastPathStrategy.java
View File

@ -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;
}

15
src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
View File

@ -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:

74
src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java
View File

@ -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();
}
}
}

14
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
View File

@ -41,7 +41,6 @@ import accord.local.Node;
import accord.local.Node.Id;
import accord.messages.Commit;
import accord.messages.Commit.Kind;
import accord.messages.ReadTxnData;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
@ -99,6 +98,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.Dispatcher;
import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard;
import static accord.primitives.Txn.Kind.Write;
import static accord.utils.Invariants.checkArgument;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteMetrics;
@ -234,10 +234,8 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
{
Node.Id id = endpointMapper.mappedId(to);
SinglePartitionReadCommand command = (SinglePartitionReadCommand) message.payload;
AccordInteropRead read = new AccordInteropRead(id, executes, txnId, readScope, executeAt.epoch(), command);
// TODO (required): understand interop and whether StableFastPath is appropriate
AccordInteropCommit commit = new AccordInteropCommit(Kind.StableFastPath, id, coordinateTopology, allTopologies,
txnId, txn, route, executeAt, deps, read);
AccordInteropStableThenRead commit = new AccordInteropStableThenRead(id, allTopologies, txnId, Kind.StableFastPath, executeAt, txn, deps, route, command);
node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this));
}
@ -384,7 +382,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
{
for (Node.Id to : executeTopology.nodes())
if (!contacted.contains(endpointMapper.mappedEndpoint(to)))
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps));
}
public void start()
@ -394,7 +392,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
for (Node.Id to : allTopologies.nodes())
{
if (!executeTopology.contains(to))
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps));
}
}
AsyncChain<Data> result;
@ -406,7 +404,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
CommandStore cs = node.commandStores().select(route.homeKey());
result.beginAsResult().withExecutor(cs).begin((data, failure) -> {
if (failure == null)
((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
((CoordinationAdapter)node.coordinationAdapter(txnId, Standard)).persist(node, executes, route, txnId, txn, executeAt, deps, txnId.is(Write) ? txn.execute(txnId, executeAt, data) : null, txn.result(txnId, executeAt, data), callback);
else
callback.accept(null, failure);
});
@ -420,7 +418,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
// and can be extended similar to MessageType which allows additional types not from Accord to be added
// This commit won't necessarily execute before the interop read repair message so there could be an insufficient which is fine
for (Node.Id to : executeTopology.nodes())
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps));
repairUpdate.runBRR(AccordInteropExecution.this);
return new TxnData();
});

20
src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
View File

@ -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);
}
}

20
src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java
View File

@ -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;
}

3
src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadCallback.java
View File

@ -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;
}
}

16
src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
View File

@ -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;
}

181
src/java/org/apache/cassandra/service/accord/interop/AccordInteropStableThenRead.java Normal file
View File

@ -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 +
'}';
}
}

34
src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java
View File

@ -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:

9
src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
View File

@ -35,6 +35,8 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import static accord.primitives.Txn.Kind.Write;
public class ApplySerializers
{
@ -68,7 +70,8 @@ public class ApplySerializers
DepsSerializers.partialDeps.serialize(apply.deps, out, version);
CommandSerializers.nullablePartialTxn.serialize(apply.txn, out, version);
KeySerializers.nullableFullRoute.serialize(apply.fullRoute, out, version);
CommandSerializers.writes.serialize(apply.writes, out, version);
if (apply.txnId.is(Write))
CommandSerializers.writes.serialize(apply.writes, out, version);
}
protected abstract A deserializeApply(TxnId txnId, Route<?> scope, long minEpoch, long waitForEpoch, Apply.Kind kind,
@ -83,7 +86,7 @@ public class ApplySerializers
DepsSerializers.partialDeps.deserialize(in, version),
CommandSerializers.nullablePartialTxn.deserialize(in, version),
KeySerializers.nullableFullRoute.deserialize(in, version),
CommandSerializers.writes.deserialize(in, version),
(txnId.is(Write) ? CommandSerializers.writes.deserialize(in, version) : null),
ResultSerializers.APPLIED);
}
@ -96,7 +99,7 @@ public class ApplySerializers
+ DepsSerializers.partialDeps.serializedSize(apply.deps, version)
+ CommandSerializers.nullablePartialTxn.serializedSize(apply.txn, version)
+ KeySerializers.nullableFullRoute.serializedSize(apply.fullRoute, version)
+ CommandSerializers.writes.serializedSize(apply.writes, version);
+ (apply.txnId.is(Write) ? CommandSerializers.writes.serializedSize(apply.writes, version) : 0);
}
}

8
src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java
View File

@ -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

2
src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
View File

@ -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,

18
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
View File

@ -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);
}
};

46
src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
View File

@ -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

4
src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
View File

@ -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);
}

36
src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
View File

@ -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));
}
};

90
src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java
View File

@ -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));
}
};
}

10
src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
View File

@ -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);

19
src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java
View File

@ -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;
}
};

2
src/java/org/apache/cassandra/tcm/transformations/AccordMarkStale.java
View File

@ -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());

2
test/distributed/org/apache/cassandra/distributed/test/accord/AccordInteroperabilityTest.java
View File

@ -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());
});
}

2
test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java
View File

@ -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();

2
test/simulator/test/org/apache/cassandra/simulator/test/AccordHarrySimulationTest.java
View File

@ -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));

9
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -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...

12
test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
View File

@ -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);

17
test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
View File

@ -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);

12
test/unit/org/apache/cassandra/service/accord/AccordFastPathCoordinatorTest.java
View File

@ -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);

6
test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java
View File

@ -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),
() -> {});

11
test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
View File

@ -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);

49
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -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;

12
test/unit/org/apache/cassandra/service/accord/AccordTopologyTest.java
View File

@ -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);
}
}

13
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -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();

2
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java
View File

@ -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);

2
test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java
View File

@ -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)
{

93
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -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);

82
test/unit/org/apache/cassandra/utils/AccordGenerators.java
View File

@ -33,7 +33,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedMap;
import accord.local.Command;
import accord.local.CommonAttributes;
import accord.local.ICommand;
import accord.local.DurableBefore;
import accord.local.RedundantBefore;
import accord.local.StoreParticipants;
@ -49,6 +49,7 @@ import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.SaveStatus;
import accord.primitives.Seekables;
import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
@ -75,6 +76,7 @@ import org.quicktheories.impl.JavaRandom;
import static accord.local.CommandStores.RangesForEpoch;
import static accord.primitives.Status.Durability.NotDurable;
import static accord.primitives.Txn.Kind.Write;
import static org.apache.cassandra.service.accord.AccordTestUtils.TABLE_ID1;
import static org.apache.cassandra.service.accord.AccordTestUtils.createPartialTxn;
@ -212,18 +214,30 @@ public class AccordGenerators
this.keysOrRanges = txn.keys();
}
private CommonAttributes attributes(SaveStatus saveStatus)
private ICommand attributes(SaveStatus saveStatus)
{
CommonAttributes.Mutable mutable = new CommonAttributes.Mutable(txnId);
ICommand.Builder builder = new ICommand.Builder(txnId);
if (saveStatus.known.isDefinitionKnown())
mutable.partialTxn(partialTxn);
if (saveStatus.known.deps.hasProposedOrDecidedDeps())
mutable.partialDeps(partialDeps);
builder.partialTxn(partialTxn);
if (saveStatus.known.deps().hasPreAcceptedOrProposedOrDecidedDeps())
builder.partialDeps(partialDeps);
mutable.setParticipants(StoreParticipants.all(route));
mutable.durability(NotDurable);
return mutable;
builder.setParticipants(StoreParticipants.all(route));
builder.durability(NotDurable);
if (saveStatus.compareTo(SaveStatus.PreAccepted) >= 0)
builder.executeAt(executeAt);
builder.promised(promised);
if (saveStatus.status.compareTo(Status.PreAccepted) > 0)
builder.acceptedOrCommitted(accepted);
if (saveStatus.compareTo(SaveStatus.Stable) >= 0 && !saveStatus.hasBeen(Status.Truncated))
builder.waitingOn(waitingOn);
if (saveStatus.hasBeen(Status.PreApplied) && !saveStatus.hasBeen(Status.Truncated))
{
if (txnId.is(Write))
builder.writes(new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)));
builder.result(new TxnData());
}
return builder;
}
public Command build(SaveStatus saveStatus)
@ -235,45 +249,61 @@ public class AccordGenerators
throw new IllegalArgumentException("TruncatedApplyWithDeps is not a valid state for a Command to be in, its for FetchData");
case Uninitialised:
case NotDefined:
return Command.SerializerSupport.notDefined(attributes(saveStatus), Ballot.ZERO);
return Command.NotDefined.notDefined(attributes(saveStatus), Ballot.ZERO);
case PreAccepted:
return Command.SerializerSupport.preaccepted(attributes(saveStatus), executeAt, Ballot.ZERO);
case PreAcceptedWithVote:
case PreAcceptedWithDeps:
return Command.PreAccepted.preaccepted(attributes(saveStatus), saveStatus);
case PreNotAccepted:
case PreNotAcceptedWithDefinition:
case PreNotAcceptedWithDefAndDeps:
case PreNotAcceptedWithDefAndVote:
case NotAccepted:
case NotAcceptedWithDefinition:
case NotAcceptedWithDefAndDeps:
case NotAcceptedWithDefAndVote:
case AcceptedInvalidate:
return Command.AcceptedInvalidateWithoutDefinition.acceptedInvalidate(attributes(saveStatus), promised, Ballot.ZERO);
return Command.NotAcceptedWithoutDefinition.acceptedInvalidate(attributes(saveStatus));
case Accepted:
case AcceptedWithDefinition:
case AcceptedMedium:
case AcceptedMediumWithDefinition:
case AcceptedMediumWithDefAndVote:
case AcceptedInvalidateWithDefinition:
case AcceptedSlow:
case AcceptedSlowWithDefinition:
case AcceptedSlowWithDefAndVote:
case PreCommittedWithDefinition:
case PreCommittedWithDefinitionAndAcceptedDeps:
case PreCommittedWithAcceptedDeps:
case PreCommittedWithDeps:
case PreCommittedWithFixedDeps:
case PreCommittedWithDefAndDeps:
case PreCommittedWithDefAndFixedDeps:
case PreCommitted:
return Command.SerializerSupport.accepted(attributes(saveStatus), saveStatus, executeAt, promised, accepted);
return Command.Accepted.accepted(attributes(saveStatus), saveStatus);
case Committed:
return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, null);
return Command.Committed.committed(attributes(saveStatus), saveStatus);
case Stable:
case ReadyToExecute:
return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, waitingOn);
return Command.Committed.committed(attributes(saveStatus), saveStatus);
case PreApplied:
case Applying:
case Applied:
return Command.SerializerSupport.executed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, waitingOn, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData());
return Command.Executed.executed(attributes(saveStatus), saveStatus);
case TruncatedApply:
if (txnId.kind().awaitsOnlyDeps()) return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null, txnId);
else return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null);
if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null, txnId);
else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, null, null);
case TruncatedApplyWithOutcome:
if (txnId.kind().awaitsOnlyDeps()) return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData(), txnId);
else return Command.SerializerSupport.truncatedApply(attributes(saveStatus), saveStatus, executeAt, new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)), new TxnData());
if (txnId.kind().awaitsOnlyDeps()) return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges,new TxnWrite(Collections.emptyList(), true)) : null, new TxnData(), txnId);
else return Command.Truncated.truncatedApply(attributes(saveStatus), saveStatus, executeAt, txnId.is(Write) ? new Writes(txnId, executeAt, keysOrRanges, new TxnWrite(Collections.emptyList(), true)) : null, new TxnData());
case Erased:
case ErasedOrVestigial:
case Invalidated:
return Command.SerializerSupport.invalidated(txnId, attributes(saveStatus).participants());
return Command.Truncated.invalidated(txnId, attributes(saveStatus).participants());
}
}
}