From 325e48ac39c47641da8879829c1fd9712c3a42f0 Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Wed, 28 Aug 2024 17:18:36 +0200 Subject: [PATCH] Switch to streaming serialization of SavedCommand Patch by Alex Petrov; reviewed by David Capwell for CASSANDRA-19865 Co-authored-by: dcapwell --- modules/accord | 2 +- .../org/apache/cassandra/journal/Journal.java | 44 +- .../service/accord/AccordCommandStore.java | 6 +- .../service/accord/AccordJournal.java | 360 ++----- .../service/accord/AccordKeyspace.java | 3 + .../service/accord/AccordObjectSizes.java | 2 +- .../service/accord/AccordSafeCommand.java | 2 +- .../cassandra/service/accord/IJournal.java | 8 +- .../cassandra/service/accord/JournalKey.java | 37 +- .../service/accord/SavedCommand.java | 895 ++++++++---------- .../service/accord/async/AsyncOperation.java | 27 +- .../serializers/WaitingOnSerializer.java | 4 +- .../CompactionAccordIteratorsTest.java | 4 +- .../accord/AccordJournalOrderTest.java | 22 +- .../service/accord/AccordJournalTest.java | 3 +- .../service/accord/AccordTestUtils.java | 14 +- .../cassandra/service/accord/MockJournal.java | 326 ++++++- .../service/accord/SavedCommandTest.java | 144 +++ .../accord/async/AsyncOperationTest.java | 15 + .../CommandsForKeySerializerTest.java | 4 +- .../serializers/WaitingOnSerializerTest.java | 2 +- .../cassandra/utils/AccordGenerators.java | 172 +++- 22 files changed, 1191 insertions(+), 905 deletions(-) create mode 100644 test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java diff --git a/modules/accord b/modules/accord index 178952b41f..449b2b4d0b 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 178952b41f05bfa307aef03dcc013e37fb6230b4 +Subproject commit 449b2b4d0bf4bb44d55a3c57f712a4d5a15e7220 diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index db86106d42..c092363af7 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -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 implements Shutdownable return null; } - // TODO: This should be improved with new index that should take better care of handling multiple items public List readAll(K id) { - EntrySerializer.EntryHolder holder = new EntrySerializer.EntryHolder<>(); List 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 holder = new EntrySerializer.EntryHolder<>(); try (ReferencedSegments segments = selectAndReference(id)) { for (Segment segment : segments.all()) @@ -357,7 +364,7 @@ public class Journal 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 implements Shutdownable }); } } - return res; } /** @@ -504,11 +510,28 @@ public class Journal implements Shutdownable * @param hosts hosts expected to invalidate the record */ public RecordPointer asyncWrite(K id, V record, Set 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 hosts) { RecordPointer recordPointer; try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get()) { - valueSerializer.serialize(id, record, dob, params.userVersion()); + writer.write(dob, params.userVersion()); ActiveSegment.Allocation alloc = allocate(dob.getLength(), hosts); recordPointer = alloc.write(id, dob.unsafeGetBufferAndFlip(), hosts); flusher.asyncFlush(alloc); @@ -521,6 +544,7 @@ public class Journal implements Shutdownable return recordPointer; } + private ActiveSegment.Allocation allocate(int entrySize, Set hosts) { ActiveSegment segment = currentSegment; @@ -942,4 +966,14 @@ public class Journal 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; + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 929a1695d7..39370a1cfc 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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 bootstrapBeganAt() { return super.bootstrapBeganAt(); } public NavigableMap safeToRead() { return super.safeToRead(); } - public void appendCommands(List commands, List sanityCheck, Runnable onFlush) + public void appendCommands(List> commands, List sanityCheck, Runnable onFlush) { journal.appendCommand(id, commands, sanityCheck, onFlush); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 09bee34641..eda7155a38 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -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 SENTINEL_HOSTS = Collections.singleton(0); - static final ThreadLocal keyCRCBytes = ThreadLocal.withInitial(() -> new byte[23]); + static final ThreadLocal keyCRCBytes = ThreadLocal.withInitial(() -> new byte[22]); public final Journal 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() + { + 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 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 loadDiffs(int commandStoreId, Timestamp txnId) + public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId) { - return (List)(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 outcomes, List sanityCheck, Runnable onFlush) + public void appendCommand(int commandStoreId, List> outcomes, List sanityCheck, Runnable onFlush) { RecordPointer pointer = null; - for (int i = 0; i < outcomes.size(); i++) + for (SavedCommand.Writer 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 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 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 - { - final IVersionedSerializer wrapped; - - private MessageSerializer(IVersionedSerializer wrapped) - { - this.wrapped = wrapped; - } - - static MessageSerializer wrap(IVersionedSerializer 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. - *

