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

Switch to streaming serialization of SavedCommand 

Patch by Alex Petrov; reviewed by David Capwell for CASSANDRA-19865

Co-authored-by: dcapwell <dcapwell@gmail.com>
This commit is contained in:
Alex Petrov 2024-08-28 17:18:36 +02:00 committed by David Capwell
parent 26d517f4ff
commit 325e48ac39
22 changed files with 1191 additions and 905 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 178952b41f05bfa307aef03dcc013e37fb6230b4
Subproject commit 449b2b4d0bf4bb44d55a3c57f712a4d5a15e7220

44
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -47,11 +47,13 @@ import org.apache.cassandra.concurrent.SequentialExecutorPlus;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.PathUtils;
import org.apache.cassandra.journal.Segments.ReferencedSegment;
import org.apache.cassandra.journal.Segments.ReferencedSegments;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.SavedCommand;
import org.apache.cassandra.utils.Crc;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Simulate;
@ -343,11 +345,16 @@ public class Journal<K, V> implements Shutdownable
return null;
}
// TODO: This should be improved with new index that should take better care of handling multiple items
public List<V> readAll(K id)
{
EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
List<V> res = new ArrayList<>(2);
readAll(id, (in, userVersion) -> res.add(valueSerializer.deserialize(id, in, userVersion)));
return res;
}
public void readAll(K id, Reader reader)
{
EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
try (ReferencedSegments<K, V> segments = selectAndReference(id))
{
for (Segment<K, V> segment : segments.all())
@ -357,7 +364,7 @@ public class Journal<K, V> implements Shutdownable
{
Invariants.checkState(Objects.equals(holder.key, id),
"%s != %s", holder.key, id);
res.add(valueSerializer.deserialize(holder.key, in, segment.descriptor.userVersion));
reader.read(in, segment.descriptor.userVersion);
holder.clear();
}
catch (IOException e)
@ -368,7 +375,6 @@ public class Journal<K, V> implements Shutdownable
});
}
}
return res;
}
/**
@ -504,11 +510,28 @@ public class Journal<K, V> implements Shutdownable
* @param hosts hosts expected to invalidate the record
*/
public RecordPointer asyncWrite(K id, V record, Set<Integer> hosts)
{
return asyncWrite(id, new SavedCommand.Writer<>()
{
public void write(DataOutputPlus out, int userVersion) throws IOException
{
valueSerializer.serialize(id, record, out, params.userVersion());
}
public K key()
{
return id;
}
},
hosts);
}
public RecordPointer asyncWrite(K id, Writer writer, Set<Integer> hosts)
{
RecordPointer recordPointer;
try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
{
valueSerializer.serialize(id, record, dob, params.userVersion());
writer.write(dob, params.userVersion());
ActiveSegment<K, V>.Allocation alloc = allocate(dob.getLength(), hosts);
recordPointer = alloc.write(id, dob.unsafeGetBufferAndFlip(), hosts);
flusher.asyncFlush(alloc);
@ -521,6 +544,7 @@ public class Journal<K, V> implements Shutdownable
return recordPointer;
}
private ActiveSegment<K, V>.Allocation allocate(int entrySize, Set<Integer> hosts)
{
ActiveSegment<K, V> segment = currentSegment;
@ -942,4 +966,14 @@ public class Journal<K, V> implements Shutdownable
advanceSegment(null);
segments.set(Segments.none());
}
public interface Writer
{
void write(DataOutputPlus out, int userVersion) throws IOException;
}
public interface Reader
{
void read(DataInputBuffer in, int userVersion) throws IOException;
}
}

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

@ -351,7 +351,9 @@ public class AccordCommandStore extends CommandStore implements CacheSize
@VisibleForTesting
public void appendToLog(Command before, Command after, Runnable runnable)
{
journal.appendCommand(id, Collections.singletonList(SavedCommand.SavedDiff.diff(before, after)), null, runnable);
journal.appendCommand(id,
Collections.singletonList(SavedCommand.diffWriter(before, after)),
null, runnable);
}
boolean validateCommand(TxnId txnId, Command evicting)
@ -574,7 +576,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
public NavigableMap<TxnId, Ranges> bootstrapBeganAt() { return super.bootstrapBeganAt(); }
public NavigableMap<Timestamp, Ranges> safeToRead() { return super.safeToRead(); }
public void appendCommands(List<SavedCommand.SavedDiff> commands, List<Command> sanityCheck, Runnable onFlush)
public void appendCommands(List<SavedCommand.Writer<TxnId>> commands, List<Command> sanityCheck, Runnable onFlush)
{
journal.appendCommand(id, commands, sanityCheck, onFlush);
}

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

