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

(Accord only) Permit nodes to join a cluster without the full transaction history 

patch by Benedict; reviewed by Blake Eggleston for CASSANDRA-18523
This commit is contained in:
Benedict Elliott Smith 2023-04-27 18:01:11 +01:00 committed by David Capwell
parent 8dc82a6369
commit 537c1f991a
24 changed files with 218 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 8226b2d7759319d7a0b0c823ab09b4344c5423f7
Subproject commit b99c4671fa0b22bed7f5a37fc5acaa2d2579e5b2

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

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

45
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -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<AccordCommandStore>
public class AccordCommandStores extends CommandStores
{
private final AccordJournal journal;
@ -50,14 +54,6 @@ public class AccordCommandStores extends CommandStores<AccordCommandStore>
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 <O> void mapReduceConsume(
PreLoadContext context,
@ -100,13 +96,6 @@ public class AccordCommandStores extends CommandStores<AccordCommandStore>
});
}
@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<AccordCommandStore>
{
return 5 << 20; // TODO (required): make configurable
}
@Override
public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology)
{
Supplier<EpochReady> 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)
}
}

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

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

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

@ -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> partialDeps = localSerializer(DepsSerializer.partialDeps);
static final LocalVersionedSerializer<Writes> writes = localSerializer(CommandSerializers.writes);
static final LocalVersionedSerializer<TxnData> result = localSerializer(TxnData.serializer);
static final LocalVersionedSerializer<CommandListener> listeners = localSerializer(ListenerSerializers.listener);
static final LocalVersionedSerializer<Command.DurableAndIdempotentListener> listeners = localSerializer(ListenerSerializers.listener);
private static <T> LocalVersionedSerializer<T> localSerializer(IVersionedSerializer<T> 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())
{

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

@ -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<K, V>
private final K key;
private Object state = UNINITIALIZED;
private Set<Command.TransientListener> transientListeners;
public AccordLoadingState(K key)
{
@ -160,4 +166,27 @@ public class AccordLoadingState<K, V>
checkState(LoadingState.PENDING, false);
return (PendingLoad<V>) 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<Command.TransientListener> transientListeners()
{
if (transientListeners == null)
return Collections.emptySet();
return transientListeners;
}
}

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

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

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

@ -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<Tx
{
return invalidated;
}
@Override
public void addListener(Command.TransientListener listener)
{
global.addListener(listener);
}
@Override
public boolean removeListener(Command.TransientListener listener)
{
return global().removeListener(listener);
}
@Override
public Collection<Command.TransientListener> transientListeners()
{
return global.transientListeners();
}
}

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

@ -144,7 +144,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected Timestamp maxConflict(Seekables<?, ?> keysOrRanges, Ranges slice)
public Timestamp maxConflict(Seekables<?, ?> keysOrRanges, Ranges slice)
{
// TODO: Seekables
// TODO: efficiency

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

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

13
src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java
View File

@ -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<TxnId> txnIds(PreLoadContext context)
{
TxnId primaryid = context.primaryTxnId();
Collection<TxnId> additionalIds = context.additionalTxnIds();
if (primaryid == null) return additionalIds;
if (additionalIds.isEmpty()) return Collections.singleton(primaryid);
return Iterables.concat(Collections.singleton(primaryid), additionalIds);
}
private <K, V, S extends AccordSafeState<K, V>> void referenceAndAssembleReads(Iterable<K> keys,
Map<K, S> context,
AccordStateCache.Instance<K, V, S> cache,

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

@ -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<R> extends AsyncChains.Head<R> implements Runnable, Function<SafeCommandStore, R>
{
private static final Logger logger = LoggerFactory.getLogger(AsyncOperation.class);
@ -143,7 +145,7 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> 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

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

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

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

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

30
src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java
View File

@ -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<Command.Listener> commandListener = new IVersionedSerializer<Command.Listener>()
private static final IVersionedSerializer<Command.ProxyListener> commandListener = new IVersionedSerializer<Command.ProxyListener>()
{
@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<CommandListener> listener = new IVersionedSerializer<CommandListener>()
public static final IVersionedSerializer<Command.DurableAndIdempotentListener> listener = new IVersionedSerializer<Command.DurableAndIdempotentListener>()
{
@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);

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

@ -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<ReadData> request = new IVersionedSerializer<ReadData>()
public static final IVersionedSerializer<ReadTxnData> request = new IVersionedSerializer<ReadTxnData>()
{
@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);
}
};
}

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

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

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

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

1
src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
View File

@ -54,6 +54,7 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> 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;

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

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

3
test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java
View File

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

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

@ -114,7 +114,7 @@ public class AccordCommandStoreTest
attrs.partialDeps(dependencies);
ImmutableSortedSet<TxnId> waitingOnCommit = ImmutableSortedSet.of(oldTxnId1);
ImmutableSortedMap<Timestamp, TxnId > waitingOnApply = ImmutableSortedMap.of(oldTimestamp, oldTxnId2);
attrs.addListener(new Command.Listener(oldTxnId1));
attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt);
Command command = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted,
waitingOnCommit, waitingOnApply, result.left, result.right);

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

@ -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<Writes, Result> processTxnResult(AccordCommandStore commandStore, TxnId txnId, PartialTxn txn, Timestamp executeAt) throws Throwable
{
AtomicReference<Pair<Writes, Result>> 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)

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

@ -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<SafeCommandStore> consumer = safeStore -> safeStore.command(txnId).readyToExecute();
PreLoadContext ctx = PreLoadContext.contextFor(singleton(txnId), Keys.EMPTY);
PreLoadContext ctx = contextFor(txnId);
AsyncOperation<Void> 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<RoutableKey>) preLoadContext.keys()) {
return new AsyncLoader(commandStore, txnIds(preLoadContext), (Iterable<RoutableKey>) 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<SafeCommandStore> 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<RoutableKey>) preLoadContext.keys())
return new AsyncLoader(commandStore, txnIds(preLoadContext), (Iterable<RoutableKey>) preLoadContext.keys())
{
@Override
Function<TxnId, Command> 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<SafeCommandStore> 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<SafeCommandStore> 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);