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

CASSANDRA-18365: Protocol fixes

This commit is contained in:
Benedict Elliott Smith 2024-01-17 17:08:21 +00:00 committed by David Capwell
parent 5385e9c60a
commit 8de3dfc711
34 changed files with 460 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit c524b6d3de3923ccb6314715bd987f3b891348ab
Subproject commit 3789c5bfec50eb96157c0a55af77f78ee0cac804

3
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -20,9 +20,6 @@ package org.apache.cassandra.locator;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.function.Supplier;
import java.util.*;

10
src/java/org/apache/cassandra/metrics/AccordMetrics.java
View File

@ -41,7 +41,7 @@ public class AccordMetrics
public final static AccordMetrics readMetrics = new AccordMetrics("ro");
public final static AccordMetrics writeMetrics = new AccordMetrics("rw");
public static final String COMMIT_LATENCY = "CommitLatency";
public static final String STABLE_LATENCY = "StableLatency";
public static final String EXECUTE_LATENCY = "ExecuteLatency";
public static final String APPLY_LATENCY = "ApplyLatency";
public static final String APPLY_DURATION = "ApplyDuration";
@ -63,7 +63,7 @@ public class AccordMetrics
/**
* The time between start on the coordinator and commit on this replica.
*/
public final Timer commitLatency;
public final Timer stableLatency;
/**
* The time between start on the coordinator and execution on this replica.
@ -135,7 +135,7 @@ public class AccordMetrics
private AccordMetrics(String scope)
{
DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope);
commitLatency = Metrics.timer(replica.createMetricName(COMMIT_LATENCY));
stableLatency = Metrics.timer(replica.createMetricName(STABLE_LATENCY));
executeLatency = Metrics.timer(replica.createMetricName(EXECUTE_LATENCY));
applyLatency = Metrics.timer(replica.createMetricName(APPLY_LATENCY));
applyDuration = Metrics.timer(replica.createMetricName(APPLY_DURATION));
@ -204,14 +204,14 @@ public class AccordMetrics
}
@Override
public void onCommitted(Command cmd)
public void onStable(Command cmd)
{
long now = AccordService.uniqueNow();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
long trxTimestamp = cmd.txnId().hlc();
metrics.commitLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
metrics.stableLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
}
}

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

@ -110,12 +110,12 @@ import static accord.messages.MessageType.BEGIN_INVALIDATE_REQ;
import static accord.messages.MessageType.BEGIN_RECOVER_REQ;
import static accord.messages.MessageType.COMMIT_INVALIDATE_REQ;
import static accord.messages.MessageType.COMMIT_MAXIMAL_REQ;
import static accord.messages.MessageType.COMMIT_MINIMAL_REQ;
import static accord.messages.MessageType.COMMIT_SLOW_PATH_REQ;
import static accord.messages.MessageType.INFORM_DURABLE_REQ;
import static accord.messages.MessageType.INFORM_OF_TXN_REQ;
import static accord.messages.MessageType.PRE_ACCEPT_REQ;
import static accord.messages.MessageType.PROPAGATE_APPLY_MSG;
import static accord.messages.MessageType.PROPAGATE_COMMIT_MSG;
import static accord.messages.MessageType.PROPAGATE_STABLE_MSG;
import static accord.messages.MessageType.PROPAGATE_OTHER_MSG;
import static accord.messages.MessageType.PROPAGATE_PRE_ACCEPT_MSG;
import static accord.messages.MessageType.SET_GLOBALLY_DURABLE_REQ;
@ -124,6 +124,9 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac
import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Daemon.NON_DAEMON;
import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Interrupts.SYNCHRONIZED;
import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.SAFE;
import static accord.messages.MessageType.STABLE_FAST_PATH_REQ;
import static accord.messages.MessageType.STABLE_MAXIMAL_REQ;
import static accord.messages.MessageType.STABLE_SLOW_PATH_REQ;
import static org.apache.cassandra.db.TypeSizes.BYTE_SIZE;
import static org.apache.cassandra.db.TypeSizes.INT_SIZE;
import static org.apache.cassandra.db.TypeSizes.LONG_SIZE;
@ -829,8 +832,11 @@ public class AccordJournal implements Shutdownable
PRE_ACCEPT (64, PRE_ACCEPT_REQ, PreacceptSerializers.request, TXN ),
ACCEPT (65, ACCEPT_REQ, AcceptSerializers.request, TXN ),
ACCEPT_INVALIDATE (66, ACCEPT_INVALIDATE_REQ, AcceptSerializers.invalidate, EPOCH),
COMMIT_MINIMAL (67, COMMIT_MINIMAL_REQ, CommitSerializers.request, TXN ),
COMMIT_SLOW_PATH (67, COMMIT_SLOW_PATH_REQ, CommitSerializers.request, TXN ),
COMMIT_MAXIMAL (68, COMMIT_MAXIMAL_REQ, CommitSerializers.request, TXN ),
STABLE_FAST_PATH (87, STABLE_FAST_PATH_REQ, CommitSerializers.request, TXN ),
STABLE_SLOW_PATH (88, STABLE_SLOW_PATH_REQ, CommitSerializers.request, TXN ),
STABLE_MAXIMAL (89, STABLE_MAXIMAL_REQ, CommitSerializers.request, TXN ),
COMMIT_INVALIDATE (69, COMMIT_INVALIDATE_REQ, CommitSerializers.invalidate, INVL ),
APPLY_MINIMAL (70, APPLY_MINIMAL_REQ, ApplySerializers.request, TXN ),
APPLY_MAXIMAL (71, APPLY_MAXIMAL_REQ, ApplySerializers.request, TXN ),
@ -845,13 +851,13 @@ public class AccordJournal implements Shutdownable
/* Accord local messages */
PROPAGATE_PRE_ACCEPT (79, PROPAGATE_PRE_ACCEPT_MSG, FetchSerializers.propagate, LOCAL),
PROPAGATE_COMMIT (80, PROPAGATE_COMMIT_MSG, FetchSerializers.propagate, LOCAL),
PROPAGATE_STABLE (80, PROPAGATE_STABLE_MSG, FetchSerializers.propagate, LOCAL),
PROPAGATE_APPLY (81, PROPAGATE_APPLY_MSG, FetchSerializers.propagate, LOCAL),
PROPAGATE_OTHER (82, PROPAGATE_OTHER_MSG, FetchSerializers.propagate, LOCAL),
/* C* interop messages */
INTEROP_COMMIT_MINIMAL (83, INTEROP_COMMIT_MINIMAL_REQ, COMMIT_MINIMAL_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_COMMIT_MAXIMAL (84, INTEROP_COMMIT_MAXIMAL_REQ, COMMIT_MAXIMAL_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_COMMIT (83, INTEROP_COMMIT_MINIMAL_REQ, STABLE_FAST_PATH_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_COMMIT_MAXIMAL (84, INTEROP_COMMIT_MAXIMAL_REQ, STABLE_MAXIMAL_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_APPLY_MINIMAL (85, INTEROP_APPLY_MINIMAL_REQ, APPLY_MINIMAL_REQ, AccordInteropApply.serializer, TXN),
INTEROP_APPLY_MAXIMAL (86, INTEROP_APPLY_MAXIMAL_REQ, APPLY_MAXIMAL_REQ, AccordInteropApply.serializer, TXN),
;
@ -1377,9 +1383,9 @@ public class AccordJournal implements Shutdownable
}
@Override
public Commit commitMinimal()
public Commit commitSlowPath()
{
return readMessage(txnId, COMMIT_MINIMAL_REQ, Commit.class);
return readMessage(txnId, COMMIT_SLOW_PATH_REQ, Commit.class);
}
@Override
@ -1389,9 +1395,21 @@ public class AccordJournal implements Shutdownable
}
@Override
public Propagate propagateCommit()
public Commit stableFastPath()
{
return readMessage(txnId, PROPAGATE_COMMIT_MSG, Propagate.class);
return readMessage(txnId, STABLE_FAST_PATH_REQ, Commit.class);
}
@Override
public Commit stableMaximal()
{
return readMessage(txnId, STABLE_MAXIMAL_REQ, Commit.class);
}
@Override
public Propagate propagateStable()
{
return readMessage(txnId, PROPAGATE_STABLE_MSG, Propagate.class);
}
@Override
@ -1470,11 +1488,11 @@ public class AccordJournal implements Shutdownable
}
@Override
public Commit commitMinimal()
public Commit commitSlowPath()
{
logger.debug("Fetching {} message for {}", COMMIT_MINIMAL_REQ, txnId);
Commit commit = provider.commitMinimal();
logger.debug("Fetched {} message for {}: {}", COMMIT_MINIMAL_REQ, txnId, commit);
logger.debug("Fetching {} message for {}", COMMIT_SLOW_PATH_REQ, txnId);
Commit commit = provider.commitSlowPath();
logger.debug("Fetched {} message for {}: {}", COMMIT_SLOW_PATH_REQ, txnId, commit);
return commit;
}
@ -1488,11 +1506,29 @@ public class AccordJournal implements Shutdownable
}
@Override
public Propagate propagateCommit()
public Commit stableFastPath()
{
logger.debug("Fetching {} message for {}", PROPAGATE_COMMIT_MSG, txnId);
Propagate propagate = provider.propagateCommit();
logger.debug("Fetched {} message for {}: {}", PROPAGATE_COMMIT_MSG, txnId, propagate);
logger.debug("Fetching {} message for {}", STABLE_FAST_PATH_REQ, txnId);
Commit commit = provider.stableFastPath();
logger.debug("Fetched {} message for {}: {}", STABLE_FAST_PATH_REQ, txnId, commit);
return commit;
}
@Override
public Commit stableMaximal()
{
logger.debug("Fetching {} message for {}", STABLE_MAXIMAL_REQ, txnId);
Commit commit = provider.stableMaximal();
logger.debug("Fetched {} message for {}: {}", STABLE_MAXIMAL_REQ, txnId, commit);
return commit;
}
@Override
public Propagate propagateStable()
{
logger.debug("Fetching {} message for {}", PROPAGATE_STABLE_MSG, txnId);
Propagate propagate = provider.propagateStable();
logger.debug("Fetched {} message for {}: {}", PROPAGATE_STABLE_MSG, txnId, propagate);
return propagate;
}