@ -21,9 +21,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@ -31,23 +29,15 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.primitives.Ints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.Node;
import accord.messages.AbstractEpochRequest;
import accord.messages.Commit;
import accord.messages.LocalRequest;
import accord.messages.Message;
import accord.messages.MessageType;
import accord.messages.ReplyContext;
import accord.messages.Request;
import accord.messages.TxnRequest;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.Long2ObjectHashMap;
@ -57,7 +47,6 @@ import org.apache.cassandra.concurrent.Interruptible;
import org.apache.cassandra.concurrent.ManyToOneConcurrentLinkedQueue;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.File;
@ -69,52 +58,12 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.accord.interop.AccordInteropApply;
import org.apache.cassandra.service.accord.interop.AccordInteropCommit;
import org.apache.cassandra.service.accord.serializers.AcceptSerializers;
import org.apache.cassandra.service.accord.serializers.ApplySerializers;
import org.apache.cassandra.service.accord.serializers.BeginInvalidationSerializers;
import org.apache.cassandra.service.accord.serializers.CommitSerializers;
import org.apache.cassandra.service.accord.serializers.EnumSerializer;
import org.apache.cassandra.service.accord.serializers.FetchSerializers;
import org.apache.cassandra.service.accord.serializers.InformDurableSerializers;
import org.apache.cassandra.service.accord.serializers.InformOfTxnIdSerializers;
import org.apache.cassandra.service.accord.serializers.PreacceptSerializers;
import org.apache.cassandra.service.accord.serializers.RecoverySerializers;
import org.apache.cassandra.service.accord.serializers.SetDurableSerializers;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.concurrent.Condition;
import static accord.messages.MessageType.ACCEPT_INVALIDATE_REQ;
import static accord.messages.MessageType.ACCEPT_REQ;
import static accord.messages.MessageType.APPLY_MAXIMAL_REQ;
import static accord.messages.MessageType.APPLY_MINIMAL_REQ;
import static accord.messages.MessageType.APPLY_THEN_WAIT_UNTIL_APPLIED_REQ;
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_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_OTHER_MSG;
import static accord.messages.MessageType.PROPAGATE_PRE_ACCEPT_MSG;
import static accord.messages.MessageType.PROPAGATE_STABLE_MSG;
import static accord.messages.MessageType.SET_GLOBALLY_DURABLE_REQ;
import static accord.messages.MessageType.SET_SHARD_DURABLE_REQ;
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.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.SAFE;
import static org.apache.cassandra.concurrent.Interruptible.State.NORMAL;
import static org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType.INTEROP_APPLY_MAXIMAL_REQ;
import static org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType.INTEROP_APPLY_MINIMAL_REQ;
import static org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType.INTEROP_COMMIT_MAXIMAL_REQ;
import static org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType.INTEROP_COMMIT_MINIMAL_REQ;
import static org.apache.cassandra.service.accord.serializers.ReadDataSerializers.applyThenWaitUntilApplied;
public class AccordJournal implements IJournal, Shutdownable
{
@ -128,7 +77,7 @@ public class AccordJournal implements IJournal, Shutdownable
private static final Set<Integer> SENTINEL_HOSTS = Collections.singleton(0);
static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[23]);
static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[22]);
public final Journal<JournalKey, Object> journal;
private final AccordEndpointMapper endpointMapper;
@ -144,7 +93,25 @@ public class AccordJournal implements IJournal, Shutdownable
public AccordJournal(AccordEndpointMapper endpointMapper, Params params)
{
File directory = new File(DatabaseDescriptor.getAccordJournalDirectory());
this.journal = new Journal<>("AccordJournal", directory, params, JournalKey.SUPPORT, RECORD_SERIALIZER);
this.journal = new Journal<>("AccordJournal", directory, params, JournalKey.SUPPORT,
// In Accord, we are using streaming serialization, i.e. Reader/Writer interfaces instead of materializing objects
new ValueSerializer<JournalKey, Object>()
{
public int serializedSize(JournalKey key, Object value, int userVersion)
{
throw new UnsupportedOperationException();
}
public void serialize(JournalKey key, Object value, DataOutputPlus out, int userVersion) throws IOException
{
throw new UnsupportedOperationException();
}
public Object deserialize(JournalKey key, DataInputPlus in, int userVersion) throws IOException
{
throw new UnsupportedOperationException();
}
});
this.endpointMapper = endpointMapper;
}
@ -224,26 +191,33 @@ public class AccordJournal implements IJournal, Shutdownable
@Override
public Command loadCommand(int commandStoreId, TxnId txnId)
{
List<SavedCommand.LoadedDiff> diffs = loadDiffs(commandStoreId, txnId);
if (diffs.isEmpty())
return null;
return SavedCommand.reconstructFromDiff(diffs);
try
{
return loadDiffs(commandStoreId, txnId).construct();
}
catch (IOException e)
{
// can only throw if serializer is buggy
throw new RuntimeException(e);
}
}
@VisibleForTesting
public List<SavedCommand.LoadedDiff> loadDiffs(int commandStoreId, Timestamp txnId)
public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId)
{
return (List<SavedCommand.LoadedDiff>)(List<?>) journal.readAll(new JournalKey(txnId, Type.SAVED_COMMAND, commandStoreId));
SavedCommand.Builder builder = new SavedCommand.Builder();
journal.readAll(new JournalKey(txnId, commandStoreId),
builder::deserializeNext);
return builder;
}
@Override
public void appendCommand(int commandStoreId, List<SavedCommand.SavedDiff> outcomes, List<Command> sanityCheck, Runnable onFlush)
public void appendCommand(int commandStoreId, List<SavedCommand.Writer<TxnId>> outcomes, List<Command> sanityCheck, Runnable onFlush)
{
RecordPointer pointer = null;
for (int i = 0; i < outcomes.size(); i++)
for (SavedCommand.Writer<TxnId> outcome : outcomes)
{
SavedCommand.SavedDiff outcome = outcomes.get(i);
JournalKey key = new JournalKey(outcome.txnId, Type.SAVED_COMMAND, commandStoreId);
JournalKey key = new JournalKey(outcome.key(), commandStoreId);
pointer = journal.asyncWrite(key, outcome, SENTINEL_HOSTS);
}
@ -273,14 +247,23 @@ public class AccordJournal implements IJournal, Shutdownable
public void sanityCheck(int commandStoreId, Command orig)
{
List<SavedCommand.LoadedDiff> diffs = loadDiffs(commandStoreId, orig.txnId());
// We can only use strict equality if we supply result.
Command reconstructed = SavedCommand.reconstructFromDiff(diffs, orig.result());
Invariants.checkState(orig.equals(reconstructed),
"\n" +
"Original: %s\n" +
"Reconstructed: %s\n" +
"Diffs: %s", orig, reconstructed, diffs);
try
{
SavedCommand.Builder diffs = loadDiffs(commandStoreId, orig.txnId());
diffs.forceResult(orig.result());
// We can only use strict equality if we supply result.
Command reconstructed = diffs.construct();
Invariants.checkState(orig.equals(reconstructed),
"\n" +
"Original: %s\n" +
"Reconstructed: %s\n" +
"Diffs: %s", orig, reconstructed, diffs);
}
catch (IOException e)
{
// can only throw if serializer is buggy
throw new RuntimeException(e);
}
}
/*
@ -391,241 +374,6 @@ public class AccordJournal implements IJournal, Shutdownable
}
}
/*
* Records ser/de in the Journal
*/
private static final ValueSerializer<JournalKey, Object> RECORD_SERIALIZER = new ValueSerializer<>()
{
@Override
public int serializedSize(JournalKey key, Object record, int userVersion)
{
return Ints.checkedCast(key.type.serializedSize(key, record, userVersion));
}
@Override
public void serialize(JournalKey key, Object record, DataOutputPlus out, int userVersion) throws IOException
{
key.type.serialize(key, record, out, userVersion);
}
@Override
public Object deserialize(JournalKey key, DataInputPlus in, int userVersion) throws IOException
{
return key.type.deserialize(key, in, userVersion);
}
};
/* Adapts vanilla message serializers to journal-expected signatures; converts user version to MS version */
static final class MessageSerializer implements ValueSerializer<JournalKey, Object>
{
final IVersionedSerializer<Message> wrapped;
private MessageSerializer(IVersionedSerializer<Message> wrapped)
{
this.wrapped = wrapped;
}
static MessageSerializer wrap(IVersionedSerializer<Message> wrapped)
{
return new MessageSerializer(wrapped);
}
@Override
public int serializedSize(JournalKey key, Object message, int userVersion)
{
return Ints.checkedCast(wrapped.serializedSize((Message) message, msVersion(userVersion)));
}
@Override
public void serialize(JournalKey key, Object message, DataOutputPlus out, int userVersion) throws IOException
{
wrapped.serialize((Message) message, out, msVersion(userVersion));
}
@Override
public Object deserialize(JournalKey key, DataInputPlus in, int userVersion) throws IOException
{
return wrapped.deserialize(in, msVersion(userVersion));
}
}
@FunctionalInterface
interface TxnIdProvider
{
TxnId txnId(Message message);
}
private static final TxnIdProvider EPOCH = msg -> ((AbstractEpochRequest<?>) msg).txnId;
private static final TxnIdProvider TXN = msg -> ((TxnRequest<?>) msg).txnId;
private static final TxnIdProvider LOCAL = msg -> ((LocalRequest<?>) msg).primaryTxnId();
private static final TxnIdProvider INVL = msg -> ((Commit.Invalidate) msg).primaryTxnId();
/**
* Accord Message type - consequently the kind of persisted record.
* <p>
* Note: {@link EnumSerializer} is intentionally not being reused here, for two reasons:
* 1. This is an internal enum, fully under our control, not part of an external library
* 2. It's persisted in the record key, so has the additional constraint of being fixed size and
* shouldn't be using varint encoding
*/
public enum Type implements ValueSerializer<JournalKey, Object>
{
/* Auxiliary journal records */
SAVED_COMMAND (1, SavedCommand.serializer),
/* Accord protocol requests */
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_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 ),
APPLY_THEN_WAIT_UNTIL_APPLIED (72, APPLY_THEN_WAIT_UNTIL_APPLIED_REQ, applyThenWaitUntilApplied, EPOCH),
BEGIN_RECOVER (73, BEGIN_RECOVER_REQ, RecoverySerializers.request, TXN ),
BEGIN_INVALIDATE (74, BEGIN_INVALIDATE_REQ, BeginInvalidationSerializers.request, EPOCH),
INFORM_OF_TXN (75, INFORM_OF_TXN_REQ, InformOfTxnIdSerializers.request, EPOCH),
INFORM_DURABLE (76, INFORM_DURABLE_REQ, InformDurableSerializers.request, TXN ),
SET_SHARD_DURABLE (77, SET_SHARD_DURABLE_REQ, SetDurableSerializers.shardDurable, EPOCH),
SET_GLOBALLY_DURABLE (78, SET_GLOBALLY_DURABLE_REQ, SetDurableSerializers.globallyDurable, EPOCH),
/* Accord local messages */
PROPAGATE_PRE_ACCEPT (79, PROPAGATE_PRE_ACCEPT_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 (83, INTEROP_COMMIT_MINIMAL_REQ, STABLE_FAST_PATH_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_COMMIT_MAXIMAL (84, INTEROP_COMMIT_MAXIMAL_REQ, STABLE_MAXIMAL_REQ, AccordInteropCommit.serializer, TXN),
INTEROP_APPLY_MINIMAL (85, INTEROP_APPLY_MINIMAL_REQ, APPLY_MINIMAL_REQ, AccordInteropApply.serializer, TXN),
INTEROP_APPLY_MAXIMAL (86, INTEROP_APPLY_MAXIMAL_REQ, APPLY_MAXIMAL_REQ, AccordInteropApply.serializer, TXN),
;
final int id;
/**
* An incoming message of a given type from Accord's perspective might have multiple
* concrete implementations some of which are supplied by the Cassandra integration.
* The incoming type specifies the handling for writing out a message to the journal.
*/
final MessageType incomingType;
/**
* The outgoing type is the type that will be returned to Accord; must be a subclass of the incoming type.
* <p>
* This type will always be from accord.messages.MessageType and never from the extended types in the integration.
*/
final MessageType outgoingType;
final TxnIdProvider txnIdProvider;
final ValueSerializer<JournalKey, Object> serializer;
Type(int id, ValueSerializer<JournalKey, Object> serializer)
{
this(id, null, null, serializer, null);
}
Type(int id, MessageType incomingType, MessageType outgoingType, IVersionedSerializer<?> serializer, TxnIdProvider txnIdProvider)
{
//noinspection unchecked
this(id, incomingType, outgoingType, MessageSerializer.wrap((IVersionedSerializer<Message>) serializer), txnIdProvider);
}
Type(int id, MessageType type, IVersionedSerializer<?> serializer, TxnIdProvider txnIdProvider)
{
//noinspection unchecked
this(id, type, type, MessageSerializer.wrap((IVersionedSerializer<Message>) serializer), txnIdProvider);
}
Type(int id, MessageType incomingType, MessageType outgoingType, ValueSerializer<JournalKey, ?> serializer, TxnIdProvider txnIdProvider)
{
if (id < 0)
throw new IllegalArgumentException("Negative Type id " + id);
if (id > Byte.MAX_VALUE)
throw new IllegalArgumentException("Type id doesn't fit in a single byte: " + id);
this.id = id;
this.incomingType = incomingType;
this.outgoingType = outgoingType;
//noinspection unchecked
this.serializer = (ValueSerializer<JournalKey, Object>) serializer;
this.txnIdProvider = txnIdProvider;
}
private static final Type[] idToTypeMapping;
static
{
Type[] types = values();
int maxId = -1;
for (Type type : types)
maxId = Math.max(type.id, maxId);
Type[] idToType = new Type[maxId + 1];
for (Type type : types)
{
if (null != idToType[type.id])
throw new IllegalStateException("Duplicate Type id " + type.id);
idToType[type.id] = type;
}
idToTypeMapping = idToType;
Map<MessageType, Type> msgTypeToType = new HashMap<>();
for (Type type : types)
{
if (null != type.incomingType && null != msgTypeToType.put(type.incomingType, type))
throw new IllegalStateException("Duplicate MessageType " + type.incomingType);
}
ImmutableMap.copyOf(msgTypeToType);
}
static Type fromId(int id)
{
if (id < 0 || id >= idToTypeMapping.length)
throw new IllegalArgumentException("Out or range Type id " + id);
Type type = idToTypeMapping[id];
if (null == type)
throw new IllegalArgumentException("Unknown Type id " + id);
return type;
}
@Override
public int serializedSize(JournalKey key, Object record, int userVersion)
{
return serializer.serializedSize(key, record, userVersion);
}
@Override
public void serialize(JournalKey key, Object record, DataOutputPlus out, int userVersion) throws IOException
{
serializer.serialize(key, record, out, userVersion);
}
@Override
public Object deserialize(JournalKey key, DataInputPlus in, int userVersion) throws IOException
{
return serializer.deserialize(key, in, userVersion);
}
}
private static int msVersion(int version)
{
switch (version)
{
default: throw new IllegalArgumentException();
case 1: return MessagingService.VERSION_51;
}
}
/*
* Handling topology changes / epoch shift
*/

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

@ -765,6 +765,9 @@ public class AccordKeyspace
public static Mutation getCommandMutation(int storeId, Command original, Command command, long timestampMicros)
{
if (command.saveStatus() == SaveStatus.Uninitialised)
return null;
try
{
Invariants.checkArgument(original != command);

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

@ -299,7 +299,7 @@ public class AccordObjectSizes
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, 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 EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.empty(EMPTY_TXNID.domain()), 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));

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

@ -127,7 +127,7 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
return original;
}
public SavedCommand.SavedDiff diff()
public SavedCommand.Writer<TxnId> diff()
{
return SavedCommand.diff(original, current);
}

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

@ -29,9 +29,9 @@ public interface IJournal
/**
* Append outcomes to the log.
*
* Returns whether an async flush was requested. If it returns false, all commands are guaranteed to be flushed by that time.
* If it returns false, onFlush runnable will run whenever flush is done.
*/
void appendCommand(int commandStoreId, List<SavedCommand.SavedDiff> command, List<Command> sanityCheck, Runnable onFlush);
void appendCommand(int commandStoreId,
List<SavedCommand.Writer<TxnId>> command,
List<Command> sanityCheck,
Runnable onFlush);
}

37
src/java/org/apache/cassandra/service/accord/JournalKey.java
View File

