diff --git a/modules/accord b/modules/accord index 8226b2d775..b99c4671fa 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 8226b2d7759319d7a0b0c823ab09b4344c5423f7 +Subproject commit b99c4671fa0b22bed7f5a37fc5acaa2d2579e5b2 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 1633dc7930..b3db3f3822 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -39,6 +39,7 @@ import accord.local.CommandStores.RangesForEpochHolder; import accord.local.NodeTimeService; import accord.local.PreLoadContext; import accord.local.SafeCommandStore; +import accord.primitives.Deps; import accord.primitives.RoutableKey; import accord.primitives.TxnId; import accord.utils.Invariants; @@ -101,6 +102,11 @@ public class AccordCommandStore extends CommandStore return Thread.currentThread().getId() == threadId; } + @Override + protected void registerHistoricalTransactions(Deps deps) + { + } + public void setCacheSize(long bytes) { checkInStoreThread(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index 235cc1be57..3bad135715 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -17,10 +17,14 @@ */ package org.apache.cassandra.service.accord; +import java.util.function.Supplier; + import accord.api.Agent; +import accord.api.ConfigurationService.EpochReady; import accord.api.DataStore; import accord.api.ProgressLog; import accord.local.CommandStores; +import accord.local.Node; import accord.local.NodeTimeService; import accord.local.PreLoadContext; import accord.local.SafeCommandStore; @@ -32,7 +36,7 @@ import accord.utils.RandomSource; import org.apache.cassandra.concurrent.ImmediateExecutor; import org.apache.cassandra.journal.AsyncWriteCallback; -public class AccordCommandStores extends CommandStores +public class AccordCommandStores extends CommandStores { private final AccordJournal journal; @@ -50,14 +54,6 @@ public class AccordCommandStores extends CommandStores new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, journal); } - @Override - public synchronized void shutdown() - { - super.shutdown(); - journal.shutdown(); - //TODO shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface) - } - @Override protected void mapReduceConsume( PreLoadContext context, @@ -100,13 +96,6 @@ public class AccordCommandStores extends CommandStores }); } - @Override - public synchronized void updateTopology(Topology newTopology) - { - super.updateTopology(newTopology); - refreshCacheSizes(); - } - private long cacheSize; synchronized void setCacheSize(long bytes) @@ -128,4 +117,28 @@ public class AccordCommandStores extends CommandStores { return 5 << 20; // TODO (required): make configurable } + + @Override + public synchronized Supplier updateTopology(Node node, Topology newTopology) + { + Supplier start = super.updateTopology(node, newTopology); + return () -> { + EpochReady ready = start.get(); + ready.metadata.addCallback(() -> { + synchronized (this) + { + refreshCacheSizes(); + } + }); + return ready; + }; + } + + @Override + public synchronized void shutdown() + { + super.shutdown(); + journal.shutdown(); + //TODO shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface) + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java index d62985b0a1..e8f1033664 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java @@ -73,15 +73,18 @@ public class AccordConfigurationService implements ConfigurationService } @Override - public void acknowledgeEpoch(long epoch) + public void acknowledgeEpoch(EpochReady ready) { - Topology acknowledged = getTopologyForEpoch(epoch); + Topology acknowledged = getTopologyForEpoch(ready.epoch); for (Node.Id node : acknowledged.nodes()) { if (node.equals(localId)) continue; - for (Listener listener : listeners) - listener.onEpochSyncComplete(node, epoch); + + ready.coordination.addCallback(() -> { + for (Listener listener : listeners) + listener.onEpochSyncComplete(node, ready.epoch); + }); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index fc719dfdf8..9d38bd7133 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -45,7 +45,6 @@ import accord.api.Result; import accord.impl.CommandsForKey; import accord.impl.CommandsForKey.CommandTimeseries; import accord.local.Command; -import accord.local.CommandListener; import accord.local.CommandStore; import accord.local.CommonAttributes; import accord.local.Listeners; @@ -192,7 +191,7 @@ public class AccordKeyspace static final LocalVersionedSerializer partialDeps = localSerializer(DepsSerializer.partialDeps); static final LocalVersionedSerializer writes = localSerializer(CommandSerializers.writes); static final LocalVersionedSerializer result = localSerializer(TxnData.serializer); - static final LocalVersionedSerializer listeners = localSerializer(ListenerSerializers.listener); + static final LocalVersionedSerializer listeners = localSerializer(ListenerSerializers.listener); private static LocalVersionedSerializer localSerializer(IVersionedSerializer serializer) { @@ -508,7 +507,7 @@ public class AccordKeyspace addCellIfModified(CommandsColumns.dependencies, Command::partialDeps, CommandsSerializers.partialDeps, builder, timestampMicros, original, command); - addSetChanges(CommandsColumns.listeners, cmd -> Sets.filter(cmd.listeners(), l -> !l.isTransient()), v -> serialize(v, CommandsSerializers.listeners), builder, timestampMicros, nowInSeconds, original, command); + addSetChanges(CommandsColumns.listeners, Command::durableListeners, v -> serialize(v, CommandsSerializers.listeners), builder, timestampMicros, nowInSeconds, original, command); if (command.isCommitted()) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordLoadingState.java b/src/java/org/apache/cassandra/service/accord/AccordLoadingState.java index 1316b02d48..8dea7389c0 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordLoadingState.java +++ b/src/java/org/apache/cassandra/service/accord/AccordLoadingState.java @@ -18,9 +18,14 @@ package org.apache.cassandra.service.accord; +import java.util.Collection; +import java.util.Collections; +import java.util.Set; import java.util.concurrent.Callable; import java.util.function.Function; +import accord.local.Command; +import accord.utils.DeterministicIdentitySet; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncResults; @@ -54,6 +59,7 @@ public class AccordLoadingState private final K key; private Object state = UNINITIALIZED; + private Set transientListeners; public AccordLoadingState(K key) { @@ -160,4 +166,27 @@ public class AccordLoadingState checkState(LoadingState.PENDING, false); return (PendingLoad) state; } + + + public void addListener(Command.TransientListener listener) + { + if (transientListeners == null) + transientListeners = new DeterministicIdentitySet<>(); + transientListeners.add(listener); + } + + public boolean removeListener(Command.TransientListener listener) + { + if (transientListeners == null) + return false; + + return transientListeners.remove(listener); + } + + public Collection transientListeners() + { + if (transientListeners == null) + return Collections.emptySet(); + return transientListeners; + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index d56e88ba57..c79d1ba5b9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -30,7 +30,6 @@ import accord.api.Result; import accord.api.RoutingKey; import accord.impl.CommandsForKey; import accord.local.Command; -import accord.local.CommandListener; import accord.local.CommonAttributes; import accord.local.Node; import accord.local.SaveStatus; @@ -233,7 +232,7 @@ public class AccordObjectSizes return size; } - private static final long EMPTY_WRITES_SIZE = measure(new Writes(null, null, null)); + private static final long EMPTY_WRITES_SIZE = measure(new Writes(null, null, null, null)); public static long writes(Writes writes) { long size = EMPTY_WRITES_SIZE; @@ -244,14 +243,12 @@ public class AccordObjectSizes return size; } - private static final long EMPTY_COMMAND_LISTENER = measure(new Command.Listener(null)); + private static final long EMPTY_COMMAND_LISTENER = measure(new Command.ProxyListener(null)); private static final long EMPTY_CFK_LISTENER = measure(new CommandsForKey.Listener((Key) null)); - public static long listener(CommandListener listener) + public static long listener(Command.DurableAndIdempotentListener listener) { - if (listener.isTransient()) - return 0; - if (listener instanceof Command.Listener) - return EMPTY_COMMAND_LISTENER + timestamp(((Command.Listener) listener).txnId()); + if (listener instanceof Command.ProxyListener) + return EMPTY_COMMAND_LISTENER + timestamp(((Command.ProxyListener) listener).txnId()); if (listener instanceof CommandsForKey.Listener) return EMPTY_CFK_LISTENER + key(((CommandsForKey.Listener) listener).key()); throw new IllegalArgumentException("Unhandled listener type: " + listener.getClass()); @@ -306,7 +303,7 @@ public class AccordObjectSizes size += sizeNullable(command.progressKey(), AccordObjectSizes::key); size += sizeNullable(command.route(), AccordObjectSizes::route); size += sizeNullable(command.promised(), AccordObjectSizes::timestamp); - for (CommandListener listener : command.listeners()) + for (Command.DurableAndIdempotentListener listener : command.durableListeners()) size += listener(listener); if (!command.isWitnessed()) diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java index 3dc37b2f3c..b44f684e83 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java @@ -18,6 +18,7 @@ package org.apache.cassandra.service.accord; +import java.util.Collection; import java.util.Objects; import com.google.common.annotations.VisibleForTesting; @@ -121,4 +122,22 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState transientListeners() + { + return global.transientListeners(); + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index c71ba18f69..e1bf80895b 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -144,7 +144,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore keysOrRanges, Ranges slice) + public Timestamp maxConflict(Seekables keysOrRanges, Ranges slice) { // TODO: Seekables // TODO: efficiency diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 232684565a..5154ca7d80 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -22,8 +22,13 @@ import accord.api.Agent; import accord.api.Result; import accord.local.Command; import accord.local.Node; +import accord.primitives.Ranges; +import accord.primitives.Seekables; import accord.primitives.Timestamp; +import accord.primitives.Txn; import accord.primitives.TxnId; +import org.apache.cassandra.service.accord.txn.TxnQuery; +import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.utils.JVMStabilityInspector; import static java.util.concurrent.TimeUnit.MICROSECONDS; @@ -46,6 +51,12 @@ public class AccordAgent implements Agent throw error; } + @Override + public void onFailedBootstrap(String phase, Ranges ranges, Runnable retry, Throwable failure) + { + + } + @Override public void onUncaughtException(Throwable t) { @@ -65,4 +76,10 @@ public class AccordAgent implements Agent // TODO: should distinguish between reads and writes return now - initiated.hlc() > getReadRpcTimeout(MICROSECONDS); } + + @Override + public Txn emptyTxn(Txn.Kind kind, Seekables keysOrRanges) + { + return new Txn.InMemory(kind, keysOrRanges, TxnRead.EMPTY, TxnQuery.ALL, null); + } } diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java index 34ba0936c6..0a486ee402 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java @@ -19,18 +19,22 @@ package org.apache.cassandra.service.accord.async; import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.function.BiConsumer; import java.util.function.Function; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.Iterables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.impl.CommandsForKey; import accord.local.Command; +import accord.local.PreLoadContext; import accord.primitives.RoutableKey; import accord.primitives.TxnId; import accord.utils.Invariants; @@ -72,6 +76,15 @@ public class AsyncLoader this.keys = keys; } + protected static Iterable txnIds(PreLoadContext context) + { + TxnId primaryid = context.primaryTxnId(); + Collection additionalIds = context.additionalTxnIds(); + if (primaryid == null) return additionalIds; + if (additionalIds.isEmpty()) return Collections.singleton(primaryid); + return Iterables.concat(Collections.singleton(primaryid), additionalIds); + } + private > void referenceAndAssembleReads(Iterable keys, Map context, AccordStateCache.Instance cache, 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 ce5bb125f1..bf86f6777b 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java @@ -42,6 +42,8 @@ import org.apache.cassandra.service.accord.AccordSafeCommandsForKey; import org.apache.cassandra.service.accord.AccordSafeCommandStore; import org.apache.cassandra.service.accord.AccordSafeState; +import static org.apache.cassandra.service.accord.async.AsyncLoader.txnIds; + public abstract class AsyncOperation extends AsyncChains.Head implements Runnable, Function { private static final Logger logger = LoggerFactory.getLogger(AsyncOperation.class); @@ -143,7 +145,7 @@ public abstract class AsyncOperation extends AsyncChains.Head implements R AsyncLoader createAsyncLoader(AccordCommandStore commandStore, PreLoadContext preLoadContext) { - return new AsyncLoader(commandStore, preLoadContext.txnIds(), toRoutableKeys(preLoadContext.keys())); + return new AsyncLoader(commandStore, txnIds(preLoadContext), toRoutableKeys(preLoadContext.keys())); } @VisibleForTesting diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java index 0441c640c8..366b469c2f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -169,6 +169,7 @@ public class CommandSerializers @Override public void serialize(Writes writes, DataOutputPlus out, int version) throws IOException { + txnId.serialize(writes.txnId, out, version); timestamp.serialize(writes.executeAt, out, version); KeySerializers.seekables.serialize(writes.keys, out, version); boolean hasWrites = writes.write != null; @@ -180,7 +181,7 @@ public class CommandSerializers @Override public Writes deserialize(DataInputPlus in, int version) throws IOException { - return new Writes(timestamp.deserialize(in, version), + return new Writes(txnId.deserialize(in, version), timestamp.deserialize(in, version), KeySerializers.seekables.deserialize(in, version), in.readBoolean() ? TxnWrite.serializer.deserialize(in, version) : null); } @@ -188,7 +189,8 @@ public class CommandSerializers @Override public long serializedSize(Writes writes, int version) { - long size = timestamp.serializedSize(writes.executeAt, version); + long size = txnId.serializedSize(writes.txnId, version); + size += timestamp.serializedSize(writes.executeAt, version); size += KeySerializers.seekables.serializedSize(writes.keys, version); boolean hasWrites = writes.write != null; size += TypeSizes.sizeof(hasWrites); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java index 1de5b96a5b..37f9302ae3 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java @@ -25,7 +25,6 @@ import accord.messages.GetDeps.GetDepsOk; import accord.primitives.PartialRoute; import accord.primitives.Seekables; import accord.primitives.Timestamp; -import accord.primitives.Txn; import accord.primitives.TxnId; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -40,7 +39,6 @@ public class GetDepsSerializers { KeySerializers.seekables.serialize(msg.keys, out, version); CommandSerializers.timestamp.serialize(msg.executeAt, out, version); - CommandSerializers.kind.serialize(msg.kind, out, version); } @Override @@ -48,16 +46,14 @@ public class GetDepsSerializers { Seekables keys = KeySerializers.seekables.deserialize(in, version); Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version); - Txn.Kind kind = CommandSerializers.kind.deserialize(in, version); - return GetDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, keys, executeAt, kind); + return GetDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, keys, executeAt); } @Override public long serializedBodySize(GetDeps msg, int version) { return KeySerializers.seekables.serializedSize(msg.keys, version) - + CommandSerializers.timestamp.serializedSize(msg.executeAt, version) - + CommandSerializers.kind.serializedSize(msg.kind, version); + + CommandSerializers.timestamp.serializedSize(msg.executeAt, version); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java index 37afe0a59a..4581ab8cad 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java @@ -22,8 +22,6 @@ import java.io.IOException; import accord.impl.CommandsForKey; import accord.local.Command; -import accord.local.CommandListener; -import accord.utils.Invariants; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -36,9 +34,9 @@ public class ListenerSerializers { COMMAND, COMMANDS_FOR_KEY; - private static Kind of(CommandListener listener) + private static Kind of(Command.DurableAndIdempotentListener listener) { - if (listener instanceof Command.Listener) + if (listener instanceof Command.ProxyListener) return COMMAND; if (listener instanceof CommandsForKey.Listener) @@ -49,22 +47,22 @@ public class ListenerSerializers } - private static final IVersionedSerializer commandListener = new IVersionedSerializer() + private static final IVersionedSerializer commandListener = new IVersionedSerializer() { @Override - public void serialize(Command.Listener listener, DataOutputPlus out, int version) throws IOException + public void serialize(Command.ProxyListener listener, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(listener.txnId(), out, version); } @Override - public Command.Listener deserialize(DataInputPlus in, int version) throws IOException + public Command.ProxyListener deserialize(DataInputPlus in, int version) throws IOException { - return new Command.Listener(CommandSerializers.txnId.deserialize(in, version)); + return new Command.ProxyListener(CommandSerializers.txnId.deserialize(in, version)); } @Override - public long serializedSize(Command.Listener listener, int version) + public long serializedSize(Command.ProxyListener listener, int version) { return CommandSerializers.txnId.serializedSize(listener.txnId(), version); } @@ -91,18 +89,17 @@ public class ListenerSerializers } }; - public static final IVersionedSerializer listener = new IVersionedSerializer() + public static final IVersionedSerializer listener = new IVersionedSerializer() { @Override - public void serialize(CommandListener listener, DataOutputPlus out, int version) throws IOException + public void serialize(Command.DurableAndIdempotentListener listener, DataOutputPlus out, int version) throws IOException { - Invariants.checkArgument(!listener.isTransient()); Kind kind = Kind.of(listener); out.write(kind.ordinal()); switch (kind) { case COMMAND: - commandListener.serialize((Command.Listener) listener, out, version); + commandListener.serialize((Command.ProxyListener) listener, out, version); break; case COMMANDS_FOR_KEY: cfkListener.serialize((CommandsForKey.Listener) listener, out, version); @@ -113,7 +110,7 @@ public class ListenerSerializers } @Override - public CommandListener deserialize(DataInputPlus in, int version) throws IOException + public Command.DurableAndIdempotentListener deserialize(DataInputPlus in, int version) throws IOException { Kind kind = Kind.values()[in.readByte()]; switch (kind) @@ -128,15 +125,14 @@ public class ListenerSerializers } @Override - public long serializedSize(CommandListener listener, int version) + public long serializedSize(Command.DurableAndIdempotentListener listener, int version) { - Invariants.checkArgument(!listener.isTransient()); Kind kind = Kind.of(listener); long size = TypeSizes.BYTE_SIZE; switch (kind) { case COMMAND: - size += commandListener.serializedSize((Command.Listener) listener, version); + size += commandListener.serializedSize((Command.ProxyListener) listener, version); break; case COMMANDS_FOR_KEY: size += cfkListener.serializedSize((CommandsForKey.Listener) listener, version); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java index a344028b49..18b8b15e51 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java @@ -20,10 +20,11 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; -import accord.messages.ReadData; import accord.messages.ReadData.ReadNack; import accord.messages.ReadData.ReadOk; import accord.messages.ReadData.ReadReply; +import accord.messages.ReadTxnData; +import accord.primitives.Ranges; import accord.primitives.Seekables; import accord.primitives.TxnId; import org.apache.cassandra.db.TypeSizes; @@ -32,12 +33,16 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.service.accord.txn.TxnData; +import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; +import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; +import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize; + public class ReadDataSerializers { - public static final IVersionedSerializer request = new IVersionedSerializer() + public static final IVersionedSerializer request = new IVersionedSerializer() { @Override - public void serialize(ReadData read, DataOutputPlus out, int version) throws IOException + public void serialize(ReadTxnData read, DataOutputPlus out, int version) throws IOException { CommandSerializers.txnId.serialize(read.txnId, out, version); KeySerializers.seekables.serialize(read.readScope, out, version); @@ -46,17 +51,17 @@ public class ReadDataSerializers } @Override - public ReadData deserialize(DataInputPlus in, int version) throws IOException + public ReadTxnData deserialize(DataInputPlus in, int version) throws IOException { TxnId txnId = CommandSerializers.txnId.deserialize(in, version); Seekables readScope = KeySerializers.seekables.deserialize(in, version); long waitForEpoch = in.readUnsignedVInt(); long executeAtEpoch = in.readUnsignedVInt() + waitForEpoch; - return ReadData.SerializerSupport.create(txnId, readScope, executeAtEpoch, waitForEpoch); + return ReadTxnData.SerializerSupport.create(txnId, readScope, executeAtEpoch, waitForEpoch); } @Override - public long serializedSize(ReadData read, int version) + public long serializedSize(ReadTxnData read, int version) { return CommandSerializers.txnId.serializedSize(read.txnId, version) + KeySerializers.seekables.serializedSize(read.readScope, version) @@ -81,6 +86,7 @@ public class ReadDataSerializers out.writeByte(0); ReadOk readOk = (ReadOk) reply; + serializeNullable(readOk.unavailable, out, version, KeySerializers.ranges); TxnData.nullableSerializer.serialize((TxnData) readOk.data, out, version); } @@ -91,7 +97,9 @@ public class ReadDataSerializers if (id != 0) return nacks[id - 1]; - return new ReadOk(TxnData.nullableSerializer.deserialize(in, version)); + Ranges ranges = deserializeNullable(in, version, KeySerializers.ranges); + TxnData data = TxnData.nullableSerializer.deserialize(in, version); + return new ReadOk(ranges, data); } @Override @@ -101,7 +109,9 @@ public class ReadDataSerializers return TypeSizes.BYTE_SIZE; ReadOk readOk = (ReadOk) reply; - return TypeSizes.BYTE_SIZE + TxnData.nullableSerializer.serializedSize((TxnData) readOk.data, version); + return TypeSizes.BYTE_SIZE + + serializedNullableSize(readOk.unavailable, version, KeySerializers.ranges) + + TxnData.nullableSerializer.serializedSize((TxnData) readOk.data, version); } }; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index fbd00f637e..b4111320d0 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -19,6 +19,7 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; +import javax.annotation.Nonnull; import javax.annotation.Nullable; import accord.api.Result; @@ -90,6 +91,7 @@ public class RecoverySerializers CommandSerializers.ballot.serialize(recoverOk.accepted, out, version); serializeNullable(recoverOk.executeAt, out, version, CommandSerializers.timestamp); DepsSerializer.partialDeps.serialize(recoverOk.deps, out, version); + serializeNullable(recoverOk.acceptedDeps, out, version, DepsSerializer.partialDeps); DepsSerializer.deps.serialize(recoverOk.earlierCommittedWitness, out, version); DepsSerializer.deps.serialize(recoverOk.earlierAcceptedNoWitness, out, version); out.writeBoolean(recoverOk.rejectsFastPath); @@ -112,9 +114,9 @@ public class RecoverySerializers return new RecoverNack(supersededBy); } - RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, PartialDeps deps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) + RecoverOk deserializeOk(TxnId txnId, Status status, Ballot accepted, Timestamp executeAt, @Nonnull PartialDeps deps, PartialDeps acceptedDeps, Deps earlierCommittedWitness, Deps earlierAcceptedNoWitness, boolean rejectsFastPath, Writes writes, Result result, DataInputPlus in, int version) { - return new RecoverOk(txnId, status, accepted, executeAt, deps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result); + return new RecoverOk(txnId, status, accepted, executeAt, deps, acceptedDeps, earlierCommittedWitness, earlierAcceptedNoWitness, rejectsFastPath, writes, result); } @Override @@ -129,6 +131,7 @@ public class RecoverySerializers CommandSerializers.ballot.deserialize(in, version), deserializeNullable(in, version, CommandSerializers.timestamp), DepsSerializer.partialDeps.deserialize(in, version), + deserializeNullable(in, version, DepsSerializer.partialDeps), DepsSerializer.deps.deserialize(in, version), DepsSerializer.deps.deserialize(in, version), in.readBoolean(), @@ -150,6 +153,7 @@ public class RecoverySerializers size += CommandSerializers.ballot.serializedSize(recoverOk.accepted, version); size += serializedNullableSize(recoverOk.executeAt, version, CommandSerializers.timestamp); size += DepsSerializer.partialDeps.serializedSize(recoverOk.deps, version); + size += serializedNullableSize(recoverOk.acceptedDeps, version, DepsSerializer.partialDeps); size += DepsSerializer.deps.serializedSize(recoverOk.earlierCommittedWitness, version); size += DepsSerializer.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version); size += TypeSizes.sizeof(recoverOk.rejectsFastPath); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java b/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java index 29b28606ba..115864d830 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java @@ -28,6 +28,7 @@ import accord.api.Query; import accord.api.Read; import accord.api.Result; import accord.api.Update; +import accord.primitives.Timestamp; import accord.primitives.TxnId; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; @@ -48,7 +49,7 @@ public abstract class TxnQuery implements Query } @Override - public Result compute(TxnId txnId, Data data, @Nullable Read read, @Nullable Update update) + public Result compute(TxnId txnId, Timestamp executeAt, Data data, @Nullable Read read, @Nullable Update update) { return data != null ? (TxnData) data : new TxnData(); } @@ -63,7 +64,7 @@ public abstract class TxnQuery implements Query } @Override - public Result compute(TxnId txnId, Data data, @Nullable Read read, @Nullable Update update) + public Result compute(TxnId txnId, Timestamp executeAt, Data data, @Nullable Read read, @Nullable Update update) { return new TxnData(); } @@ -78,7 +79,7 @@ public abstract class TxnQuery implements Query } @Override - public Result compute(TxnId txnId, Data data, @Nullable Read read, Update update) + public Result compute(TxnId txnId, Timestamp executeAt, Data data, @Nullable Read read, Update update) { checkNotNull(txnId, "txnId should not be null"); checkNotNull(data, "data should not be null"); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java index 5b5812aed8..072a21c1df 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java @@ -54,6 +54,7 @@ public class TxnRead extends AbstractKeySorted implements Read public static final String SERIAL_READ_NAME = "SERIAL_READ"; public static final TxnDataName SERIAL_READ = TxnDataName.user(SERIAL_READ_NAME); private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnRead(new TxnNamedRead[0], null)); + public static final TxnRead EMPTY = new TxnRead(new TxnNamedRead[0], Keys.EMPTY); private final Keys txnKeys; diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index f7b6565dbe..a35ae36020 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -33,6 +33,7 @@ import accord.api.Update; import accord.api.Write; import accord.primitives.Keys; import accord.primitives.Ranges; +import accord.primitives.Timestamp; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; @@ -170,7 +171,7 @@ public class TxnUpdate implements Update } @Override - public Write apply(Data data) + public Write apply(Timestamp executeAt, Data data) { if (!checkCondition(data)) return TxnWrite.EMPTY; diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java index 2f9da8910c..317d3711d4 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java @@ -44,6 +44,7 @@ import accord.primitives.FullRoute; import accord.primitives.Keys; import accord.primitives.Ranges; import accord.primitives.Seekables; +import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; import accord.topology.Topologies; @@ -244,7 +245,7 @@ public class AccordJournalSimulationTest extends SimulationTestBase } @Override - public Write apply(@Nullable Data data) + public Write apply(Timestamp executeAt, @Nullable Data data) { return null; } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java index dce4312d68..6031f01e7c 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java @@ -114,7 +114,7 @@ public class AccordCommandStoreTest attrs.partialDeps(dependencies); ImmutableSortedSet waitingOnCommit = ImmutableSortedSet.of(oldTxnId1); ImmutableSortedMap waitingOnApply = ImmutableSortedMap.of(oldTimestamp, oldTxnId2); - attrs.addListener(new Command.Listener(oldTxnId1)); + attrs.addListener(new Command.ProxyListener(oldTxnId1)); Pair result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt); Command command = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted, waitingOnCommit, waitingOnApply, result.left, result.right); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 0317777640..f09c2ab4d0 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -44,6 +44,7 @@ import accord.api.Write; import accord.impl.CommandsForKey; import accord.impl.InMemoryCommandStore; import accord.local.Command; +import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.CommonAttributes; import accord.local.Node; @@ -197,7 +198,7 @@ public class AccordTestUtils public static Pair processTxnResult(AccordCommandStore commandStore, TxnId txnId, PartialTxn txn, Timestamp executeAt) throws Throwable { AtomicReference> result = new AtomicReference<>(); - getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(Collections.emptyList(), txn.keys()), + getUninterruptibly(commandStore.execute(PreLoadContext.contextFor(txn.keys()), safeStore -> { TxnRead read = (TxnRead) txn.read(); Data readData = read.keys().stream().map(key -> { @@ -215,9 +216,9 @@ public class AccordTestUtils } }) .reduce(null, TxnData::merge); - Write write = txn.update().apply(readData); - result.set(Pair.create(new Writes(executeAt, (Keys)txn.keys(), write), - txn.query().compute(txnId, readData, txn.read(), txn.update()))); + Write write = txn.update().apply(executeAt, readData); + result.set(Pair.create(new Writes(txnId, executeAt, txn.keys(), write), + txn.query().compute(txnId, executeAt, readData, txn.read(), txn.update()))); })); return result.get(); } @@ -292,7 +293,11 @@ public class AccordTestUtils public SingleEpochRanges(Ranges ranges) { this.ranges = ranges; - this.current = new CommandStores.RangesForEpoch(1, ranges); + } + + private void set(CommandStore store) + { + this.current = new CommandStores.RangesForEpoch(1, ranges, store); } } @@ -309,12 +314,15 @@ public class AccordTestUtils @Override public long now() {return now.getAsLong(); } @Override public Timestamp uniqueNow(Timestamp atLeast) { return Timestamp.fromValues(1, now.getAsLong(), node); } }; - return new InMemoryCommandStore.Synchronized(0, + + SingleEpochRanges holder = new SingleEpochRanges(Ranges.of(range)); + InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0, time, new AccordAgent(), null, - cs -> null, - new SingleEpochRanges(Ranges.of(range))); + cs -> null, holder); + holder.set(result); + return result; } public static AccordCommandStore createAccordCommandStore(Node.Id node, LongSupplier now, Topology topology) @@ -326,12 +334,16 @@ public class AccordTestUtils @Override public long now() {return now.getAsLong(); } @Override public Timestamp uniqueNow(Timestamp atLeast) { return Timestamp.fromValues(1, now.getAsLong(), node); } }; - return new AccordCommandStore(0, + + SingleEpochRanges holder = new SingleEpochRanges(topology.rangesForNode(node)); + AccordCommandStore result = new AccordCommandStore(0, time, new AccordAgent(), null, cs -> NOOP_PROGRESS_LOG, - new SingleEpochRanges(topology.rangesForNode(node))); + holder); + holder.set(result); + return result; } public static AccordCommandStore createAccordCommandStore(LongSupplier now, String keyspace, String table) 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 6eb1af434e..22f562146f 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java @@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord.async; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; @@ -92,7 +91,6 @@ import org.mockito.Mockito; import static accord.local.PreLoadContext.contextFor; import static accord.utils.Property.qt; import static accord.utils.async.AsyncChains.getUninterruptibly; -import static java.util.Collections.singleton; import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse; import static org.apache.cassandra.service.accord.AccordTestUtils.createAccordCommandStore; import static org.apache.cassandra.service.accord.AccordTestUtils.createPartialTxn; @@ -100,6 +98,7 @@ import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn; import static org.apache.cassandra.service.accord.AccordTestUtils.keys; import static org.apache.cassandra.service.accord.AccordTestUtils.loaded; import static org.apache.cassandra.service.accord.AccordTestUtils.txnId; +import static org.apache.cassandra.service.accord.async.AsyncLoader.txnIds; public class AsyncOperationTest { @@ -150,7 +149,7 @@ public class AsyncOperationTest Txn txn = createTxn((int)clock.incrementAndGet()); PartitionKey key = (PartitionKey) Iterables.getOnlyElement(txn.keys()); - getUninterruptibly(commandStore.execute(contextFor(Collections.emptyList(), Keys.of(key)), instance -> { + getUninterruptibly(commandStore.execute(contextFor(key), instance -> { SafeCommandsForKey cfk = ((AccordSafeCommandStore) instance).maybeCommandsForKey(key); Assert.assertNull(cfk); })); @@ -193,7 +192,7 @@ public class AsyncOperationTest PartialDeps deps = PartialDeps.builder(ranges).build(); try { - return getUninterruptibly(commandStore.submit(PreLoadContext.contextFor(Collections.singleton(txnId), partialTxn.keys()), safe -> { + return getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys()), safe -> { CheckedCommands.preaccept(safe, txnId, partialTxn, route, null); CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps); CheckedCommands.commit(safe, txnId, route, null, partialTxn, executeAt, deps); @@ -240,7 +239,7 @@ public class AsyncOperationTest createCommittedAndPersist(commandStore, txnId); Consumer consumer = safeStore -> safeStore.command(txnId).readyToExecute(); - PreLoadContext ctx = PreLoadContext.contextFor(singleton(txnId), Keys.EMPTY); + PreLoadContext ctx = contextFor(txnId); AsyncOperation operation = new AsyncOperation.ForConsumer(commandStore, ctx, consumer) { @@ -252,7 +251,7 @@ public class AsyncOperationTest @Override AsyncLoader createAsyncLoader(AccordCommandStore commandStore, PreLoadContext preLoadContext) { - return new AsyncLoader(commandStore, preLoadContext.txnIds(), (Iterable) preLoadContext.keys()) { + return new AsyncLoader(commandStore, txnIds(preLoadContext), (Iterable) preLoadContext.keys()) { @Override void state(State state) @@ -325,7 +324,7 @@ public class AsyncOperationTest assertNoReferences(commandStore, ids, keys); - PreLoadContext ctx = PreLoadContext.contextFor(ids, keys); + PreLoadContext ctx = contextFor(ids, keys); Consumer consumer = Mockito.mock(Consumer.class); @@ -334,7 +333,7 @@ public class AsyncOperationTest @Override AsyncLoader createAsyncLoader(AccordCommandStore commandStore, PreLoadContext preLoadContext) { - return new AsyncLoader(commandStore, preLoadContext.txnIds(), (Iterable) preLoadContext.keys()) + return new AsyncLoader(commandStore, txnIds(preLoadContext), (Iterable) preLoadContext.keys()) { @Override Function loadCommandFunction() @@ -386,7 +385,7 @@ public class AsyncOperationTest createCommand(commandStore, rs, ids); assertNoReferences(commandStore, ids, keys); - PreLoadContext ctx = PreLoadContext.contextFor(ids, keys); + PreLoadContext ctx = contextFor(ids, keys); Consumer consumer = Mockito.mock(Consumer.class); String errorMsg = "txn_ids " + ids; @@ -422,7 +421,7 @@ public class AsyncOperationTest assertNoReferences(commandStore, ids, keys); - PreLoadContext ctx = PreLoadContext.contextFor(ids, keys); + PreLoadContext ctx = contextFor(ids, keys); Consumer consumer = store -> ids.forEach(id -> store.command(id).readyToExecute()); @@ -466,7 +465,7 @@ public class AsyncOperationTest SafeCommand command = store.command(id); Command current = command.current(); Assertions.assertThat(current.status()).isEqualTo(Status.ReadyToExecute); - Writes writes = current.partialTxn().execute(current.executeAt(), new TxnData()); + Writes writes = current.partialTxn().execute(current.txnId(), current.executeAt(), new TxnData()); command.preapplied(current, current.txnId(), current.asCommitted().waitingOn(), writes, null); })); getUninterruptibly(o2);