11
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
View File

@ -821,11 +821,11 @@ public class AccordKeyspace
addEnumCellIfModified(CommandsColumns.status, Command::saveStatus, builder, timestampMicros, nowInSeconds, original, command);
addCellIfModified(CommandsColumns.execute_at, Command::executeAt, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command);
addCellIfModified(CommandsColumns.promised_ballot, Command::promised, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command);
addCellIfModified(CommandsColumns.accepted_ballot, Command::accepted, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command);
addCellIfModified(CommandsColumns.accepted_ballot, Command::acceptedOrCommitted, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, original, command);
// TODO review this is just to work around Truncated not being committed but having a status after committed
// so status claims it is committed.
if (!command.isTruncated() && command.isCommitted())
if (command.isStable() && !command.isTruncated())
{
Command.Committed committed = command.asCommitted();
Command.Committed originalCommitted = original != null && original.isCommitted() ? original.asCommitted() : null;
@ -1284,8 +1284,11 @@ public class AccordKeyspace
return (deps) ->
{
if (bytes == null || !bytes.hasRemaining())
return deps == null ? WaitingOn.EMPTY : WaitingOn.none(deps);
if (bytes == null)
return null;
if (!bytes.hasRemaining())
return WaitingOn.none(deps);
try
{

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

@ -124,8 +124,11 @@ public class AccordMessageSink implements MessageSink
builder.put(MessageType.ACCEPT_INVALIDATE_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ);
builder.put(MessageType.GET_DEPS_REQ, Verb.ACCORD_GET_DEPS_REQ);
builder.put(MessageType.GET_DEPS_RSP, Verb.ACCORD_GET_DEPS_RSP);
builder.put(MessageType.COMMIT_MINIMAL_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.COMMIT_SLOW_PATH_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.COMMIT_MAXIMAL_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.STABLE_FAST_PATH_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.STABLE_SLOW_PATH_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.STABLE_MAXIMAL_REQ, Verb.ACCORD_COMMIT_REQ);
builder.put(MessageType.COMMIT_INVALIDATE_REQ, Verb.ACCORD_COMMIT_INVALIDATE_REQ);
builder.put(MessageType.APPLY_MINIMAL_REQ, Verb.ACCORD_APPLY_REQ);
builder.put(MessageType.APPLY_MAXIMAL_REQ, Verb.ACCORD_APPLY_REQ);

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

@ -296,7 +296,7 @@ public class AccordObjectSizes
final static long NOT_DEFINED = measure(Command.SerializerSupport.notDefined(attrs(false, false), Ballot.ZERO));
final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true), EMPTY_TXNID, null));;
final static long ACCEPTED = measure(Command.SerializerSupport.accepted(attrs(true, false), SaveStatus.Accepted, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO));
final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.EMPTY));
final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null));
final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.EMPTY, EMPTY_WRITES, EMPTY_RESULT));
final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, null));
@ -314,6 +314,7 @@ public class AccordObjectSizes
case PreCommitted:
return ACCEPTED;
case Committed:
case Stable:
case ReadyToExecute:
return COMMITTED;
case PreApplied:
@ -346,13 +347,13 @@ public class AccordObjectSizes
size += sizeNullable(command.executeAt(), AccordObjectSizes::timestamp);
size += sizeNullable(command.partialTxn(), AccordObjectSizes::txn);
size += sizeNullable(command.partialDeps(), AccordObjectSizes::dependencies);
size += sizeNullable(command.accepted(), AccordObjectSizes::timestamp);
size += sizeNullable(command.acceptedOrCommitted(), AccordObjectSizes::timestamp);
size += sizeNullable(command.writes(), AccordObjectSizes::writes);
if (command.result() instanceof TxnResult)
size += sizeNullable(command.result(), AccordObjectSizes::results);
if (!(command instanceof Command.Committed))
if (!(command instanceof Command.Committed && command.saveStatus().hasBeen(Status.Stable)))
return size;
Command.Committed committed = command.asCommitted();

7
src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java
View File