@ -30,7 +30,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.journal.KeySupport;
import org.apache.cassandra.utils.ByteArrayUtil;
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;
import static org.apache.cassandra.db.TypeSizes.SHORT_SIZE;
@ -38,19 +37,18 @@ import static org.apache.cassandra.db.TypeSizes.SHORT_SIZE;
public final class JournalKey
{
final Timestamp timestamp;
final AccordJournal.Type type; // TODO (desired): do we even need type here anymore?
// TODO: command store id _before_ timestamp
final int commandStoreId;
JournalKey(Timestamp timestamp, AccordJournal.Type type)
JournalKey(Timestamp timestamp)
{
this(timestamp, type, -1);
this(timestamp, -1);
}
JournalKey(Timestamp timestamp, AccordJournal.Type type, int commandStoreId)
JournalKey(Timestamp timestamp, int commandStoreId)
{
if (timestamp == null) throw new NullPointerException("Null timestamp for type " + type);
if (timestamp == null) throw new NullPointerException("Null timestamp");
this.timestamp = timestamp;
this.type = type;
this.commandStoreId = commandStoreId;
}
@ -67,8 +65,7 @@ public final class JournalKey
private static final int HLC_OFFSET = 0;
private static final int EPOCH_AND_FLAGS_OFFSET = HLC_OFFSET + LONG_SIZE;
private static final int NODE_OFFSET = EPOCH_AND_FLAGS_OFFSET + LONG_SIZE;
private static final int TYPE_OFFSET = NODE_OFFSET + INT_SIZE;
private static final int CS_ID_OFFSET = TYPE_OFFSET + BYTE_SIZE;
private static final int CS_ID_OFFSET = NODE_OFFSET + INT_SIZE;
@Override
public int serializedSize(int userVersion)
@ -77,7 +74,6 @@ public final class JournalKey
+ 6 // timestamp.epoch()
+ 2 // timestamp.flags()
+ INT_SIZE // timestamp.node
+ BYTE_SIZE // type
+ SHORT_SIZE; // commandStoreId
}
@ -85,33 +81,29 @@ public final class JournalKey
public void serialize(JournalKey key, DataOutputPlus out, int userVersion) throws IOException
{
serializeTimestamp(key.timestamp, out);
out.writeByte(key.type.id);
out.writeShort(key.commandStoreId);
}
private void serialize(JournalKey key, byte[] out)
{
serializeTimestamp(key.timestamp, out);
out[20] = (byte) (key.type.id & 0xFF);
ByteArrayUtil.putShort(out, 21, (short) key.commandStoreId);
ByteArrayUtil.putShort(out, 20, (short) key.commandStoreId);
}
@Override
public JournalKey deserialize(DataInputPlus in, int userVersion) throws IOException
{
Timestamp timestamp = deserializeTimestamp(in);
int type = in.readByte();
int commandStoreId = in.readShort();
return new JournalKey(timestamp, AccordJournal.Type.fromId(type), commandStoreId);
return new JournalKey(timestamp, commandStoreId);
}
@Override
public JournalKey deserialize(ByteBuffer buffer, int position, int userVersion)
{
Timestamp timestamp = deserializeTimestamp(buffer, position);
int type = buffer.get(position + TYPE_OFFSET);
int commandStoreId = buffer.getShort(position + CS_ID_OFFSET);
return new JournalKey(timestamp, AccordJournal.Type.fromId(type), commandStoreId);
return new JournalKey(timestamp, commandStoreId);
}
private void serializeTimestamp(Timestamp timestamp, DataOutputPlus out) throws IOException
@ -158,10 +150,6 @@ public final class JournalKey
int cmp = compareWithTimestampAt(k.timestamp, buffer, position);
if (cmp != 0) return cmp;
byte type = buffer.get(position + TYPE_OFFSET);
cmp = Byte.compare((byte) k.type.id, type);
if (cmp != 0) return cmp;
short commandStoreId = buffer.getShort(position + CS_ID_OFFSET);
cmp = Short.compare((byte) k.commandStoreId, commandStoreId);
return cmp;
@ -186,7 +174,6 @@ public final class JournalKey
public int compare(JournalKey k1, JournalKey k2)
{
int cmp = compare(k1.timestamp, k2.timestamp);
if (cmp == 0) cmp = Byte.compare((byte) k1.type.id, (byte) k2.type.id);
if (cmp == 0) cmp = Short.compare((short) k1.commandStoreId, (short) k2.commandStoreId);
return cmp;
}
@ -225,22 +212,20 @@ public final class JournalKey
boolean equals(JournalKey other)
{
return this.type == other.type &&
this.timestamp.equals(other.timestamp) &&
return this.timestamp.equals(other.timestamp) &&
this.commandStoreId == other.commandStoreId;
}
@Override
public int hashCode()
{
return Objects.hash(timestamp, type, commandStoreId);
return Objects.hash(timestamp, commandStoreId);
}
public String toString()
{
return "Key{" +
"timestamp=" + timestamp +
", type=" + type +
", commandStoreId=" + commandStoreId +
'}';
}

895
src/java/org/apache/cassandra/service/accord/SavedCommand.java
View File