- * 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 - { - /* 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. - *

- * 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 serializer; - - Type(int id, ValueSerializer 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) serializer), txnIdProvider); - } - - Type(int id, MessageType type, IVersionedSerializer serializer, TxnIdProvider txnIdProvider) - { - //noinspection unchecked - this(id, type, type, MessageSerializer.wrap((IVersionedSerializer) serializer), txnIdProvider); - } - - Type(int id, MessageType incomingType, MessageType outgoingType, ValueSerializer 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) 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 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 */ diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index eab58afefb..08fb1ce4c9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index 46e0d8c158..4262749991 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -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)); diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java index 0cafdfec08..28220cbf47 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java @@ -127,7 +127,7 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState diff() { return SavedCommand.diff(original, current); } diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java index eb0f627c1c..8e44ad2aac 100644 --- a/src/java/org/apache/cassandra/service/accord/IJournal.java +++ b/src/java/org/apache/cassandra/service/accord/IJournal.java @@ -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 command, List sanityCheck, Runnable onFlush); + void appendCommand(int commandStoreId, + List> command, + List sanityCheck, + Runnable onFlush); } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java index 04688e1f2a..aa0c747396 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -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 + '}'; } diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java index 282773c6d9..bb1fe6da73 100644 --- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java +++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java @@ -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 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 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 listeners) + public interface Writer 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 { - 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 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 diffs) + + public static Writer 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 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 int collectFlags(OBJ lo, OBJ ro, Function 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 VAL ifNotEqual(OBJ lo, OBJ ro, Function 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 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 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 - { - @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); diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java index 34d80250a2..0b984446dc 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java @@ -251,34 +251,31 @@ public abstract class AsyncOperation extends AsyncChains.Head 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 diffs = null; + List> diffs = null; for (AccordSafeCommand commandState : context.commands.values()) { - SavedCommand.SavedDiff diff = commandState.diff(); - if (diff != null) + SavedCommand.Writer 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: diff --git a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java index 9bd2e0082a..fab09f2354 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java @@ -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); } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index 3e1f75f62a..a955a9c315 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -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); } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java index 0f2b9d395b..2c6dc9221c 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java @@ -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 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()); } }; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalTest.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalTest.java index 375fffa8fe..75a07196e2 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordJournalTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordJournalTest.java @@ -122,7 +122,6 @@ public class AccordJournalTest private Gen keyGen() { Gen txnIdGen = AccordGens.txnIds(); - Gen typeGen = Gens.enums().all(AccordJournal.Type.class); - return rs -> new JournalKey(txnIdGen.next(rs), typeGen.next(rs)); + return rs -> new JournalKey(txnIdGen.next(rs)); } } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 999ae87576..bfbf4e5095 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -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 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); } } diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java index d09ff6eeff..8f3192afa5 100644 --- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java +++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java @@ -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> commands = new HashMap<>(); + private final Map> commands = new HashMap<>(); @Override public Command loadCommand(int commandStoreId, TxnId txnId) { - Type type = Type.SAVED_COMMAND; - JournalKey key = new JournalKey(txnId, type, commandStoreId); - List saved = commands.get(key); + JournalKey key = new JournalKey(txnId, commandStoreId); + List 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 diffs, List sanityCheck, Runnable onFlush) + public void appendCommand(int commandStoreId, List> diffs, List sanityCheck, Runnable onFlush) { - Type type = Type.SAVED_COMMAND; - for (SavedCommand.SavedDiff diff : diffs) + for (SavedCommand.Writer 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 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 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 NewValue ifNotEqual(OBJ lo, OBJ ro, Function 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 + { + final T value; + + private NewValue(T value) + { + this.value = value; + } + + public boolean isNull() + { + return value == null; + } + + public T get() + { + return value; + } + + public static NewValue 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 executeAt; + public final NewValue saveStatus; + public final NewValue durability; + + public final NewValue acceptedOrCommitted; + public final NewValue promised; + + public final NewValue> route; + public final NewValue partialTxn; + public final NewValue partialDeps; + public final NewValue> additionalKeysOrRanges; + + public final NewValue writes; + public final NewValue> listeners; + public final NewValue waitingOn; + + public LoadedDiff(TxnId txnId, + NewValue executeAt, + NewValue saveStatus, + NewValue durability, + + NewValue acceptedOrCommitted, + NewValue promised, + + NewValue> route, + NewValue partialTxn, + NewValue partialDeps, + NewValue> additionalKeysOrRanges, + + NewValue waitingOn, + NewValue writes, + NewValue> 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 + + '}'; + } + } } diff --git a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java new file mode 100644 index 0000000000..5135bde35f --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java @@ -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 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 has = EnumSet.of(Fields.TXN_ID, Fields.SAVE_STATUS, Fields.DURABILITY, Fields.PROMISED, + Fields.ACCEPTED /* this is Zero... which kinda means null... */); + Set missing = Sets.difference(ALL, has); + assertHas(flags, has); + assertMissing(flags, missing); + } + + @Test + public void serde() + { + Gen 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 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 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(); + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java index 11ed33b42f..f80e036b50 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java @@ -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 { diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 81462c0e65..c2a2b32af1 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -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: diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java index f83a109977..ce5aabba9f 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java @@ -73,7 +73,7 @@ public class WaitingOnSerializerTest Gen 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); diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index f587fd85af..19c21f3215 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -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 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 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 commandsBuilder(Gen txnIdGen, Gen fastPath, Gen recover, TriFunction 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 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));