@ -31,6 +31,7 @@ import accord.api.Key;
import accord.api.RoutingKey;
import accord.local.ShardDistributor;
import accord.primitives.Range;
import accord.primitives.RangeFactory;
import accord.primitives.Ranges;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.IPartitioner;
@ -63,6 +64,12 @@ public abstract class AccordRoutingKey extends AccordRoutableKey implements Rout
public abstract long estimatedSizeOnHeap();
public abstract AccordRoutingKey withTable(TableId table);
@Override
public RangeFactory rangeFactory()
{
return (s, e) -> new TokenRange((AccordRoutingKey) s, (AccordRoutingKey) e);
}
public SentinelKey asSentinelKey()
{
return (SentinelKey) this;

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

@ -25,6 +25,7 @@ import accord.local.Node;
import accord.messages.Commit;
import accord.messages.MessageType;
import accord.messages.ReadData;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
@ -44,20 +45,20 @@ public class AccordInteropCommit extends Commit
public static final IVersionedSerializer<AccordInteropCommit> serializer = new CommitSerializer<AccordInteropCommit, AccordInteropRead>(AccordInteropRead.class, AccordInteropRead.requestSerializer)
{
@Override
protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Kind kind, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, executeAt, partialTxn, partialDeps, fullRoute, read);
return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, ballot, executeAt, partialTxn, partialDeps, fullRoute, read);
}
};
public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
{
super(kind, txnId, scope, waitForEpoch, executeAt, partialTxn, partialDeps, fullRoute, readData);
super(kind, txnId, scope, waitForEpoch, ballot, executeAt, partialTxn, partialDeps, fullRoute, readData);
}
public AccordInteropCommit(Kind kind, Node.Id to, Topology coordinateTopology, Topologies topologies, TxnId txnId, Txn txn, FullRoute<?> route, Timestamp executeAt, Deps deps, AccordInteropRead read)
{
super(kind, to, coordinateTopology, topologies, txnId, txn, route, executeAt, deps, (t, u, p) -> read);
super(kind, to, coordinateTopology, topologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, read);
}
@Override
@ -65,8 +66,8 @@ public class AccordInteropCommit extends Commit
{
switch (kind)
{
case Minimal: return AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ;
case Maximal: return AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ;
case StableFastPath: return AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ;
case StableWithTxnAndDeps: return AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ;
default: throw new IllegalStateException();
}
}

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

@ -28,6 +28,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import accord.messages.ReadTxnData;
import accord.primitives.Ballot;
import org.apache.cassandra.schema.TableId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -37,7 +39,7 @@ import accord.api.Data;
import accord.api.Result;
import accord.coordinate.Execute;
import accord.coordinate.Persist;
import accord.coordinate.TxnExecute;
import accord.coordinate.ExecuteTxn;
import accord.local.AgentExecutor;
import accord.local.CommandStore;
import accord.local.Node;
@ -153,13 +155,13 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
}
@Override
public Execute create(Node node, TxnId txnId, Txn txn, FullRoute<?> route, Participants<?> readScope, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
public Execute create(Node node, Topologies topologies, Path path, TxnId txnId, Txn txn, FullRoute<?> route, Participants<?> readScope, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback)
{
// Unrecoverable repair always needs to be run by AccordInteropExecution
AccordUpdate.Kind updateKind = AccordUpdate.kind(txn.update());
ConsistencyLevel consistencyLevel = txn.read() instanceof TxnRead ? ((TxnRead) txn.read()).cassandraConsistencyLevel() : null;
if (updateKind != AccordUpdate.Kind.UNRECOVERABLE_REPAIR && (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ONE || txn.read().keys().isEmpty()))
return TxnExecute.FACTORY.create(node, txnId, txn, route, readScope, executeAt, deps, callback);
return ExecuteTxn.FACTORY.create(node, topologies, path, txnId, txn, route, readScope, executeAt, deps, callback);
return new AccordInteropExecution(node, txnId, txn, updateKind, route, readScope, executeAt, deps, callback, executor, consistencyLevel, endpointMapper);
}
}
@ -252,7 +254,8 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
Node.Id id = endpointMapper.mappedId(to);
SinglePartitionReadCommand command = (SinglePartitionReadCommand) message.payload;
AccordInteropRead read = new AccordInteropRead(id, executes, txnId, readScope, executeAt, command);
AccordInteropCommit commit = new AccordInteropCommit(Commit.Kind.Minimal, id, coordinateTopology, allTopologies,
// TODO (required): understand interop and whether StableFastPath is appropriate
AccordInteropCommit commit = new AccordInteropCommit(Kind.StableFastPath, id, coordinateTopology, allTopologies,
txnId, txn, route, executeAt, deps, read);
node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this));
}
@ -337,14 +340,14 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
for (int i = 0; i < digestRequests.size(); i++)
contacted.add(digestRequests.endpoint(i));
if (readsCurrentlyUnderConstruction.decrementAndGet() == 0)
sendCommitsToUncontacted();
sendStableToUncontacted();
}
private void sendCommitsToUncontacted()
private void sendStableToUncontacted()
{
for (Node.Id to : executeTopology.nodes())
if (!contacted.contains(endpointMapper.mappedEndpoint(to)))
node.send(to, new Commit(Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false));
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
}
@Override
@ -355,7 +358,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
for (Node.Id to : allTopologies.nodes())
{
if (!executeTopology.contains(to))
node.send(to, new Commit(Commit.Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false));
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
}
}
AsyncChain<Data> result;
@ -367,7 +370,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
CommandStore cs = node.commandStores().select(route.homeKey());
result.beginAsResult().withExecutor(cs).begin((data, failure) -> {
if (failure == null)
Persist.persist(node, executes, txnId, route, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
Persist.persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
else
callback.accept(null, failure);
});
@ -380,7 +383,7 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
// TODO (expected): We should send the read in the same message as the commit. This requires refactor ReadData.Kind so that it doesn't specify the ordinal encoding
// and can be extended similar to MessageType which allows additional types not from Accord to be added
for (Node.Id to : executeTopology.nodes())
node.send(to, new Commit(Kind.Minimal, to, coordinateTopology, allTopologies, txnId, txn, route, readScope, executeAt, deps, false));
node.send(to, new Commit(Kind.StableFastPath, to, coordinateTopology, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps, (ReadTxnData) null));
repairUpdate.runBRR(AccordInteropExecution.this);
return new TxnData();
});
@ -408,6 +411,6 @@ public class AccordInteropExecution implements Execute, ReadCoordinator, Maximal
@Override
public void sendMaximalCommit(Id to)
{
Commit.commitMaximal(node, to, txn, txnId, executeAt, route, deps, readScope);
Commit.stableMaximal(node, to, txn, txnId, executeAt, route, deps);
}
}

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

@ -23,7 +23,7 @@ import java.util.function.BiConsumer;
import accord.api.Result;
import accord.api.Update;
import accord.coordinate.Persist;
import accord.coordinate.TxnPersist;
import accord.coordinate.PersistTxn;
import accord.coordinate.tracking.AppliedTracker;
import accord.coordinate.tracking.QuorumTracker;
import accord.coordinate.tracking.RequestStatus;
@ -57,7 +57,7 @@ public class AccordInteropPersist extends Persist
Update update = txn.update();
ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null;
if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty())
return TxnPersist.FACTORY.create(node, topologies, txnId, route, txn, executeAt, deps, writes, result);
return PersistTxn.FACTORY.create(node, topologies, txnId, route, txn, executeAt, deps, writes, result);
return new AccordInteropPersist(node, topologies, txnId, route, txn, executeAt, deps, writes, result, consistencyLevel);
}
};

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