@ -20,9 +20,10 @@ package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.function.Function;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Result;
@ -39,27 +40,25 @@ import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.Invariants;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.journal.ValueSerializer;
import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.DepsSerializer;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer;
import org.apache.cassandra.utils.Throwables;
import static org.apache.cassandra.db.TypeSizes.SHORT_SIZE;
import static accord.utils.Invariants.illegalState;
public class SavedCommand
{
public static final ValueSerializer<JournalKey, Object> serializer = new SavedCommandSerializer();
// This enum is order-dependent
private enum HasFields
public enum Fields
{
TXN_ID,
EXECUTE_AT,
EXECUTES_AT_LEAST,
SAVE_STATUS,
DURABILITY,
ACCEPTED,
@ -73,98 +72,210 @@ public class SavedCommand
LISTENERS
}
public final TxnId txnId;
public final Timestamp executeAt;
public final SaveStatus saveStatus;
public final Status.Durability durability;
public final Ballot acceptedOrCommitted;
public final Ballot promised;
public final Route<?> route;
public final PartialTxn partialTxn;
public final PartialDeps partialDeps;
public final Seekables<?, ?> additionalKeysOrRanges;
public final Writes writes;
public final Listeners.Immutable<Command.DurableAndIdempotentListener> listeners;
public SavedCommand(TxnId txnId,
Timestamp executeAt,
SaveStatus saveStatus,
Status.Durability durability,
Ballot acceptedOrCommitted,
Ballot promised,
Route<?> route,
PartialTxn partialTxn,
PartialDeps partialDeps,
Seekables<?, ?> additionalKeysOrRanges,
Writes writes,
Listeners.Immutable<Command.DurableAndIdempotentListener> listeners)
public interface Writer<K> extends Journal.Writer
{
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
this.durability = durability;
this.acceptedOrCommitted = acceptedOrCommitted;
this.promised = promised;
this.route = route;
this.partialTxn = partialTxn;
this.partialDeps = partialDeps;
this.additionalKeysOrRanges = additionalKeysOrRanges;
this.writes = writes;
this.listeners = listeners;
void write(DataOutputPlus out, int userVersion) throws IOException;
K key();
}
public static SavedDiff diff(Command before, Command after)
public static class DiffWriter implements Writer<TxnId>
{
if (before == after)
private final Command before;
private final Command after;
private final TxnId txnId;
public DiffWriter(Command before, Command after)
{
this(after.txnId(), before, after);
}
public DiffWriter(TxnId txnId, Command before, Command after)
{
this.txnId = txnId;
this.before = before;
this.after = after;
}
@VisibleForTesting
public Command before()
{
return before;
}
@VisibleForTesting
public Command after()
{
return after;
}
public void write(DataOutputPlus out, int userVersion) throws IOException
{
serialize(before, after, out, userVersion);
}
public TxnId key()
{
return txnId;
}
}
@Nullable
public static Writer<TxnId> diff(Command original, Command current)
{
if (original == current
|| current == null
|| current.saveStatus() == SaveStatus.Uninitialised)
return null;
// TODO: we do not need to save `waitingOn` _every_ time.
Command.WaitingOn waitingOn = getWaitingOn(after);
return new SavedDiff(after.txnId(),
ifNotEqual(before, after, Command::executeAt, true),
ifNotEqual(before, after, Command::saveStatus, false),
ifNotEqual(before, after, Command::durability, false),
ifNotEqual(before, after, Command::acceptedOrCommitted, false),
ifNotEqual(before, after, Command::promised, false),
ifNotEqual(before, after, Command::route, true),
ifNotEqual(before, after, Command::partialTxn, false),
ifNotEqual(before, after, Command::partialDeps, false),
ifNotEqual(before, after, Command::additionalKeysOrRanges, false),
waitingOn,
ifNotEqual(before, after, Command::writes, false),
ifNotEqual(before, after, Command::durableListeners, true));
return new SavedCommand.DiffWriter(original, current);
}
static Command reconstructFromDiff(List<LoadedDiff> diffs)
public static Writer<TxnId> diffWriter(Command before, Command after)
{
return reconstructFromDiff(diffs, CommandSerializers.APPLIED);
return new DiffWriter(before, after);
}
public static void serialize(Command before, Command after, DataOutputPlus out, int userVersion) throws IOException
{
int flags = getFlags(before, after);
out.writeInt(flags);
// We encode all changed fields unless their value is null
if (getFieldChanged(Fields.TXN_ID, flags) && after.txnId() != null)
CommandSerializers.txnId.serialize(after.txnId(), out, userVersion);
if (getFieldChanged(Fields.EXECUTE_AT, flags) && after.executeAt() != null)
CommandSerializers.timestamp.serialize(after.executeAt(), out, userVersion);
// TODO (desired): check if this can fold into executeAt
if (getFieldChanged(Fields.EXECUTES_AT_LEAST, flags) && after.executesAtLeast() != null)
CommandSerializers.timestamp.serialize(after.executesAtLeast(), out, userVersion);
if (getFieldChanged(Fields.SAVE_STATUS, flags))
out.writeInt(after.saveStatus().ordinal());
if (getFieldChanged(Fields.DURABILITY, flags) && after.durability() != null)
out.writeInt(after.durability().ordinal());
if (getFieldChanged(Fields.ACCEPTED, flags) && after.acceptedOrCommitted() != null)
CommandSerializers.ballot.serialize(after.acceptedOrCommitted(), out, userVersion);
if (getFieldChanged(Fields.PROMISED, flags) && after.promised() != null)
CommandSerializers.ballot.serialize(after.promised(), out, userVersion);
if (getFieldChanged(Fields.ROUTE, flags) && after.route() != null)
AccordKeyspace.LocalVersionedSerializers.route.serialize(after.route(), out); // TODO (required): user version
if (getFieldChanged(Fields.PARTIAL_TXN, flags) && after.partialTxn() != null)
CommandSerializers.partialTxn.serialize(after.partialTxn(), out, userVersion);
if (getFieldChanged(Fields.PARTIAL_DEPS, flags) && after.partialDeps() != null)
DepsSerializer.partialDeps.serialize(after.partialDeps(), out, userVersion);
if (getFieldChanged(Fields.ADDITIONAL_KEYS, flags) && after.additionalKeysOrRanges() != null)
KeySerializers.seekables.serialize(after.additionalKeysOrRanges(), out, userVersion);
Command.WaitingOn waitingOn = getWaitingOn(after);
if (getFieldChanged(Fields.WAITING_ON, flags) && waitingOn != null)
{
long size = WaitingOnSerializer.serializedSize(waitingOn);
ByteBuffer serialized = WaitingOnSerializer.serialize(after.txnId(), waitingOn);
out.writeInt((int) size);
out.write(serialized);
}
if (getFieldChanged(Fields.WRITES, flags) && after.writes() != null)
CommandSerializers.writes.serialize(after.writes(), out, userVersion);
if (getFieldChanged(Fields.LISTENERS, flags) && after.durableListeners() != null)
{
out.writeByte(after.durableListeners().size());
for (Command.DurableAndIdempotentListener listener : after.durableListeners())
AccordKeyspace.LocalVersionedSerializers.listeners.serialize(listener, out);
}
}
/**
* @param result is exposed because we are _not_ persisting result, since during loading or replay
* we do not expect we will have to send a result to the client, and data results
* can potentially contain a large number of entries, so it's best if they are not
* written into the log.
*/
@VisibleForTesting
static Command reconstructFromDiff(List<LoadedDiff> diffs, Result result)
static int getFlags(Command before, Command after)
{
int flags = 0;
flags = collectFlags(before, after, Command::txnId, true, Fields.TXN_ID, flags);
flags = collectFlags(before, after, Command::executeAt, true, Fields.EXECUTE_AT, flags);
flags = collectFlags(before, after, Command::executesAtLeast, true, Fields.EXECUTES_AT_LEAST, flags);
flags = collectFlags(before, after, Command::saveStatus, false, Fields.SAVE_STATUS, flags);
flags = collectFlags(before, after, Command::durability, false, Fields.DURABILITY, flags);
flags = collectFlags(before, after, Command::acceptedOrCommitted, false, Fields.ACCEPTED, flags);
flags = collectFlags(before, after, Command::promised, false, Fields.PROMISED, flags);
flags = collectFlags(before, after, Command::route, true, Fields.ROUTE, flags);
flags = collectFlags(before, after, Command::partialTxn, false, Fields.PARTIAL_TXN, flags);
flags = collectFlags(before, after, Command::partialDeps, false, Fields.PARTIAL_DEPS, flags);
flags = collectFlags(before, after, Command::additionalKeysOrRanges, false, Fields.ADDITIONAL_KEYS, flags);
flags = collectFlags(before, after, SavedCommand::getWaitingOn, false, Fields.WAITING_ON, flags);
flags = collectFlags(before, after, Command::writes, false, Fields.WRITES, flags);
flags = collectFlags(before, after, c -> c.durableListeners().isEmpty() ? null : c.durableListeners(), true, Fields.LISTENERS, flags);
return flags;
}
static Command.WaitingOn getWaitingOn(Command command)
{
if (command instanceof Command.Committed)
return command.asCommitted().waitingOn();
return null;
}
private static <OBJ, VAL> int collectFlags(OBJ lo, OBJ ro, Function<OBJ, VAL> convert, boolean allowClassMismatch, Fields field, int oldFlags)
{
VAL l = null;
VAL r = null;
if (lo != null) l = convert.apply(lo);
if (ro != null) r = convert.apply(ro);
if (r == null)
oldFlags = setFieldIsNull(field, oldFlags);
if (l == r)
return oldFlags; // no change
if (l == null || r == null)
return setFieldChanged(field, oldFlags);
assert allowClassMismatch || l.getClass() == r.getClass() : String.format("%s != %s", l.getClass(), r.getClass());
if (l.equals(r))
return oldFlags; // no change
return setFieldChanged(field, oldFlags);
}
private static int setFieldChanged(Fields field, int oldFlags)
{
return oldFlags | (1 << (field.ordinal() + Short.SIZE));
}
@VisibleForTesting
static boolean getFieldChanged(Fields field, int oldFlags)
{
return (oldFlags & (1 << (field.ordinal() + Short.SIZE))) != 0;
}
@VisibleForTesting
static boolean getFieldIsNull(Fields field, int oldFlags)
{
return (oldFlags & (1 << field.ordinal())) != 0;
}
private static int setFieldIsNull(Fields field, int oldFlags)
{
return oldFlags | (1 << field.ordinal());
}
public static class Builder
{
TxnId txnId = null;
Timestamp executeAt = null;
Timestamp executeAtLeast = null;
SaveStatus saveStatus = null;
Status.Durability durability = null;
@ -176,148 +287,245 @@ public class SavedCommand
PartialDeps partialDeps = null;
Seekables<?, ?> additionalKeysOrRanges = null;
WaitingOnProvider waitingOnProvider = null;
SavedCommand.WaitingOnProvider waitingOn = (txn, deps) -> null;
Writes writes = null;
Listeners.Immutable listeners = null;
Listeners.Immutable<?> listeners = null;
Result result = CommandSerializers.APPLIED;
for (LoadedDiff diff : diffs)
boolean nextCalled = false;
int count = 0;
public int count()
{
if (diff.txnId != null)
txnId = diff.txnId;
if (diff.executeAt != null)
executeAt = diff.executeAt;
if (diff.saveStatus != null)
saveStatus = diff.saveStatus;
if (diff.durability != null)
durability = diff.durability;
if (diff.acceptedOrCommitted != null)
acceptedOrCommitted = diff.acceptedOrCommitted;
if (diff.promised != null)
promised = diff.promised;
if (diff.route != null)
route = diff.route;
if (diff.partialTxn != null)
partialTxn = diff.partialTxn;
if (diff.partialDeps != null)
partialDeps = diff.partialDeps;
if (diff.additionalKeysOrRanges != null)
additionalKeysOrRanges = diff.additionalKeysOrRanges;
if (diff.waitingOn != null)
waitingOnProvider = diff.waitingOn;
if (diff.writes != null)
writes = diff.writes;
if (diff.listeners != null)
listeners = diff.listeners;
return count;
}
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
if (partialTxn != null)
attrs.partialTxn(partialTxn);
if (durability != null)
attrs.durability(durability);
if (route != null)
attrs.route(route);
if (partialDeps != null &&
(saveStatus.known.deps != Status.KnownDeps.NoDeps &&
saveStatus.known.deps != Status.KnownDeps.DepsErased &&
saveStatus.known.deps != Status.KnownDeps.DepsUnknown))
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
if (listeners != null && !listeners.isEmpty())
attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (waitingOnProvider != null)
waitingOn = waitingOnProvider.provide(txnId, partialDeps);
Invariants.checkState(saveStatus != null,
"Save status is null after applying %s", diffs);
switch (saveStatus.status)
@SuppressWarnings({ "rawtypes", "unchecked" })
public void deserializeNext(DataInputPlus in, int userVersion) throws IOException
{
case NotDefined:
return saveStatus == SaveStatus.Uninitialised ? Command.NotDefined.uninitialised(attrs.txnId())
: Command.NotDefined.notDefined(attrs, promised);
case PreAccepted:
return Command.PreAccepted.preAccepted(attrs, executeAt, promised);
case AcceptedInvalidate:
case Accepted:
case PreCommitted:
return Command.Accepted.accepted(attrs, saveStatus, executeAt, promised, acceptedOrCommitted);
case Committed:
case Stable:
return Command.Committed.committed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn);
case PreApplied:
case Applied:
return Command.Executed.executed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn, writes, result);
case Truncated:
case Invalidated:
default:
throw new IllegalStateException();
}
}
nextCalled = true;
count++;
// TODO (required): this convert function was added only because AsyncOperationTest was failing without it; maybe after switching to loading from the log we can just pass l and r directly or remove != null checks.
private static <OBJ, VAL> VAL ifNotEqual(OBJ lo, OBJ ro, Function<OBJ, VAL> convert, boolean allowClassMismatch)
{
VAL l = null;
VAL r = null;
if (lo != null) l = convert.apply(lo);
if (ro != null) r = convert.apply(ro);
final int flags = in.readInt();
if (l == r)
return null;
if (l == null || r == null)
return r;
assert allowClassMismatch || l.getClass() == r.getClass() : String.format("%s != %s", l.getClass(), r.getClass());
if (getFieldChanged(Fields.TXN_ID, flags))
{
if (getFieldIsNull(Fields.TXN_ID, flags))
txnId = null;
else
txnId = CommandSerializers.txnId.deserialize(in, userVersion);
}
if (l.equals(r))
return null;
if (getFieldChanged(Fields.EXECUTE_AT, flags))
{
if (getFieldIsNull(Fields.EXECUTE_AT, flags))
executeAt = null;
else
executeAt = CommandSerializers.timestamp.deserialize(in, userVersion);
}
return r;
}
if (getFieldChanged(Fields.EXECUTES_AT_LEAST, flags))
{
if (getFieldIsNull(Fields.EXECUTES_AT_LEAST, flags))
executeAtLeast = null;
else
executeAtLeast = CommandSerializers.timestamp.deserialize(in, userVersion);
}
static Command.WaitingOn getWaitingOn(Command command)
{
if (command instanceof Command.Committed)
return command.asCommitted().waitingOn();
if (getFieldChanged(Fields.SAVE_STATUS, flags))
{
if (getFieldIsNull(Fields.SAVE_STATUS, flags))
saveStatus = null;
else
saveStatus = SaveStatus.values()[in.readInt()];
}
if (getFieldChanged(Fields.DURABILITY, flags))
{
if (getFieldIsNull(Fields.DURABILITY, flags))
durability = null;
else
durability = Status.Durability.values()[in.readInt()];
}
return null;
}
if (getFieldChanged(Fields.ACCEPTED, flags))
{
if (getFieldIsNull(Fields.ACCEPTED, flags))
acceptedOrCommitted = null;
else
acceptedOrCommitted = CommandSerializers.ballot.deserialize(in, userVersion);
}
public static class SavedDiff extends SavedCommand
{
public final Command.WaitingOn waitingOn;
if (getFieldChanged(Fields.PROMISED, flags))
{
if (getFieldIsNull(Fields.PROMISED, flags))
promised = null;
else
promised = CommandSerializers.ballot.deserialize(in, userVersion);
}
public SavedDiff(TxnId txnId,
Timestamp executeAt,
SaveStatus saveStatus,
Status.Durability durability,
if (getFieldChanged(Fields.ROUTE, flags))
{
if (getFieldIsNull(Fields.ROUTE, flags))
route = null;
else
route = AccordKeyspace.LocalVersionedSerializers.route.deserialize(in);
}
Ballot acceptedOrCommitted,
Ballot promised,
if (getFieldChanged(Fields.PARTIAL_TXN, flags))
{
if (getFieldIsNull(Fields.PARTIAL_TXN, flags))
partialTxn = null;
else
partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion);
}
Route<?> route,
PartialTxn partialTxn,
PartialDeps partialDeps,
Seekables<?, ?> additionalKeysOrRanges,
if (getFieldChanged(Fields.PARTIAL_DEPS, flags))
{
if (getFieldIsNull(Fields.PARTIAL_DEPS, flags))
partialDeps = null;
else
partialDeps = DepsSerializer.partialDeps.deserialize(in, userVersion);
}
Command.WaitingOn waitingOn,
Writes writes,
Listeners.Immutable<Command.DurableAndIdempotentListener> listeners)
{
super(txnId, executeAt, saveStatus, durability, acceptedOrCommitted, promised, route, partialTxn, partialDeps, additionalKeysOrRanges, writes, listeners);
this.waitingOn = waitingOn;
if (getFieldChanged(Fields.ADDITIONAL_KEYS, flags))
{
if (getFieldIsNull(Fields.ADDITIONAL_KEYS, flags))
additionalKeysOrRanges = null;
else
additionalKeysOrRanges = KeySerializers.seekables.deserialize(in, userVersion);
}
if (getFieldChanged(Fields.WAITING_ON, flags))
{
if (getFieldIsNull(Fields.WAITING_ON, flags))
{
waitingOn = null;
}
else
{
int size = in.readInt();
byte[] bytes = new byte[size];
in.readFully(bytes);
ByteBuffer buffer = ByteBuffer.wrap(bytes);
waitingOn = (localTxnId, deps) -> {
try
{
return WaitingOnSerializer.deserialize(localTxnId, deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, buffer);
}
catch (IOException e)
{
throw Throwables.unchecked(e);
}
};
}
}
if (getFieldChanged(Fields.WRITES, flags))
{
if (getFieldIsNull(Fields.WRITES, flags))
writes = null;
else
writes = CommandSerializers.writes.deserialize(in, userVersion);
}
if (getFieldChanged(Fields.LISTENERS, flags))
{
if (getFieldIsNull(Fields.LISTENERS, flags))
{
listeners = null;
}
else
{
Listeners builder = Listeners.Immutable.EMPTY.mutable();
int cnt = in.readByte();
for (int i = 0; i < cnt; i++)
builder.add(AccordKeyspace.LocalVersionedSerializers.listeners.deserialize(in));
listeners = new Listeners.Immutable(builder);
}
}
}
public void forceResult(Result newValue)
{
this.result = newValue;
}
public Command construct() throws IOException
{
if (!nextCalled)
return null;
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
if (partialTxn != null)
attrs.partialTxn(partialTxn);
if (durability != null)
attrs.durability(durability);
if (route != null)
attrs.route(route);
if (partialDeps != null &&
(saveStatus.known.deps != Status.KnownDeps.NoDeps &&
saveStatus.known.deps != Status.KnownDeps.DepsErased &&
saveStatus.known.deps != Status.KnownDeps.DepsUnknown))
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
if (listeners != null && !listeners.isEmpty())
attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (this.waitingOn != null)
waitingOn = this.waitingOn.provide(txnId, partialDeps);
switch (saveStatus.status)
{
case NotDefined:
return saveStatus == SaveStatus.Uninitialised ? Command.NotDefined.uninitialised(attrs.txnId())
: Command.NotDefined.notDefined(attrs, promised);
case PreAccepted:
return Command.PreAccepted.preAccepted(attrs, executeAt, promised);
case AcceptedInvalidate:
case Accepted:
case PreCommitted:
return Command.Accepted.accepted(attrs, saveStatus, executeAt, promised, acceptedOrCommitted);
case Committed:
case Stable:
return Command.Committed.committed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn);
case PreApplied:
case Applied:
return Command.Executed.executed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn, writes, result);
case Truncated:
case Invalidated:
return truncated(attrs, saveStatus, executeAt, executeAtLeast, writes, result);
default:
throw new IllegalStateException();
}
}
private static Command.Truncated truncated(CommonAttributes.Mutable attrs, SaveStatus status, Timestamp executeAt, Timestamp executesAtLeast, Writes writes, Result result)
{
switch (status)
{
default:
throw illegalState("Unhandled SaveStatus: " + status);
case TruncatedApplyWithOutcome:
case TruncatedApplyWithDeps:
case TruncatedApply:
if (attrs.txnId().kind().awaitsOnlyDeps())
return Command.Truncated.truncatedApply(attrs, status, executeAt, writes, result, executesAtLeast);
return Command.Truncated.truncatedApply(attrs, status, executeAt, writes, result, null);
case ErasedOrInvalidOrVestigial:
return Command.Truncated.erasedOrInvalidOrVestigial(attrs.txnId(), attrs.durability(), attrs.route());
case Erased:
return Command.Truncated.erased(attrs.txnId(), attrs.durability(), attrs.route());
case Invalidated:
return Command.Truncated.invalidated(attrs.txnId(), attrs.durableListeners());
}
}
@Override
public String toString()
{
return "SavedDiff{" +
" txnId=" + txnId +
return "Diff {" +
"txnId=" + txnId +
", executeAt=" + executeAt +
", saveStatus=" + saveStatus +
", durability=" + durability +
@ -326,285 +534,14 @@ public class SavedCommand
", route=" + route +
", partialTxn=" + partialTxn +
", partialDeps=" + partialDeps +
", writes=" + writes +
", additionalKeysOrRanges=" + additionalKeysOrRanges +
", waitingOn=" + waitingOn +
", writes=" + writes +
", listeners=" + listeners +
'}';
}
}
public static class LoadedDiff extends SavedCommand
{
public final WaitingOnProvider waitingOn;
public LoadedDiff(TxnId txnId,
Timestamp executeAt,
SaveStatus saveStatus,
Status.Durability durability,
Ballot acceptedOrCommitted,
Ballot promised,
Route<?> route,
PartialTxn partialTxn,
PartialDeps partialDeps,
Seekables<?, ?> additionalKeysOrRanges,
WaitingOnProvider waitingOn,
Writes writes,
Listeners.Immutable<Command.DurableAndIdempotentListener> listeners)
{
super(txnId, executeAt, saveStatus, durability, acceptedOrCommitted, promised, route, partialTxn, partialDeps, additionalKeysOrRanges, writes, listeners);
this.waitingOn = waitingOn;
}
public String toString()
{
return "LoadedDiff{" +
"waitingOn=" + waitingOn +
'}';
}
}
final static class SavedCommandSerializer implements ValueSerializer<JournalKey, Object>
{
@Override
public int serializedSize(JournalKey key, Object value, int userVersion)
{
SavedDiff diff = (SavedDiff) value;
long size = 0;
size += SHORT_SIZE; // flags
if (diff.txnId != null)
size += CommandSerializers.txnId.serializedSize(diff.txnId, userVersion);
if (diff.executeAt != null)
size += CommandSerializers.timestamp.serializedSize(diff.executeAt, userVersion);
if (diff.saveStatus != null)
size += Integer.BYTES;
if (diff.durability != null)
size += Integer.BYTES;
if (diff.acceptedOrCommitted != null)
size += CommandSerializers.ballot.serializedSize(diff.acceptedOrCommitted, userVersion);
if (diff.promised != null)
size += CommandSerializers.ballot.serializedSize(diff.promised, userVersion);
if (diff.route != null)
size += AccordKeyspace.LocalVersionedSerializers.route.serializedSize(diff.route);
if (diff.partialTxn != null)
CommandSerializers.partialTxn.serializedSize(diff.partialTxn, userVersion);
if (diff.partialDeps != null)
DepsSerializer.partialDeps.serializedSize(diff.partialDeps, userVersion);
if (diff.additionalKeysOrRanges != null)
KeySerializers.seekables.serializedSize(diff.additionalKeysOrRanges, userVersion);
if (diff.waitingOn != null)
{
size += Integer.BYTES;
size += WaitingOnSerializer.serializedSize(diff.waitingOn);
}
if (diff.writes != null)
CommandSerializers.writes.serializedSize(diff.writes, userVersion);
if (diff.listeners != null && !diff.listeners.isEmpty())
{
size += Byte.BYTES;
for (Command.DurableAndIdempotentListener listener : diff.listeners)
size += AccordKeyspace.LocalVersionedSerializers.listeners.serializedSize(listener);
}
return (int) size;
}
@Override
public void serialize(JournalKey key, Object value, DataOutputPlus out, int userVersion) throws IOException
{
SavedDiff diff = (SavedDiff) value;
int flags = getFlags(diff);
out.writeShort(flags);
if (diff.txnId != null)
CommandSerializers.txnId.serialize(diff.txnId, out, userVersion);
if (diff.executeAt != null)
CommandSerializers.timestamp.serialize(diff.executeAt, out, userVersion);
if (diff.saveStatus != null)
out.writeInt(diff.saveStatus.ordinal());
if (diff.durability != null)
out.writeInt(diff.durability.ordinal());
if (diff.acceptedOrCommitted != null)
CommandSerializers.ballot.serialize(diff.acceptedOrCommitted, out, userVersion);
if (diff.promised != null)
CommandSerializers.ballot.serialize(diff.promised, out, userVersion);
if (diff.route != null)
AccordKeyspace.LocalVersionedSerializers.route.serialize(diff.route, out); // TODO (required): user version
if (diff.partialTxn != null)
CommandSerializers.partialTxn.serialize(diff.partialTxn, out, userVersion);
if (diff.partialDeps != null)
DepsSerializer.partialDeps.serialize(diff.partialDeps, out, userVersion);
if (diff.additionalKeysOrRanges != null)
KeySerializers.seekables.serialize(diff.additionalKeysOrRanges, out, userVersion);
if (diff.waitingOn != null)
{
long size = WaitingOnSerializer.serializedSize(diff.waitingOn);
ByteBuffer serialized = WaitingOnSerializer.serialize(diff.txnId, diff.waitingOn);
out.writeInt((int) size);
out.write(serialized);
}
if (diff.writes != null)
CommandSerializers.writes.serialize(diff.writes, out, userVersion);
if (diff.listeners != null && !diff.listeners.isEmpty())
{
out.writeByte(diff.listeners.size());
for (Command.DurableAndIdempotentListener listener : diff.listeners)
AccordKeyspace.LocalVersionedSerializers.listeners.serialize(listener, out);
}
}
private static int getFlags(SavedDiff diff)
{
int flags = 0;
if (diff.txnId != null)
flags = setBit(flags, HasFields.TXN_ID.ordinal());
if (diff.executeAt != null)
flags = setBit(flags, HasFields.EXECUTE_AT.ordinal());
if (diff.saveStatus != null)
flags = setBit(flags, HasFields.SAVE_STATUS.ordinal());
if (diff.durability != null)
flags = setBit(flags, HasFields.DURABILITY.ordinal());
if (diff.acceptedOrCommitted != null)
flags = setBit(flags, HasFields.ACCEPTED.ordinal());
if (diff.promised != null)
flags = setBit(flags, HasFields.PROMISED.ordinal());
if (diff.route != null)
flags = setBit(flags, HasFields.ROUTE.ordinal());
if (diff.partialTxn != null)
flags = setBit(flags, HasFields.PARTIAL_TXN.ordinal());
if (diff.partialDeps != null)
flags = setBit(flags, HasFields.PARTIAL_DEPS.ordinal());
if (diff.additionalKeysOrRanges != null)
flags = setBit(flags, HasFields.ADDITIONAL_KEYS.ordinal());
if (diff.waitingOn != null)
flags = setBit(flags, HasFields.WAITING_ON.ordinal());
if (diff.writes != null)
flags = setBit(flags, HasFields.WRITES.ordinal());
if (diff.listeners != null && !diff.listeners.isEmpty())
flags = setBit(flags, HasFields.LISTENERS.ordinal());
return flags;
}
@Override
public Object deserialize(JournalKey key, DataInputPlus in, int userVersion) throws IOException
{
int flags = in.readShort();
TxnId txnId = null;
Timestamp executedAt = null;
SaveStatus saveStatus = null;
Status.Durability durability = null;
Ballot acceptedOrCommitted = null;
Ballot promised = null;
Route<?> route = null;
PartialTxn partialTxn = null;
PartialDeps partialDeps = null;
Seekables<?, ?> additionalKeysOrRanges = null;
WaitingOnProvider waitingOn = (txn, deps) -> null;
Writes writes = null;
Listeners.Immutable listeners = null;
if (isSet(flags, HasFields.TXN_ID.ordinal()))
txnId = CommandSerializers.txnId.deserialize(in, userVersion);
if (isSet(flags, HasFields.EXECUTE_AT.ordinal()))
executedAt = CommandSerializers.timestamp.deserialize(in, userVersion);
if (isSet(flags, HasFields.SAVE_STATUS.ordinal()))
saveStatus = SaveStatus.values()[in.readInt()];
if (isSet(flags, HasFields.DURABILITY.ordinal()))
durability = Status.Durability.values()[in.readInt()];
if (isSet(flags, HasFields.ACCEPTED.ordinal()))
acceptedOrCommitted = CommandSerializers.ballot.deserialize(in, userVersion);
if (isSet(flags, HasFields.PROMISED.ordinal()))
promised = CommandSerializers.ballot.deserialize(in, userVersion);
if (isSet(flags, HasFields.ROUTE.ordinal()))
route = AccordKeyspace.LocalVersionedSerializers.route.deserialize(in);
if (isSet(flags, HasFields.PARTIAL_TXN.ordinal()))
partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion);
if (isSet(flags, HasFields.PARTIAL_DEPS.ordinal()))
partialDeps = DepsSerializer.partialDeps.deserialize(in, userVersion);
if (isSet(flags, HasFields.ADDITIONAL_KEYS.ordinal()))
additionalKeysOrRanges = KeySerializers.seekables.deserialize(in, userVersion);
if (isSet(flags, HasFields.WAITING_ON.ordinal()))
{
int size = in.readInt();
byte[] bytes = new byte[size];
in.readFully(bytes);
ByteBuffer buffer = ByteBuffer.wrap(bytes);
waitingOn = (localTxnId, deps) -> {
try
{
return WaitingOnSerializer.deserialize(localTxnId, deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, buffer);
}
catch (IOException e)
{
throw Throwables.unchecked(e);
}
};
}
if (isSet(flags, HasFields.WRITES.ordinal()))
writes = CommandSerializers.writes.deserialize(in, userVersion);
if (isSet(flags, HasFields.LISTENERS.ordinal()))
{
Listeners builder = Listeners.Immutable.EMPTY.mutable();
int cnt = in.readByte();
for (int i = 0; i < cnt; i++)
builder.add(AccordKeyspace.LocalVersionedSerializers.listeners.deserialize(in));
listeners = new Listeners.Immutable(builder);
}
return new LoadedDiff(txnId,
executedAt,
saveStatus,
durability,
acceptedOrCommitted,
promised,
route,
partialTxn,
partialDeps,
additionalKeysOrRanges,
waitingOn,
writes,
listeners);
}
}
static int setBit(int value, int bit)
{
return value | (1 << bit);
}
static boolean isSet(int value, int bit)
{
return (value & (1 << bit)) != 0;
}
public interface WaitingOnProvider
{
Command.WaitingOn provide(TxnId txnId, PartialDeps deps);

27
src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
View File

@ -251,34 +251,31 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
result = apply(safeStore);
// TODO (required): currently, we are not very efficient about ensuring that we persist the absolute minimum amount of state. Improve that.
List<SavedCommand.SavedDiff> diffs = null;
List<SavedCommand.Writer<TxnId>> diffs = null;
for (AccordSafeCommand commandState : context.commands.values())
{
SavedCommand.SavedDiff diff = commandState.diff();
if (diff != null)
SavedCommand.Writer<TxnId> diff = commandState.diff();
if (diff == null)
continue;
if (diffs == null)
diffs = new ArrayList<>(context.commands.size());
diffs.add(diff);
if (CassandraRelevantProperties.DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED.getBoolean())
{
if (diffs == null)
diffs = new ArrayList<>(context.commands.size());
diffs.add(diff);
if (CassandraRelevantProperties.DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED.getBoolean())
{
if (sanityCheck == null)
sanityCheck = new ArrayList<>(context.commands.size());
sanityCheck.add(commandState.current());
}
if (sanityCheck == null)
sanityCheck = new ArrayList<>(context.commands.size());
sanityCheck.add(commandState.current());
}
}
commandStore.completeOperation(safeStore);
context.releaseResources(commandStore);
state(COMPLETING);
if (diffs != null)
{
state(COMPLETING);
this.commandStore.appendCommands(diffs, sanityCheck, () -> finish(result, null));
return false;
}
state(COMPLETING);
case COMPLETING:
finish(result, null);
case FINISHED:

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

@ -92,7 +92,7 @@ public class WaitingOnSerializer
int a = VIntCoding.readUnsignedVInt32(in, position);
position += TypeSizes.sizeofUnsignedVInt(a);
int b = VIntCoding.readUnsignedVInt32(in, position);
position += TypeSizes.sizeofUnsignedVInt(a);
position += TypeSizes.sizeofUnsignedVInt(b);
ImmutableBitSet waitingOn = deserialize(position, waitingOnLength, in);
ImmutableBitSet appliedOrInvalidated = null;
if (txnId.domain() == Routable.Domain.Range)
@ -110,7 +110,7 @@ public class WaitingOnSerializer
for (int i = 0 ; i < length ; ++i)
{
bits[i] = in.getLong(position);
position += 8;
position += Long.BYTES;
}
return ImmutableBitSet.SerializationSupport.construct(bits);
}

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

@ -118,7 +118,7 @@ public class CompactionAccordIteratorsTest
private static final TxnId LT_TXN_ID = AccordTestUtils.txnId(EPOCH, HLC_START, NODE);
private static final TxnId TXN_ID = AccordTestUtils.txnId(EPOCH, LT_TXN_ID.hlc() + 1, NODE);
private static final TxnId SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 1, NODE, Kind.Read);
private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 1, NODE, Kind.Read, Routable.Domain.Range);
private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 2, NODE, Kind.Read, Routable.Domain.Range);
private static final TxnId GT_TXN_ID = SECOND_TXN_ID;
// For CommandsForKey where we test with two commands
private static final TxnId[] TXN_IDS = new TxnId[]{ TXN_ID, SECOND_TXN_ID };
@ -380,10 +380,10 @@ public class CompactionAccordIteratorsTest
};
}
private static RedundantBefore redundantBefore(TxnId txnId)
{
Ranges ranges = AccordTestUtils.fullRange(AccordTestUtils.keys(table, 42));
txnId = txnId.as(Kind.Read, Routable.Domain.Range);
return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, LT_TXN_ID);
}

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

@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
@ -75,22 +74,10 @@ public class AccordJournalOrderTest
for (int i = 0; i < 10_000; i++)
{
TxnId txnId = randomSource.nextBoolean() ? id1 : id2;
JournalKey key = new JournalKey(txnId, AccordJournal.Type.SAVED_COMMAND, randomSource.nextInt(5));
JournalKey key = new JournalKey(txnId, randomSource.nextInt(5));
res.compute(key, (k, prev) -> prev == null ? 1 : prev + 1);
accordJournal.appendCommand(key.commandStoreId,
Collections.singletonList(new SavedCommand.SavedDiff(txnId,
AccordGens.timestamps().next(randomSource),
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null)),
Collections.singletonList(new SavedCommand.DiffWriter(txnId, null, null)),
null,
() -> {});
}
@ -98,8 +85,9 @@ public class AccordJournalOrderTest
Runnable check = () -> {
for (JournalKey key : res.keySet())
{
List<SavedCommand.LoadedDiff> diffs = accordJournal.loadDiffs(key.commandStoreId, key.timestamp);
Assert.assertEquals(diffs.size(), res.get(key).intValue());
SavedCommand.Builder diffs = accordJournal.loadDiffs(key.commandStoreId, (TxnId) key.timestamp);
Assert.assertEquals(String.format("%d != %d for key %s", diffs.count(), res.get(key).intValue(), key),
diffs.count(), res.get(key).intValue());
}
};