@ -30,7 +30,7 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.RequestCallback;
import static accord.messages.ReadData.ReadNack.NotCommitted;
import static accord.messages.ReadData.CommitOrReadNack.Insufficient;
public abstract class AccordInteropReadCallback<T> implements Callback<ReadReply>
{
@ -63,7 +63,7 @@ public abstract class AccordInteropReadCallback<T> implements Callback<ReadReply
{
wrapped.onResponse(message.responseWith(convertResponse((ReadOk) reply)).withFrom(endpoint));
}
else if (reply == NotCommitted)
else if (reply == Insufficient)
{
// Might still send a response if we send a maximal commit. Accord would tryAlternative and send
// both the commit and an additional repair, but Cassandra doesn't have tryAlternative unless we add

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

@ -173,8 +173,9 @@ public class CheckStatusSerializers
foundKnownMap.serialize(ok.map, out, version);
CommandSerializers.saveStatus.serialize(ok.maxKnowledgeSaveStatus, out, version);
CommandSerializers.saveStatus.serialize(ok.maxSaveStatus, out, version);
CommandSerializers.ballot.serialize(ok.promised, out, version);
CommandSerializers.ballot.serialize(ok.accepted, out, version);
CommandSerializers.ballot.serialize(ok.maxPromised, out, version);
CommandSerializers.ballot.serialize(ok.maxAcceptedOrCommitted, out, version);
CommandSerializers.ballot.serialize(ok.acceptedOrCommitted, out, version);
CommandSerializers.nullableTimestamp.serialize(ok.executeAt, out, version);
out.writeBoolean(ok.isCoordinating);
CommandSerializers.durability.serialize(ok.durability, out, version);
@ -186,7 +187,7 @@ public class CheckStatusSerializers
CheckStatusOkFull okFull = (CheckStatusOkFull) ok;
CommandSerializers.nullablePartialTxn.serialize(okFull.partialTxn, out, version);
DepsSerializer.nullablePartialDeps.serialize(okFull.committedDeps, out, version);
DepsSerializer.nullablePartialDeps.serialize(okFull.stableDeps, out, version);
CommandSerializers.nullableWrites.serialize(okFull.writes, out, version);
}
@ -204,8 +205,9 @@ public class CheckStatusSerializers
FoundKnownMap map = foundKnownMap.deserialize(in, version);
SaveStatus maxKnowledgeStatus = CommandSerializers.saveStatus.deserialize(in, version);
SaveStatus maxStatus = CommandSerializers.saveStatus.deserialize(in, version);
Ballot promised = CommandSerializers.ballot.deserialize(in, version);
Ballot accepted = CommandSerializers.ballot.deserialize(in, version);
Ballot maxPromised = CommandSerializers.ballot.deserialize(in, version);
Ballot maxAcceptedOrCommitted = CommandSerializers.ballot.deserialize(in, version);
Ballot acceptedOrCommitted = CommandSerializers.ballot.deserialize(in, version);
Timestamp executeAt = CommandSerializers.nullableTimestamp.deserialize(in, version);
boolean isCoordinating = in.readBoolean();
Durability durability = CommandSerializers.durability.deserialize(in, version);
@ -213,7 +215,7 @@ public class CheckStatusSerializers
RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
if (kind == OK)
return createOk(map, maxKnowledgeStatus, maxStatus, promised, accepted, executeAt,
return createOk(map, maxKnowledgeStatus, maxStatus, maxPromised, maxAcceptedOrCommitted, acceptedOrCommitted, executeAt,
isCoordinating, durability, route, homeKey);
PartialTxn partialTxn = CommandSerializers.nullablePartialTxn.deserialize(in, version);
@ -225,7 +227,7 @@ public class CheckStatusSerializers
|| maxKnowledgeStatus == SaveStatus.TruncatedApply || maxKnowledgeStatus == SaveStatus.TruncatedApplyWithOutcome || maxKnowledgeStatus == SaveStatus.TruncatedApplyWithDeps)
result = CommandSerializers.APPLIED;
return createOk(map, maxKnowledgeStatus, maxStatus, promised, accepted, executeAt,
return createOk(map, maxKnowledgeStatus, maxStatus, maxPromised, maxAcceptedOrCommitted, acceptedOrCommitted, executeAt,
isCoordinating, durability, route, homeKey, partialTxn, committedDeps, writes, result);
}
@ -242,8 +244,9 @@ public class CheckStatusSerializers
size += foundKnownMap.serializedSize(ok.map, version);
size += CommandSerializers.saveStatus.serializedSize(ok.maxKnowledgeSaveStatus, version);
size += CommandSerializers.saveStatus.serializedSize(ok.maxSaveStatus, version);
size += CommandSerializers.ballot.serializedSize(ok.promised, version);
size += CommandSerializers.ballot.serializedSize(ok.accepted, version);
size += CommandSerializers.ballot.serializedSize(ok.maxPromised, version);
size += CommandSerializers.ballot.serializedSize(ok.maxAcceptedOrCommitted, version);
size += CommandSerializers.ballot.serializedSize(ok.acceptedOrCommitted, version);
size += CommandSerializers.nullableTimestamp.serializedSize(ok.executeAt, version);
size += TypeSizes.BOOL_SIZE;
size += CommandSerializers.durability.serializedSize(ok.durability, version);
@ -255,7 +258,7 @@ public class CheckStatusSerializers
CheckStatusOkFull okFull = (CheckStatusOkFull) ok;
size += CommandSerializers.nullablePartialTxn.serializedSize(okFull.partialTxn, version);
size += DepsSerializer.nullablePartialDeps.serializedSize(okFull.committedDeps, version);
size += DepsSerializer.nullablePartialDeps.serializedSize(okFull.stableDeps, version);
size += CommandSerializers.nullableWrites.serializedSize(okFull.writes, version);
return size;
}

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