3
test/unit/org/apache/cassandra/service/accord/AccordJournalTest.java
View File

@ -122,7 +122,6 @@ public class AccordJournalTest
private Gen<JournalKey> keyGen()
{
Gen<TxnId> txnIdGen = AccordGens.txnIds();
Gen<AccordJournal.Type> typeGen = Gens.enums().all(AccordJournal.Type.class);
return rs -> new JournalKey(txnIdGen.next(rs), typeGen.next(rs));
return rs -> new JournalKey(txnIdGen.next(rs));
}
}

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

@ -152,7 +152,7 @@ public class AccordTestUtils
executeAt,
Ballot.ZERO,
Ballot.ZERO,
Command.WaitingOn.EMPTY);
Command.WaitingOn.empty(txnId.domain()));
}
private static FullRoute<?> route(PartialTxn txn)
@ -517,12 +517,10 @@ public class AccordTestUtils
public static void appendCommandsBlocking(AccordCommandStore commandStore, Command before, Command after)
{
SavedCommand.SavedDiff diff = SavedCommand.diff(before, after);
if (diff != null)
{
Condition condition = Condition.newOneTimeCondition();
commandStore.appendCommands(Collections.singletonList(diff), null, condition::signal);
condition.awaitUninterruptibly(30, TimeUnit.SECONDS);
}
SavedCommand.Writer<TxnId> diff = SavedCommand.diff(before, after);
if (diff == null) return;
Condition condition = Condition.newOneTimeCondition();
commandStore.appendCommands(Collections.singletonList(diff), null, condition::signal);
condition.awaitUninterruptibly(30, TimeUnit.SECONDS);
}
}

326
test/unit/org/apache/cassandra/service/accord/MockJournal.java
View File

@ -18,52 +18,324 @@
package org.apache.cassandra.service.accord;
import java.util.List;
import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Result;
import accord.local.Command;
import accord.local.CommonAttributes;
import accord.local.Listeners;
import accord.local.SaveStatus;
import accord.local.Status;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.Invariants;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import accord.local.Command;
import accord.primitives.TxnId;
import org.apache.cassandra.service.accord.AccordJournal.Type;
public class MockJournal implements IJournal
{
private final Map<JournalKey, List<SavedCommand.LoadedDiff>> commands = new HashMap<>();
private final Map<JournalKey, List<LoadedDiff>> commands = new HashMap<>();
@Override
public Command loadCommand(int commandStoreId, TxnId txnId)
{
Type type = Type.SAVED_COMMAND;
JournalKey key = new JournalKey(txnId, type, commandStoreId);
List<SavedCommand.LoadedDiff> saved = commands.get(key);
JournalKey key = new JournalKey(txnId, commandStoreId);
List<LoadedDiff> saved = commands.get(key);
if (saved == null)
return null;
return SavedCommand.reconstructFromDiff(new ArrayList<>(saved));
return reconstructFromDiff(new ArrayList<>(saved));
}
@Override
public void appendCommand(int commandStoreId, List<SavedCommand.SavedDiff> diffs, List<Command> sanityCheck, Runnable onFlush)
public void appendCommand(int commandStoreId, List<SavedCommand.Writer<TxnId>> diffs, List<Command> sanityCheck, Runnable onFlush)
{
Type type = Type.SAVED_COMMAND;
for (SavedCommand.SavedDiff diff : diffs)
for (SavedCommand.Writer<TxnId> diff : diffs)
{
JournalKey key = new JournalKey(diff.txnId, type, commandStoreId);
SavedCommand.DiffWriter writer = (SavedCommand.DiffWriter) diff;
JournalKey key = new JournalKey(diff.key(), commandStoreId);
commands.computeIfAbsent(key, (ignore_) -> new ArrayList<>())
.add(new SavedCommand.LoadedDiff(diff.txnId,
diff.executeAt,
diff.saveStatus,
diff.durability,
diff.acceptedOrCommitted,
diff.promised,
diff.route,
diff.partialTxn,
diff.partialDeps,
diff.additionalKeysOrRanges,
(i1, i2) -> diff.waitingOn,
diff.writes,
diff.listeners));
.add(diff(writer.before(), writer.after()));
}
onFlush.run();
}
/**
* Emulating journal behaviour
*/
public static LoadedDiff diff(Command before, Command after)
{
if (before == after)
return null;
// TODO: we do not need to save `waitingOn` _every_ time.
Command.WaitingOn waitingOn = getWaitingOn(after);
return new LoadedDiff(after.txnId(),
ifNotEqual(before, after, Command::executeAt, true),
ifNotEqual(before, after, Command::saveStatus, false),
ifNotEqual(before, after, Command::durability, false),
ifNotEqual(before, after, Command::acceptedOrCommitted, false),
ifNotEqual(before, after, Command::promised, false),
ifNotEqual(before, after, Command::route, true),
ifNotEqual(before, after, Command::partialTxn, false),
ifNotEqual(before, after, Command::partialDeps, false),
ifNotEqual(before, after, Command::additionalKeysOrRanges, false),
new NewValue<>((k, deps) -> waitingOn),
ifNotEqual(before, after, Command::writes, false),
ifNotEqual(before, after, Command::durableListeners, true));
}
static Command reconstructFromDiff(List<LoadedDiff> diffs)
{
return reconstructFromDiff(diffs, CommandSerializers.APPLIED);
}
/**
* @param result is exposed because we are _not_ persisting result, since during loading or replay
* we do not expect we will have to send a result to the client, and data results
* can potentially contain a large number of entries, so it's best if they are not
* written into the log.
*/
@VisibleForTesting
static Command reconstructFromDiff(List<LoadedDiff> diffs, Result result)
{
TxnId txnId = null;
Timestamp executeAt = null;
SaveStatus saveStatus = null;
Status.Durability durability = null;
Ballot acceptedOrCommitted = Ballot.ZERO;
Ballot promised = null;
Route<?> route = null;
PartialTxn partialTxn = null;
PartialDeps partialDeps = null;
Seekables<?, ?> additionalKeysOrRanges = null;
SavedCommand.WaitingOnProvider waitingOnProvider = null;
Writes writes = null;
Listeners.Immutable listeners = null;
for (LoadedDiff diff : diffs)
{
if (diff.txnId != null)
txnId = diff.txnId;
if (diff.executeAt != null)
executeAt = diff.executeAt.get();
if (diff.saveStatus != null)
saveStatus = diff.saveStatus.get();
if (diff.durability != null)
durability = diff.durability.get();
if (diff.acceptedOrCommitted != null)
acceptedOrCommitted = diff.acceptedOrCommitted.get();
if (diff.promised != null)
promised = diff.promised.get();
if (diff.route != null)
route = diff.route.get();
if (diff.partialTxn != null)
partialTxn = diff.partialTxn.get();
if (diff.partialDeps != null)
partialDeps = diff.partialDeps.get();
if (diff.additionalKeysOrRanges != null)
additionalKeysOrRanges = diff.additionalKeysOrRanges.get();
if (diff.waitingOn != null)
waitingOnProvider = diff.waitingOn.get();
if (diff.writes != null)
writes = diff.writes.get();
if (diff.listeners != null)
listeners = diff.listeners.get();
}
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
if (partialTxn != null)
attrs.partialTxn(partialTxn);
if (durability != null)
attrs.durability(durability);
if (route != null)
attrs.route(route);
if (partialDeps != null &&
(saveStatus.known.deps != Status.KnownDeps.NoDeps &&
saveStatus.known.deps != Status.KnownDeps.DepsErased &&
saveStatus.known.deps != Status.KnownDeps.DepsUnknown))
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
if (listeners != null && !listeners.isEmpty())
attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (waitingOnProvider != null)
waitingOn = waitingOnProvider.provide(txnId, partialDeps);
Invariants.checkState(saveStatus != null,
"Save status is null after applying %s", diffs);
switch (saveStatus.status)
{
case NotDefined:
return saveStatus == SaveStatus.Uninitialised ? Command.NotDefined.uninitialised(attrs.txnId())
: Command.NotDefined.notDefined(attrs, promised);
case PreAccepted:
return Command.PreAccepted.preAccepted(attrs, executeAt, promised);
case AcceptedInvalidate:
case Accepted:
case PreCommitted:
return Command.Accepted.accepted(attrs, saveStatus, executeAt, promised, acceptedOrCommitted);
case Committed:
case Stable:
return Command.Committed.committed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn);
case PreApplied:
case Applied:
return Command.Executed.executed(attrs, saveStatus, executeAt, promised, acceptedOrCommitted, waitingOn, writes, result);
case Truncated:
case Invalidated:
default:
throw new IllegalStateException();
}
}
// TODO (required): this convert function was added only because AsyncOperationTest was failing without it;
// maybe after switching to loading from the log we can just pass l and r directly or remove != null checks.
private static <OBJ, VAL> NewValue<VAL> ifNotEqual(OBJ lo, OBJ ro, Function<OBJ, VAL> convert, boolean allowClassMismatch)
{
VAL l = null;
VAL r = null;
if (lo != null) l = convert.apply(lo);
if (ro != null) r = convert.apply(ro);
if (l == r)
return null; // null here means there was no change
if (l == null || r == null)
return NewValue.of(r);
assert allowClassMismatch || l.getClass() == r.getClass() : String.format("%s != %s", l.getClass(), r.getClass());
if (l.equals(r))
return null;
return NewValue.of(r);
}
public static class NewValue<T>
{
final T value;
private NewValue(T value)
{
this.value = value;
}
public boolean isNull()
{
return value == null;
}
public T get()
{
return value;
}
public static <T> NewValue<T> of(T value)
{
return new NewValue<>(value);
}
public String toString()
{
return "" + value;
}
}
static Command.WaitingOn getWaitingOn(Command command)
{
if (command instanceof Command.Committed)
return command.asCommitted().waitingOn();
return null;
}
public static class LoadedDiff extends SavedCommand
{
public final TxnId txnId;
public final NewValue<Timestamp> executeAt;
public final NewValue<SaveStatus> saveStatus;
public final NewValue<Status.Durability> durability;
public final NewValue<Ballot> acceptedOrCommitted;
public final NewValue<Ballot> promised;
public final NewValue<Route<?>> route;
public final NewValue<PartialTxn> partialTxn;
public final NewValue<PartialDeps> partialDeps;
public final NewValue<Seekables<?, ?>> additionalKeysOrRanges;
public final NewValue<Writes> writes;
public final NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners;
public final NewValue<WaitingOnProvider> waitingOn;
public LoadedDiff(TxnId txnId,
NewValue<Timestamp> executeAt,
NewValue<SaveStatus> saveStatus,
NewValue<Status.Durability> durability,
NewValue<Ballot> acceptedOrCommitted,
NewValue<Ballot> promised,
NewValue<Route<?>> route,
NewValue<PartialTxn> partialTxn,
NewValue<PartialDeps> partialDeps,
NewValue<Seekables<?, ?>> additionalKeysOrRanges,
NewValue<SavedCommand.WaitingOnProvider> waitingOn,
NewValue<Writes> writes,
NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners)
{
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
this.durability = durability;
this.acceptedOrCommitted = acceptedOrCommitted;
this.promised = promised;
this.route = route;
this.partialTxn = partialTxn;
this.partialDeps = partialDeps;
this.additionalKeysOrRanges = additionalKeysOrRanges;
this.writes = writes;
this.listeners = listeners;
this.waitingOn = waitingOn;
}
public String toString()
{
return "LoadedDiff{" +
"waitingOn=" + waitingOn +
'}';
}
}
}

144
test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java Normal file
View File

@ -0,0 +1,144 @@
/*
* 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;
import java.util.EnumSet;
import java.util.Set;
import com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.local.Command;
import accord.local.SaveStatus;
import accord.primitives.TxnId;
import accord.utils.Gen;
import accord.utils.LazyToString;
import accord.utils.ReflectionUtils;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.SavedCommand.Fields;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.AccordGenerators;
import org.assertj.core.api.SoftAssertions;
import static accord.utils.Property.qt;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
import static org.apache.cassandra.service.accord.SavedCommand.getFlags;
public class SavedCommandTest
{
private static final EnumSet<Fields> ALL = EnumSet.allOf(Fields.class);
@BeforeClass
public static void beforeClass() throws Throwable
{
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1),
parse("CREATE TABLE tbl (k int, c int, v int, primary key (k, c)) WITH transactional_mode='full'", "ks"));
TableMetadata tbl = Schema.instance.getTableMetadata("ks", "tbl");
Assert.assertEquals(TransactionalMode.full, tbl.params.transactionalMode);
StorageService.instance.initServer();
}
@Test
public void allNull()
{
int flags = getFlags(null, null);
assertMissing(flags, ALL);
}
@Test
public void simpleNullChangeCheck()
{
int flags = getFlags(null, Command.NotDefined.uninitialised(TxnId.NONE));
EnumSet<Fields> has = EnumSet.of(Fields.TXN_ID, Fields.SAVE_STATUS, Fields.DURABILITY, Fields.PROMISED,
Fields.ACCEPTED /* this is Zero... which kinda means null... */);
Set<Fields> missing = Sets.difference(ALL, has);
assertHas(flags, has);
assertMissing(flags, missing);
}
@Test
public void serde()
{
Gen<AccordGenerators.CommandBuilder> gen = AccordGenerators.commandsBuilder();
try (DataOutputBuffer out = new DataOutputBuffer())
{
qt().forAll(gen).check(cmdBuilder -> {
int userVersion = 1; //TODO (maintance): where can we fetch all supported versions?
SoftAssertions checks = new SoftAssertions();
for (SaveStatus saveStatus : SaveStatus.values())
{
if (saveStatus == SaveStatus.TruncatedApplyWithDeps) continue;
out.clear();
Command orig = cmdBuilder.build(saveStatus);
SavedCommand.serialize(null, orig, out, userVersion);
SavedCommand.Builder builder = new SavedCommand.Builder();
builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), userVersion);
// We are not persisting the result, so force it for strict equality
builder.forceResult(orig.result());
Command reconstructed = builder.construct();
checks.assertThat(reconstructed)
.describedAs("lhs=expected\nrhs=actual\n%s", new LazyToString(() -> ReflectionUtils.recursiveEquals(orig, reconstructed).toString()))
.isEqualTo(orig);
}
checks.assertAll();
});
}
}
private void assertHas(int flags, Set<Fields> missing)
{
SoftAssertions checks = new SoftAssertions();
for (Fields field : missing)
{
checks.assertThat(SavedCommand.getFieldChanged(field, flags))
.describedAs("field %s changed", field).
isTrue();
checks.assertThat(SavedCommand.getFieldIsNull(field, flags))
.describedAs("field %s not null", field)
.isFalse();
}
checks.assertAll();
}
private void assertMissing(int flags, Set<Fields> missing)
{
SoftAssertions checks = new SoftAssertions();
for (Fields field : missing)
{
checks.assertThat(SavedCommand.getFieldChanged(field, flags))
.describedAs("field %s changed", field)
.isFalse();
checks.assertThat(SavedCommand.getFieldIsNull(field, flags))
.describedAs("field %s not null", field)
.isTrue();
}
checks.assertAll();
}
}

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