@ -46,6 +46,7 @@ import org.apache.cassandra.db.marshal.ValueAccessor;
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.serializers.SmallEnumSerializer.NullableSmallEnumSerializer;
import org.apache.cassandra.service.accord.txn.AccordUpdate;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
@ -241,20 +242,21 @@ public class CommandSerializers
public static final IVersionedSerializer<Writes> nullableWrites = NullableSerializer.wrap(writes);
public static final EnumSerializer<Status.KnownRoute> route = new EnumSerializer<>(Status.KnownRoute.class);
public static final EnumSerializer<Status.Definition> definition = new EnumSerializer<>(Status.Definition.class);
public static final EnumSerializer<Status.KnownExecuteAt> knownExecuteAt = new EnumSerializer<>(Status.KnownExecuteAt.class);
public static final EnumSerializer<Status.KnownDeps> knownDeps = new EnumSerializer<>(Status.KnownDeps.class);
public static final EnumSerializer<Status.Outcome> outcome = new EnumSerializer<>(Status.Outcome.class);
public static final EnumSerializer<Infer.InvalidIfNot> invalidIfNot = new EnumSerializer<>(Infer.InvalidIfNot.class);
public static final EnumSerializer<Infer.IsPreempted> isPreempted = new EnumSerializer<>(Infer.IsPreempted.class);
public static final SmallEnumSerializer<Status.KnownRoute> knownRoute = new SmallEnumSerializer<>(Status.KnownRoute.class);
public static final SmallEnumSerializer<Status.Definition> definition = new SmallEnumSerializer<>(Status.Definition.class);
public static final SmallEnumSerializer<Status.KnownExecuteAt> knownExecuteAt = new SmallEnumSerializer<>(Status.KnownExecuteAt.class);
public static final SmallEnumSerializer<Status.KnownDeps> knownDeps = new SmallEnumSerializer<>(Status.KnownDeps.class);
public static final NullableSmallEnumSerializer<Status.KnownDeps> nullableKnownDeps = new NullableSmallEnumSerializer<>(knownDeps);
public static final SmallEnumSerializer<Status.Outcome> outcome = new SmallEnumSerializer<>(Status.Outcome.class);
public static final SmallEnumSerializer<Infer.InvalidIfNot> invalidIfNot = new SmallEnumSerializer<>(Infer.InvalidIfNot.class);
public static final SmallEnumSerializer<Infer.IsPreempted> isPreempted = new SmallEnumSerializer<>(Infer.IsPreempted.class);
public static final IVersionedSerializer<Known> known = new IVersionedSerializer<>()
{
@Override
public void serialize(Known known, DataOutputPlus out, int version) throws IOException
{
route.serialize(known.route, out, version);
knownRoute.serialize(known.route, out, version);
definition.serialize(known.definition, out, version);
knownExecuteAt.serialize(known.executeAt, out, version);
knownDeps.serialize(known.deps, out, version);
@ -264,7 +266,7 @@ public class CommandSerializers
@Override
public Known deserialize(DataInputPlus in, int version) throws IOException
{
return new Known(route.deserialize(in, version),
return new Known(knownRoute.deserialize(in, version),
definition.deserialize(in, version),
knownExecuteAt.deserialize(in, version),
knownDeps.deserialize(in, version),
@ -274,7 +276,7 @@ public class CommandSerializers
@Override
public long serializedSize(Known known, int version)
{
return route.serializedSize(known.route, version)
return knownRoute.serializedSize(known.route, version)
+ definition.serializedSize(known.definition, version)
+ knownExecuteAt.serializedSize(known.executeAt, version)
+ knownDeps.serializedSize(known.deps, version)

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

@ -23,6 +23,7 @@ import javax.annotation.Nullable;
import accord.messages.Commit;
import accord.messages.ReadData;
import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
@ -30,7 +31,6 @@ import accord.primitives.PartialTxn;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@ -43,25 +43,7 @@ import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSi
public class CommitSerializers
{
private static final IVersionedSerializer<Commit.Kind> kind = new IVersionedSerializer<Commit.Kind>()
{
public void serialize(Commit.Kind kind, DataOutputPlus out, int version) throws IOException
{
Invariants.checkArgument(kind == Commit.Kind.Minimal || kind == Commit.Kind.Maximal);
out.writeBoolean(kind == Commit.Kind.Maximal);
}
public Commit.Kind deserialize(DataInputPlus in, int version) throws IOException
{
return in.readBoolean() ? Commit.Kind.Maximal : Commit.Kind.Minimal;
}
public long serializedSize(Commit.Kind kind, int version)
{
return TypeSizes.BOOL_SIZE;
}
};
private static final IVersionedSerializer<Commit.Kind> kind = new EnumSerializer<>(Commit.Kind.class);
public abstract static class CommitSerializer<C extends Commit, R extends ReadData> extends TxnRequestSerializer<C>
{
@ -76,6 +58,7 @@ public class CommitSerializers
public void serializeBody(C msg, DataOutputPlus out, int version) throws IOException
{
kind.serialize(msg.kind, out, version);
CommandSerializers.ballot.serialize(msg.ballot, out, version);
CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
CommandSerializers.nullablePartialTxn.serialize(msg.partialTxn, out, version);
DepsSerializer.partialDeps.serialize(msg.partialDeps, out, version);
@ -83,7 +66,8 @@ public class CommitSerializers
serializeNullable(msg.readData, out, version, read);
}
protected abstract C deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind, Timestamp executeAt,
protected abstract C deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind,
Ballot ballot, Timestamp executeAt,
@Nullable PartialTxn partialTxn, PartialDeps partialDeps,
@Nullable FullRoute<?> fullRoute, @Nullable ReadData read);
@ -92,6 +76,7 @@ public class CommitSerializers
{
return deserializeCommit(txnId, scope, waitForEpoch,
kind.deserialize(in, version),
CommandSerializers.ballot.deserialize(in, version),
CommandSerializers.timestamp.deserialize(in, version),
CommandSerializers.nullablePartialTxn.deserialize(in, version),
DepsSerializer.partialDeps.deserialize(in, version),
@ -104,6 +89,7 @@ public class CommitSerializers
public long serializedBodySize(C msg, int version)
{
return kind.serializedSize(msg.kind, version)
+ CommandSerializers.ballot.serializedSize(msg.ballot, version)
+ CommandSerializers.timestamp.serializedSize(msg.executeAt, version)
+ CommandSerializers.nullablePartialTxn.serializedSize(msg.partialTxn, version)
+ DepsSerializer.partialDeps.serializedSize(msg.partialDeps, version)
@ -115,9 +101,9 @@ public class CommitSerializers
public static final IVersionedSerializer<Commit> request = new CommitSerializer<Commit, ReadData>(ReadData.class, ReadDataSerializers.readData)
{
@Override
protected Commit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
protected Commit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, executeAt, partialTxn, partialDeps, fullRoute, read);
return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, ballot, executeAt, partialTxn, partialDeps, fullRoute, read);
}
};

1
src/java/org/apache/cassandra/service/accord/serializers/DepsSerializer.java
View File

@ -51,6 +51,7 @@ public abstract class DepsSerializer<D extends Deps> implements IVersionedSerial
return new Deps(keyDeps, rangeDeps);
}
};
public static final IVersionedSerializer<Deps> nullableDeps = NullableSerializer.wrap(deps);
public static final DepsSerializer<PartialDeps> partialDeps = new DepsSerializer<PartialDeps>()
{

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

@ -32,6 +32,7 @@ import accord.messages.CheckStatus;
import accord.messages.Propagate;
import accord.messages.ReadData;
import accord.messages.ReadData.ReadReply;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
@ -94,7 +95,7 @@ public class FetchSerializers
public static final IVersionedSerializer<ReadReply> reply = new IVersionedSerializer<ReadReply>()
{
final ReadData.ReadNack[] nacks = ReadData.ReadNack.values();
final ReadData.CommitOrReadNack[] nacks = ReadData.CommitOrReadNack.values();
final IVersionedSerializer<Data> streamDataSerializer = new CastingSerializer<>(StreamData.class, StreamData.serializer);
@Override
@ -102,7 +103,7 @@ public class FetchSerializers
{
if (!reply.isOk())
{
out.writeByte(1 + ((ReadData.ReadNack) reply).ordinal());
out.writeByte(1 + ((ReadData.CommitOrReadNack) reply).ordinal());
return;
}
@ -148,14 +149,15 @@ public class FetchSerializers
KeySerializers.route.serialize(p.route, out, version);
CommandSerializers.saveStatus.serialize(p.maxKnowledgeSaveStatus, out, version);
CommandSerializers.saveStatus.serialize(p.maxSaveStatus, out, version);
CommandSerializers.ballot.serialize(p.ballot, out, version);
CommandSerializers.durability.serialize(p.durability, out, version);
KeySerializers.nullableRoutingKey.serialize(p.homeKey, out, version);
KeySerializers.nullableRoutingKey.serialize(p.progressKey, out, version);
CommandSerializers.known.serialize(p.achieved, out, version);
CheckStatusSerializers.foundKnownMap.serialize(p.known, out, version);
out.writeBoolean(p.isTruncated);
out.writeBoolean(p.isShardTruncated);
CommandSerializers.nullablePartialTxn.serialize(p.partialTxn, out, version);
DepsSerializer.nullablePartialDeps.serialize(p.committedDeps, out, version);
DepsSerializer.nullablePartialDeps.serialize(p.stableDeps, out, version);
out.writeLong(p.toEpoch);
CommandSerializers.nullableTimestamp.serialize(p.committedExecuteAt, out, version);
CommandSerializers.nullableWrites.serialize(p.writes, out, version);
@ -168,6 +170,7 @@ public class FetchSerializers
Route<?> route = KeySerializers.route.deserialize(in, version);
SaveStatus maxKnowledgeSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
SaveStatus maxSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
Durability durability = CommandSerializers.durability.deserialize(in, version);
RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
RoutingKey progressKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
@ -197,6 +200,7 @@ public class FetchSerializers
route,
maxKnowledgeSaveStatus,
maxSaveStatus,
ballot,
durability,
homeKey,
progressKey,
@ -218,6 +222,7 @@ public class FetchSerializers
+ KeySerializers.route.serializedSize(p.route, version)
+ CommandSerializers.saveStatus.serializedSize(p.maxKnowledgeSaveStatus, version)
+ CommandSerializers.saveStatus.serializedSize(p.maxSaveStatus, version)
+ CommandSerializers.ballot.serializedSize(p.ballot, version)
+ CommandSerializers.durability.serializedSize(p.durability, version)
+ KeySerializers.nullableRoutingKey.serializedSize(p.homeKey, version)
+ KeySerializers.nullableRoutingKey.serializedSize(p.progressKey, version)
@ -225,7 +230,7 @@ public class FetchSerializers
+ CheckStatusSerializers.foundKnownMap.serializedSize(p.known, version)
+ TypeSizes.BOOL_SIZE
+ CommandSerializers.nullablePartialTxn.serializedSize(p.partialTxn, version)
+ DepsSerializer.nullablePartialDeps.serializedSize(p.committedDeps, version)
+ DepsSerializer.nullablePartialDeps.serializedSize(p.stableDeps, version)
+ TypeSizes.sizeof(p.toEpoch)
+ CommandSerializers.nullableTimestamp.serializedSize(p.committedExecuteAt, version)
+ CommandSerializers.nullableWrites.serializedSize(p.writes, version)

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

@ -23,7 +23,7 @@ import java.io.IOException;
import accord.api.Data;
import accord.messages.ApplyThenWaitUntilApplied;
import accord.messages.ReadData;
import accord.messages.ReadData.ReadNack;
import accord.messages.ReadData.CommitOrReadNack;
import accord.messages.ReadData.ReadOk;
import accord.messages.ReadData.ReadReply;
import accord.messages.ReadData.ReadType;
@ -172,7 +172,7 @@ public class ReadDataSerializers
public static final class ReplySerializer<D extends Data> implements IVersionedSerializer<ReadReply>
{
// TODO (now): use something other than ordinal
final ReadNack[] nacks = ReadNack.values();
final CommitOrReadNack[] nacks = CommitOrReadNack.values();
private final IVersionedSerializer<D> dataSerializer;
public ReplySerializer(IVersionedSerializer<D> dataSerializer)
@ -185,7 +185,7 @@ public class ReadDataSerializers
{
if (!reply.isOk())
{
out.writeByte(1 + ((ReadNack) reply).ordinal());
out.writeByte(1 + ((CommitOrReadNack) reply).ordinal());
return;
}

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

@ -23,6 +23,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.local.Status;
import accord.messages.BeginRecovery;
import accord.messages.BeginRecovery.RecoverNack;
@ -31,7 +32,7 @@ import accord.messages.BeginRecovery.RecoverReply;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.LatestDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Timestamp;
@ -89,8 +90,7 @@ public class RecoverySerializers
CommandSerializers.status.serialize(recoverOk.status, out, version);
CommandSerializers.ballot.serialize(recoverOk.accepted, out, version);
CommandSerializers.nullableTimestamp.serialize(recoverOk.executeAt, out, version);
DepsSerializer.partialDeps.serialize(recoverOk.deps, out, version);
DepsSerializer.nullablePartialDeps.serialize(recoverOk.acceptedDeps, out, version);
latestDeps.serialize(recoverOk.deps, out, version);
DepsSerializer.deps.serialize(recoverOk.earlierCommittedWitness, out, version);
DepsSerializer.deps.serialize(recoverOk.earlierAcceptedNoWitness, out, version);
out.writeBoolean(recoverOk.rejectsFastPath);
@ -112,9 +112,9 @@ public class RecoverySerializers
return new RecoverNack(supersededBy);
}
RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull PartialDeps deps, PartialDeps acceptedDeps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version)
RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull LatestDeps deps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version)
{
return new RecoverOk(txnId, status, accepted, executeAt, deps, acceptedDeps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result);
return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result);
}
@Override
@ -135,8 +135,7 @@ public class RecoverySerializers
status,
CommandSerializers.ballot.deserialize(in, version),
CommandSerializers.nullableTimestamp.deserialize(in, version),
DepsSerializer.partialDeps.deserialize(in, version),
DepsSerializer.nullablePartialDeps.deserialize(in, version),
latestDeps.deserialize(in, version),
DepsSerializer.deps.deserialize(in, version),
DepsSerializer.deps.deserialize(in, version),
in.readBoolean(),
@ -157,8 +156,7 @@ public class RecoverySerializers
size += CommandSerializers.status.serializedSize(recoverOk.status, version);
size += CommandSerializers.ballot.serializedSize(recoverOk.accepted, version);
size += CommandSerializers.nullableTimestamp.serializedSize(recoverOk.executeAt, version);
size += DepsSerializer.partialDeps.serializedSize(recoverOk.deps, version);
size += DepsSerializer.nullablePartialDeps.serializedSize(recoverOk.acceptedDeps, version);
size += latestDeps.serializedSize(recoverOk.deps, version);
size += DepsSerializer.deps.serializedSize(recoverOk.earlierCommittedWitness, version);
size += DepsSerializer.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version);
size += TypeSizes.sizeof(recoverOk.rejectsFastPath);
@ -173,4 +171,80 @@ public class RecoverySerializers
+ (reply.isOk() ? serializedOkSize((RecoverOk) reply, version) : serializedNackSize((RecoverNack) reply, version));
}
};
public static final IVersionedSerializer<LatestDeps> latestDeps = new IVersionedSerializer<LatestDeps>()
{
@Override
public void serialize(LatestDeps t, DataOutputPlus out, int version) throws IOException
{
out.writeUnsignedVInt32(t.size());
for (int i = 0 ; i < t.size() ; ++i)
{
RoutingKey start = t.startAt(i);
KeySerializers.routingKey.serialize(start, out, version);
LatestDeps.LatestEntry e = t.valueAt(i);
if (e == null)
{
CommandSerializers.nullableKnownDeps.serialize(null, out, version);
}
else
{
CommandSerializers.nullableKnownDeps.serialize(e.known, out, version);
CommandSerializers.ballot.serialize(e.ballot, out, version);
DepsSerializer.nullableDeps.serialize(e.coordinatedDeps, out, version);
DepsSerializer.nullableDeps.serialize(e.localDeps, out, version);
}
}
KeySerializers.routingKey.serialize(t.startAt(t.size()), out, version);
}
@Override
public LatestDeps deserialize(DataInputPlus in, int version) throws IOException
{
int size = in.readUnsignedVInt32();
RoutingKey[] starts = new RoutingKey[size + 1];
LatestDeps.LatestEntry[] values = new LatestDeps.LatestEntry[size];
for (int i = 0 ; i < size ; ++i)
{
starts[i] = KeySerializers.routingKey.deserialize(in, version);
Status.KnownDeps knownDeps = CommandSerializers.nullableKnownDeps.deserialize(in, version);
if (knownDeps == null)
continue;
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
Deps coordinatedDeps = DepsSerializer.nullableDeps.deserialize(in, version);
Deps localDeps = DepsSerializer.nullableDeps.deserialize(in, version);
values[i] = new LatestDeps.LatestEntry(knownDeps, ballot, coordinatedDeps, localDeps);
}
starts[size] = KeySerializers.routingKey.deserialize(in, version);
return LatestDeps.SerializerSupport.create(true, starts, values);
}
@Override
public long serializedSize(LatestDeps t, int version)
{
long size = 0;
size += TypeSizes.sizeofUnsignedVInt(t.size());
for (int i = 0 ; i < t.size() ; ++i)
{
RoutingKey start = t.startAt(i);
size += KeySerializers.routingKey.serializedSize(start, version);
LatestDeps.LatestEntry e = t.valueAt(i);
if (e == null)
{
size += CommandSerializers.nullableKnownDeps.serializedSize(null, version);
}
else
{
size += CommandSerializers.nullableKnownDeps.serializedSize(e.known, version);
size += CommandSerializers.ballot.serializedSize(e.ballot, version);
size += DepsSerializer.nullableDeps.serializedSize(e.coordinatedDeps, version);
size += DepsSerializer.nullableDeps.serializedSize(e.localDeps, version);
}
}
size += KeySerializers.routingKey.serializedSize(t.startAt(t.size()), version);
return size;
}
};
}

118
src/java/org/apache/cassandra/service/accord/serializers/SmallEnumSerializer.java Normal file
View File

@ -0,0 +1,118 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import java.nio.ByteBuffer;
import javax.annotation.Nullable;
import accord.messages.SimpleReply;
import accord.utils.Invariants;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class SmallEnumSerializer<E extends Enum<E>> implements IVersionedSerializer<E>
{
public static final SmallEnumSerializer<SimpleReply> simpleReply = new SmallEnumSerializer<>(SimpleReply.class);
// TODO: should use something other than ordinal for ser/deser
final E[] values;
public SmallEnumSerializer(Class<E> clazz)
{
this.values = clazz.getEnumConstants();
Invariants.checkArgument(values.length < 255); // allow an extra 1 for nullable variant to ensure consistency
}
public E forOrdinal(int ordinal)
{
return values[ordinal];
}
@Override
public void serialize(E t, DataOutputPlus out, int version) throws IOException
{
out.write(t.ordinal());
}
@Override
public E deserialize(DataInputPlus in, int version) throws IOException
{
return values[in.readByte()];
}
public ByteBuffer serialize(E e)
{
ByteBuffer out = ByteBuffer.allocate(1);
out.put((byte)e.ordinal());
out.flip();
return out;
}
@Override
public long serializedSize(E t, int version)
{
return 1;
}
public static class NullableSmallEnumSerializer<E extends Enum<E>> implements IVersionedSerializer<E>
{
// TODO: should use something other than ordinal for ser/deser
final E[] values;
public NullableSmallEnumSerializer(SmallEnumSerializer<E> wrap)
{
this.values = wrap.values;
}
public E forOrdinal(int ordinal)
{
return values[ordinal];
}
@Override
public void serialize(@Nullable E t, DataOutputPlus out, int version) throws IOException
{
out.write(t == null ? 0 : 1 + t.ordinal());
}
@Override
public E deserialize(DataInputPlus in, int version) throws IOException
{
int ordinal = in.readByte();
return ordinal == 0 ? null : values[ordinal - 1];
}
public ByteBuffer serialize(E e)
{
ByteBuffer out = ByteBuffer.allocate(1);
out.put((byte)(e == null ? 0 : (1 + e.ordinal())));
out.flip();
return out;
}
@Override
public long serializedSize(E t, int version)
{
return 1;
}
}
}

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java
View File

@ -231,12 +231,12 @@ public class AccordMetricsTest extends AccordTestBase
}
}
private void assertReplicaMetrics(int node, String scope, long commits, long executions, long applications)
private void assertReplicaMetrics(int node, String scope, long stable, long executions, long applications)
{
DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_REPLICA, scope);
Map<String, Long> metrics = diff(countingMetrics0).get(node);
Function<String, Long> metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName());
assertThat(metric.apply(AccordMetrics.COMMIT_LATENCY)).isEqualTo(commits);
assertThat(metric.apply(AccordMetrics.STABLE_LATENCY)).isEqualTo(stable);
assertThat(metric.apply(AccordMetrics.EXECUTE_LATENCY)).isEqualTo(executions);
assertThat(metric.apply(AccordMetrics.APPLY_LATENCY)).isEqualTo(applications);
assertThat(metric.apply(AccordMetrics.APPLY_DURATION)).isEqualTo(applications);