@ -143,6 +143,21 @@ public class AsyncOperationTest
Assert.assertTrue(result.isEmpty());
}
@Test
public void touchUnknownTxn() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getUninterruptibly(commandStore.execute(contextFor(txnId), safe -> {
SafeCommand command = safe.get(txnId, txnId, safe.ranges().currentRanges());
Assert.assertNotNull(command);
}));
UntypedResultSet result = AccordKeyspace.loadCommandRow(commandStore, txnId);
Assert.assertTrue(result.isEmpty());
}
@Test
public void optionalCommandsForKeyTest() throws Throwable
{

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

@ -166,12 +166,12 @@ public class CommandsForKeySerializerTest
case Stable:
case ReadyToExecute:
return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.EMPTY);
return Command.SerializerSupport.committed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.empty(txnId.domain()));
case PreApplied:
case Applying:
case Applied:
return Command.SerializerSupport.executed(attributes(), saveStatus, executeAt, ballot, ballot, Command.WaitingOn.EMPTY, new Writes(txnId, executeAt, txn.keys(), new TxnWrite(Collections.emptyList(), true)), new TxnData());
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());
case TruncatedApplyWithDeps:
case TruncatedApply:

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

@ -73,7 +73,7 @@ public class WaitingOnSerializerTest
Gen<WaitingOnSets> sets = Gens.enums().all(WaitingOnSets.class);
return rs -> {
Deps deps = depsGen.next(rs);
if (deps.isEmpty()) return Command.WaitingOn.EMPTY;
if (deps.isEmpty()) return Command.WaitingOn.empty(Routable.Domain.Key);
int txnIdCount = deps.rangeDeps.txnIdCount() + deps.directKeyDeps.txnIdCount();
int keyCount = deps.keyDeps.keys().size();
int[] selected = Gens.arrays(Gens.ints().between(0, txnIdCount + keyCount - 1)).unique().ofSizeBetween(0, txnIdCount + keyCount).next(rs);

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

@ -20,6 +20,7 @@ package org.apache.cassandra.utils;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@ -27,22 +28,30 @@ import java.util.function.BiFunction;
import java.util.stream.Stream;
import accord.local.Command;
import accord.local.CommonAttributes;
import accord.local.Listeners;
import accord.local.RedundantBefore;
import accord.local.SaveStatus;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.KeyDeps;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Range;
import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.RandomSource;
import accord.utils.TriFunction;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.AccordSplitter;
import org.apache.cassandra.dht.IPartitioner;
@ -52,8 +61,11 @@ import org.apache.cassandra.service.accord.AccordTestUtils;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.txn.TxnData;
import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.quicktheories.impl.JavaRandom;
import static accord.local.Status.Durability.NotDurable;
import static org.apache.cassandra.service.accord.AccordTestUtils.TABLE_ID1;
import static org.apache.cassandra.service.accord.AccordTestUtils.createPartialTxn;
@ -71,7 +83,7 @@ public class AccordGenerators
}
private enum SupportedCommandTypes
{notDefined, preaccepted, committed}
{notDefined, preaccepted, committed, stable}
public static Gen<Command> commands()
{
@ -81,13 +93,18 @@ public class AccordGenerators
//TODO goes against fuzz testing, and also limits to a very specific table existing...
// There is a branch that can generate random transactions, so maybe look into that?
PartialTxn txn = createPartialTxn(0);
FullRoute<?> route = txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null));
return rs -> {
TxnId id = ids.next(rs);
Timestamp executeAt = id;
TxnId executeAt = id;
if (rs.nextBoolean())
executeAt = ids.next(rs);
if (executeAt.compareTo(id) < 0)
{
TxnId tmp = id;
id = executeAt;
executeAt = tmp;
}
SupportedCommandTypes targetType = supportedTypes.next(rs);
switch (targetType)
{
@ -97,12 +114,159 @@ public class AccordGenerators
return AccordTestUtils.Commands.preaccepted(id, txn, executeAt);
case committed:
return AccordTestUtils.Commands.committed(id, txn, executeAt);
case stable:
return AccordTestUtils.Commands.stable(id, txn, executeAt);
default:
throw new UnsupportedOperationException("Unexpected type: " + targetType);
}
};
}
public enum RecoveryStatus { None, Started, Complete }
public static Gen<CommandBuilder> commandsBuilder()
{
return commandsBuilder(AccordGens.txnIds(), Gens.bools().all(), Gens.enums().all(RecoveryStatus.class), (rs, txnId, txn) -> AccordGens.depsFor(txnId, txn).next(rs));
}
public static Gen<CommandBuilder> commandsBuilder(Gen<TxnId> txnIdGen, Gen<Boolean> fastPath, Gen<RecoveryStatus> recover, TriFunction<RandomSource, TxnId, Txn, Deps> depsGen)
{
return rs -> {
TxnId txnId = txnIdGen.next(rs);
Txn txn = AccordTestUtils.createTxn(0, 0);
Deps deps = depsGen.apply(rs, txnId, txn);
Timestamp executeAt = fastPath.next(rs) ? txnId
: AccordGens.timestamps(AccordGens.epochs(txnId.epoch()),
AccordGens.hlcs(txnId.hlc()),
AccordGens.flags(),
RandomSource::nextInt).next(rs);
Ranges slice = AccordTestUtils.fullRange(txn);
PartialTxn partialTxn = txn.slice(slice, true); //TODO (correctness): find the case where includeQuery=false and replicate
PartialDeps partialDeps = deps.intersecting(slice);
Ballot promised;
Ballot accepted;
switch (recover.next(rs))
{
case None:
{
promised = Ballot.ZERO;
accepted = Ballot.ZERO;
}
break;
case Started:
{
promised = AccordGens.ballot(AccordGens.epochs(executeAt.epoch()),
AccordGens.hlcs(executeAt.hlc()),
AccordGens.flags(),
RandomSource::nextInt).next(rs);
accepted = Ballot.ZERO;
}
break;
case Complete:
{
promised = accepted = AccordGens.ballot(AccordGens.epochs(executeAt.epoch()),
AccordGens.hlcs(executeAt.hlc()),
AccordGens.flags(),
RandomSource::nextInt).next(rs);
}
break;
default:
throw new UnsupportedOperationException();
}
Command.WaitingOn waitingOn = Command.WaitingOn.none(txnId.domain(), deps);
return new CommandBuilder(txnId, txn, executeAt, partialTxn, partialDeps, promised, accepted, waitingOn);
};
}
public static class CommandBuilder
{
public final TxnId txnId;
public final FullRoute<?> route;
public final Seekables<?, ?> keysOrRanges;
private final Timestamp executeAt;
private final PartialTxn partialTxn;
private final PartialDeps partialDeps;
private final Ballot promised, accepted;
private final Command.WaitingOn waitingOn;
public CommandBuilder(TxnId txnId, Txn txn, Timestamp executeAt, PartialTxn partialTxn, PartialDeps partialDeps, Ballot promised, Ballot accepted, Command.WaitingOn waitingOn)
{
this.txnId = txnId;
this.executeAt = executeAt;
this.partialTxn = partialTxn;
this.partialDeps = partialDeps;
this.promised = promised;
this.accepted = accepted;
this.waitingOn = waitingOn;
this.route = txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null));
this.keysOrRanges = txn.keys();
}
private CommonAttributes attributes(SaveStatus saveStatus)
{
CommonAttributes.Mutable mutable = new CommonAttributes.Mutable(txnId);
if (saveStatus.known.isDefinitionKnown())
mutable.partialTxn(partialTxn);
if (saveStatus.known.deps.hasProposedOrDecidedDeps())
mutable.partialDeps(partialDeps);
mutable.route(route);
mutable.durability(NotDurable);
return mutable;
}
public Command build(SaveStatus saveStatus)
{
switch (saveStatus)
{
default: throw new AssertionError("Unhandled saveStatus: " + saveStatus);
case TruncatedApplyWithDeps:
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);
case PreAccepted:
return Command.SerializerSupport.preaccepted(attributes(saveStatus), executeAt, Ballot.ZERO);
case Accepted:
case AcceptedInvalidate:
case AcceptedWithDefinition:
case AcceptedInvalidateWithDefinition:
case PreCommittedWithDefinition:
case PreCommittedWithDefinitionAndAcceptedDeps:
case PreCommittedWithAcceptedDeps:
case PreCommitted:
return Command.SerializerSupport.accepted(attributes(saveStatus), saveStatus, executeAt, promised, accepted);
case Committed:
return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, null);
case Stable:
case ReadyToExecute:
return Command.SerializerSupport.committed(attributes(saveStatus), saveStatus, executeAt, promised, accepted, waitingOn);
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());
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);
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());
case Erased:
case ErasedOrInvalidOrVestigial:
case Invalidated:
return Command.SerializerSupport.invalidated(txnId, Listeners.Immutable.EMPTY);
}
}
}
public static Gen<PartitionKey> keys()
{
return keys(fromQT(CassandraGenerators.TABLE_ID_GEN),
@ -186,7 +350,7 @@ public class AccordGenerators
while (offset.compareTo(size) < 0)
{
BigInteger end = offset.add(update);
TokenRange r = (TokenRange) splitter.subRange(range, offset, end);
TokenRange r = splitter.subRange(range, offset, end);
for (TableId id : tables)
{
ranges.add(r.withTable(id));