1
test/unit/org/apache/cassandra/cql3/selection/SelectionColumnMappingTest.java
View File

@ -40,7 +40,6 @@ import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.utils.ByteBufferUtil;
import static java.util.Arrays.asList;
import static org.apache.cassandra.ServerTestUtils.daemonInitialization;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

1
test/unit/org/apache/cassandra/cql3/validation/entities/FrozenCollectionsTest.java
View File

@ -42,7 +42,6 @@ import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.ServerTestUtils.daemonInitialization;
import static org.junit.Assert.assertEquals;
public class FrozenCollectionsTest extends CQLTester

1
test/unit/org/apache/cassandra/cql3/validation/entities/JsonTest.java
View File

@ -39,7 +39,6 @@ import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;
import static org.apache.cassandra.ServerTestUtils.daemonInitialization;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

2
test/unit/org/apache/cassandra/cql3/validation/entities/UserTypesTest.java
View File

@ -27,8 +27,6 @@ import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.ServerTestUtils.daemonInitialization;
public class UserTypesTest extends CQLTester
{
@BeforeClass

2
test/unit/org/apache/cassandra/cql3/validation/operations/SelectLimitTest.java
View File

@ -28,8 +28,6 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.ServerTestUtils.daemonInitialization;
public class SelectLimitTest extends CQLTester
{
// This method will be ran instead of the CQLTester#setUpClass

8
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -83,8 +83,6 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
import org.apache.cassandra.service.accord.AccordTestUtils;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.utils.FBUtilities;
@ -461,7 +459,7 @@ public class CompactionAccordIteratorsTest
TxnId[] txnIds = additionalCommand ? TXN_IDS : new TxnId[] {TXN_ID};
for (TxnId txnId : txnIds)
{
Txn txn = txnId.rw().isWrite() ? AccordTestUtils.createWriteTxn(42) : AccordTestUtils.createTxn(42);
Txn txn = txnId.kind().isWrite() ? AccordTestUtils.createWriteTxn(42) : AccordTestUtils.createTxn(42);
Seekable key = txn.keys().get(0);
PartialDeps partialDeps = Deps.NONE.slice(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true);
@ -482,9 +480,9 @@ public class CompactionAccordIteratorsTest
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {
Commit commit =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Minimal, txnId, partialTxn, partialDeps, route, null);
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableFastPath, Ballot.ZERO, txnId, partialTxn, partialDeps, route, null);
commandStore.appendToJournal(commit);
CheckedCommands.commit(safe, txnId, route, null, partialTxn, txnId, partialDeps);
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, txnId, partialDeps);
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {

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

@ -157,7 +157,7 @@ public class AccordCommandTest
}));
// check commit
Commit commit = Commit.SerializerSupport.create(txnId, route, 1, Commit.Kind.Maximal, executeAt, partialTxn, deps, fullRoute, null);
Commit commit = Commit.SerializerSupport.create(txnId, route, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute, null);
commandStore.appendToJournal(commit);
getUninterruptibly(commandStore.execute(commit, commit::apply));

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

@ -82,18 +82,19 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
common.route(route);
common.partialDeps(deps.slice(scope));
common.durability(Status.Durability.NotDurable);
Command.WaitingOn waitingOn = Command.WaitingOn.none(deps.slice(scope));
Command.WaitingOn waitingOn = null;
Command.Committed committed = Command.SerializerSupport.committed(common, SaveStatus.Committed, id, Ballot.ZERO, Ballot.ZERO, waitingOn);
AccordSafeCommand safeCommand = new AccordSafeCommand(AccordTestUtils.loaded(id, null));
safeCommand.set(committed);
Commit commit = Commit.SerializerSupport.create(id, route.slice(scope), 1, Commit.Kind.Maximal, id, partialTxn, partialDeps, route, null);
Commit commit = Commit.SerializerSupport.create(id, route.slice(scope), 1, Commit.Kind.StableFastPath, Ballot.ZERO, id, partialTxn, partialDeps, route, null);
store.appendToJournal(commit);
Mutation mutation = AccordKeyspace.getCommandMutation(store, safeCommand, 42);
mutation.apply();
Assertions.assertThat(AccordKeyspace.loadCommand(store, id)).isEqualTo(committed);
Command loaded = AccordKeyspace.loadCommand(store, id);
Assertions.assertThat(loaded).isEqualTo(committed);
}
}

4
test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
View File

@ -92,7 +92,7 @@ public class AccordMessageSinkTest
checkRequestReplies(request,
new AbstractFetchCoordinator.FetchResponse(null, null, id),
ReadData.ReadNack.NotCommitted);
ReadData.CommitOrReadNack.Insufficient);
}
@ -103,7 +103,7 @@ public class AccordMessageSinkTest
Request request = new ReadTxnData(node, topologies, txnId, topology.ranges(), txnId);
checkRequestReplies(request,
new ReadData.ReadOk(null, null),
ReadData.ReadNack.NotCommitted);
ReadData.CommitOrReadNack.Insufficient);
}
private static void checkRequestReplies(Request request, Reply... replies)

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

@ -128,6 +128,19 @@ public class AccordTestUtils
executeAt,
Ballot.ZERO,
Ballot.ZERO,
null);
}
public static Command stable(TxnId txnId, PartialTxn txn, Timestamp executeAt)
{
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId).partialDeps(PartialDeps.NONE);
attrs.partialTxn(txn);
attrs.route(route(txn));
return Command.SerializerSupport.committed(attrs,
SaveStatus.Stable,
executeAt,
Ballot.ZERO,
Ballot.ZERO,
Command.WaitingOn.EMPTY);
}
@ -189,7 +202,7 @@ public class AccordTestUtils
@Override public void preaccepted(Command command, ProgressShard progressShard) {}
@Override public void accepted(Command command, ProgressShard progressShard) {}
@Override public void precommitted(Command command) {}
@Override public void committed(Command command, ProgressShard progressShard) {}
@Override public void stable(Command command, ProgressShard progressShard) {}
@Override public void readyToExecute(Command command) {}
@Override public void executed(Command command, ProgressShard progressShard) {}
@Override public void clear(TxnId txnId) {}

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

@ -84,7 +84,7 @@ public class CommandsForRangesTest
assertThat(cfr.knownIds()).containsExactly(max);
assertThat(cfr.maxRedundant()).isEqualTo(knownIds.size() == 1 ? null : knownIds.get(knownIds.size() - 2));
cfr.prune(new TxnId(max.logicalNext(max.node), max.rw(), max.domain()), FULL_RANGE);
cfr.prune(new TxnId(max.logicalNext(max.node), max.kind(), max.domain()), FULL_RANGE);
assertThat(cfr.knownIds()).isEmpty();
assertThat(cfr.maxRedundant()).isEqualTo(max);
});

83
test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
View File

@ -42,6 +42,7 @@ import accord.local.Command;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.SaveStatus;
import accord.messages.Accept;
import accord.messages.Commit;
import accord.messages.PreAccept;
@ -162,9 +163,9 @@ public class AsyncOperationTest
}
}
private static Command createCommittedAndPersist(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt)
private static Command createStableAndPersist(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt)
{
Command command = AccordTestUtils.Commands.committed(txnId, createPartialTxn(0), executeAt);
Command command = AccordTestUtils.Commands.stable(txnId, createPartialTxn(0), executeAt);
AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null));
safeCommand.set(command);
@ -173,7 +174,8 @@ public class AsyncOperationTest
Commit.SerializerSupport.create(txnId,
command.route().slice(AccordTestUtils.fullRange(command.partialTxn().keys())),
txnId.epoch(),
Commit.Kind.Maximal,
Commit.Kind.StableWithTxnAndDeps,
Ballot.ZERO,
executeAt,
command.partialTxn(),
command.partialDeps(),
@ -184,17 +186,64 @@ public class AsyncOperationTest
return command;
}
private static Command createCommittedAndPersist(AccordCommandStore commandStore, TxnId txnId)
private static Command createStableAndPersist(AccordCommandStore commandStore, TxnId txnId)
{
return createCommittedAndPersist(commandStore, txnId, txnId);
return createStableAndPersist(commandStore, txnId, txnId);
}
private static Command createCommittedUsingLifeCycle(AccordCommandStore commandStore, TxnId txnId)
private static Command createStableUsingFastLifeCycle(AccordCommandStore commandStore, TxnId txnId)
{
return createCommittedUsingLifeCycle(commandStore, txnId, txnId);
return createStableUsingFastLifeCycle(commandStore, txnId, txnId);
}
private static Command createCommittedUsingLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt)
private static Command createStableUsingFastLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt)
{
PartialTxn partialTxn = createPartialTxn(0);
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
PartialRoute<?> partialRoute = route.slice(ranges);
PartialDeps deps = PartialDeps.builder(ranges).build();
// create and write messages to the journal for loading to succeed
PreAccept preAccept =
PreAccept.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), txnId.epoch(), false, txnId.epoch(), partialTxn, route);
Commit stable =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableFastPath, Ballot.ZERO, executeAt, partialTxn, deps, route, null);
commandStore.appendToJournal(preAccept);
commandStore.appendToJournal(stable);
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys()), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null);
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps);
return safe.ifInitialised(txnId).current();
}).beginAsResult());
// clear cache
commandStore.executeBlocking(() -> {
long cacheSize = commandStore.capacity();
commandStore.setCapacity(0);
commandStore.setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
return command;
}
catch (ExecutionException e)
{
throw new AssertionError(e);
}
}
private static Command createStableUsingSlowLifeCycle(AccordCommandStore commandStore, TxnId txnId)
{
return createStableUsingSlowLifeCycle(commandStore, txnId, txnId);
}
private static Command createStableUsingSlowLifeCycle(AccordCommandStore commandStore, TxnId txnId, Timestamp executeAt)
{
PartialTxn partialTxn = createPartialTxn(0);
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
@ -209,18 +258,22 @@ public class AsyncOperationTest
Accept accept =
Accept.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), txnId.epoch(), false, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
Commit commit =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Minimal, executeAt, partialTxn, deps, route, null);
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Commit, Ballot.ZERO, executeAt, partialTxn, deps, route, null);
Commit stable =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableSlowPath, Ballot.ZERO, executeAt, partialTxn, deps, route, null);
commandStore.appendToJournal(preAccept);
commandStore.appendToJournal(accept);
commandStore.appendToJournal(commit);
commandStore.appendToJournal(stable);
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys()), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null);
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps);
CheckedCommands.commit(safe, txnId, route, null, partialTxn, executeAt, deps);
CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps);
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps);
return safe.ifInitialised(txnId).current();
}).beginAsResult());
@ -265,7 +318,7 @@ public class AsyncOperationTest
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
createCommittedAndPersist(commandStore, txnId);
createStableAndPersist(commandStore, txnId);
Consumer<SafeCommandStore> consumer = safeStore -> safeStore.ifInitialised(txnId).readyToExecute();
PreLoadContext ctx = contextFor(txnId);
@ -396,8 +449,12 @@ public class AsyncOperationTest
private static void createCommand(AccordCommandStore commandStore, RandomSource rs, List<TxnId> ids)
{
// to simulate CommandsForKey not being found, use createCommittedAndPersist periodically as it does not update
if (rs.nextBoolean()) ids.forEach(id -> createCommittedAndPersist(commandStore, id));
else ids.forEach(id -> createCommittedUsingLifeCycle(commandStore, id));
switch (rs.nextInt(3))
{
case 0: ids.forEach(id -> createStableAndPersist(commandStore, id)); break;
case 1: ids.forEach(id -> createStableUsingFastLifeCycle(commandStore, id)); break;
case 2: ids.forEach(id -> createStableUsingSlowLifeCycle(commandStore, id));
}
commandStore.unsafeClearCache();
}

1
tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java
View File

@ -28,7 +28,6 @@ import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.operations.PartitionOperation;
import org.apache.cassandra.stress.report.Timer;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.CqlVersion;
import org.apache.cassandra.stress.settings.StressSettings;
public abstract class PredefinedOperation extends PartitionOperation