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

Redesign progress mechanisms to be memory efficient, use fewer messages and to resolve dependency chains promptly. 

The SimpleProgressLog had a number of problems:

1. It polled for progress with no attempt to determine whether progress could realistically be made, so:
 - as the number of pending transactions grew, the proportion of useful work dropped (as many would be unable to make progress without earlier transactions completing)
 - each transaction in the chain could recover only on average 1/2 poll interval behind the last transaction to complete
2. It requested full transaction state from every replica on each attempt
3. It maintained a lot of in-memory state
4. Polling happened en-masse, allowing for little per-transaction control

We also separately maintained fairly expensive per-command listener state that negatively affected our command loading and caching.

The new DefaultProgressLog makes use of several new features: LocalListeners, RemoteListeners, Timers and Await messages.
 - LocalListeners provide a memory-efficient collection for managing each CommandStore<E2><80><99>s transaction listeners, with dedicated record keeping for inter-transaction relationships.
 - RemoteListeners provide a mechanism for request/response pairs that may be separated by longer than the normal Cassandra message timeout, and require minimal state on sender and recipient. This permits replicas to cheaply update their local state machine as soon as distributed information becomes available.

The DefaultProgressLog tracks each transaction with separate timers to handle per-transaction scheduling, backoff etc, and a succinct state machine. To reduce overhead correspondence is preferentially limited to a handful of replicas, and limited to the home shard where appropriate.

patch by Benedict; reviewed by Ariel Weisberg for CASSANDRA-19870
This commit is contained in:
Benedict Elliott Smith 2024-08-22 21:12:48 +01:00 committed by David Capwell
parent 1866b8ec73
commit 9ebce5d6df
69 changed files with 670 additions and 861 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
conf/cassandra.yaml
View File

@ -2659,8 +2659,8 @@ storage_compatibility_mode: NONE
# # The number of Accord shards on this node; -1 means use the number of cores
# shard_count: -1
#
# # Progress log scheduling delay
# progress_log_schedule_delay: 1s
# # Recover delay: the time between a transaction being initiated and a remote replica being willing to interrupt it to complete it
# recover_delay: 1s
#
# # how quickly the fast path is reconfigured when nodes go up/down
# fast_path_update_delay: 5s

2
modules/accord

@ -1 +1 @@
Subproject commit a171322f417c117733ca5b514d03a5202b1ac202
Subproject commit fb3efe9b8a87f0a182545791a2b0563690d52d00

2
src/java/org/apache/cassandra/config/AccordSpec.java
View File

@ -30,7 +30,7 @@ public class AccordSpec
public volatile OptionaldPositiveInt shard_count = OptionaldPositiveInt.UNDEFINED;
public volatile DurationSpec.IntMillisecondsBound progress_log_schedule_delay = new DurationSpec.IntMillisecondsBound(100);
public volatile DurationSpec.IntMillisecondsBound recover_delay = new DurationSpec.IntMillisecondsBound(1000);
/**
* When a barrier transaction is requested how many times to repeat attempting the barrier before giving up

4
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -96,6 +96,7 @@ import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.invalidated;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.maybeDropTruncatedCommandColumns;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.truncatedApply;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeysAccessor;
@ -849,6 +850,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
// We can still encounter sliced command state just because compaction inputs are random
return BTreeRow.emptyDeletedRow(row.clustering(), new Row.Deletion(DeletionTime.build(row.primaryKeyLivenessInfo().timestamp(), nowInSec), false));
case INVALIDATE:
return invalidated(cleanup.appliesIfNot, row, nowInSec);
case TRUNCATE_WITH_OUTCOME:
case TRUNCATE:
if (saveStatus.compareTo(cleanup.appliesIfNot) >= 0)

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

@ -206,7 +206,7 @@ public class AccordMetrics
@Override
public void onStable(Command cmd)
{
long now = AccordService.uniqueNow();
long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@ -218,7 +218,7 @@ public class AccordMetrics
@Override
public void onExecuted(Command cmd)
{
long now = AccordService.uniqueNow();
long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@ -232,7 +232,7 @@ public class AccordMetrics
@Override
public void onApplied(Command cmd, long applyStartTimestamp)
{
long now = AccordService.uniqueNow();
long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@ -270,7 +270,7 @@ public class AccordMetrics
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
{
long now = AccordService.uniqueNow();
long now = AccordService.now();
metrics.recoveryDuration.update(now - recoveryTimestamp.hlc(), MICROSECONDS);
metrics.recoveryDelay.update(recoveryTimestamp.hlc() - txnId.hlc(), MICROSECONDS);

21
src/java/org/apache/cassandra/net/Verb.java
View File

@ -92,18 +92,16 @@ import org.apache.cassandra.service.accord.serializers.CheckStatusSerializers;
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.GetDepsSerializers;
import org.apache.cassandra.service.accord.serializers.CalculateDepsSerializers;
import org.apache.cassandra.service.accord.serializers.GetEphmrlReadDepsSerializers;
import org.apache.cassandra.service.accord.serializers.GetMaxConflictSerializers;
import org.apache.cassandra.service.accord.serializers.InformDurableSerializers;
import org.apache.cassandra.service.accord.serializers.InformHomeDurableSerializers;
import org.apache.cassandra.service.accord.serializers.InformOfTxnIdSerializers;
import org.apache.cassandra.service.accord.serializers.PreacceptSerializers;
import org.apache.cassandra.service.accord.serializers.QueryDurableBeforeSerializers;
import org.apache.cassandra.service.accord.serializers.ReadDataSerializers;
import org.apache.cassandra.service.accord.serializers.RecoverySerializers;
import org.apache.cassandra.service.accord.serializers.SetDurableSerializers;
import org.apache.cassandra.service.accord.serializers.WaitOnCommitSerializer;
import org.apache.cassandra.service.accord.serializers.AwaitSerializer;
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState;
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.ConsensusKeyMigrationFinished;
import org.apache.cassandra.service.paxos.Commit;
@ -321,21 +319,20 @@ public enum Verb
ACCORD_COMMIT_REQ (127, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_COMMIT_INVALIDATE_REQ (128, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.invalidate, AccordService::verbHandlerOrNoop ),
ACCORD_APPLY_RSP (129, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.reply, RESPONSE_HANDLER ),
ACCORD_APPLY_REQ (130, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_APPLY_RSP ),
ACCORD_APPLY_REQ (130, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_APPLY_RSP ),
ACCORD_BEGIN_RECOVER_RSP (131, P2, writeTimeout, IMMEDIATE, () -> RecoverySerializers.reply, RESPONSE_HANDLER ),
ACCORD_BEGIN_RECOVER_REQ (132, P2, writeTimeout, IMMEDIATE, () -> RecoverySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_BEGIN_RECOVER_RSP ),
ACCORD_BEGIN_INVALIDATE_RSP (133, P2, writeTimeout, IMMEDIATE, () -> BeginInvalidationSerializers.reply, RESPONSE_HANDLER ),
ACCORD_BEGIN_INVALIDATE_REQ (134, P2, writeTimeout, IMMEDIATE, () -> BeginInvalidationSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_BEGIN_INVALIDATE_RSP ),
ACCORD_WAIT_ON_COMMIT_RSP (136, P2, writeTimeout, IMMEDIATE, () -> WaitOnCommitSerializer.reply, RESPONSE_HANDLER ),
ACCORD_WAIT_ON_COMMIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> WaitOnCommitSerializer.request, AccordService::verbHandlerOrNoop, ACCORD_WAIT_ON_COMMIT_RSP ),
ACCORD_WAIT_UNTIL_APPLIED_REQ (137, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_INFORM_OF_TXN_REQ (138, P2, writeTimeout, IMMEDIATE, () -> InformOfTxnIdSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_HOME_DURABLE_REQ (139, P2, writeTimeout, IMMEDIATE, () -> InformHomeDurableSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_AWAIT_RSP (136, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.syncReply, RESPONSE_HANDLER ),
ACCORD_AWAIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.request, AccordService::verbHandlerOrNoop, ACCORD_AWAIT_RSP ),
ACCORD_AWAIT_ASYNC_RSP_REQ (137, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.asyncReply, AccordService::verbHandlerOrNoop ),
ACCORD_WAIT_UNTIL_APPLIED_REQ (138, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_INFORM_DURABLE_REQ (140, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.reply, RESPONSE_HANDLER ),
ACCORD_CHECK_STATUS_REQ (142, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ),
ACCORD_GET_DEPS_RSP (143, P2, writeTimeout, IMMEDIATE, () -> GetDepsSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_DEPS_REQ (144, P2, writeTimeout, IMMEDIATE, () -> GetDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_GET_DEPS_RSP ),
ACCORD_CALCULATE_DEPS_RSP (143, P2, writeTimeout, IMMEDIATE, () -> CalculateDepsSerializers.reply, RESPONSE_HANDLER ),
ACCORD_CALCULATE_DEPS_REQ (144, P2, writeTimeout, IMMEDIATE, () -> CalculateDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_CALCULATE_DEPS_RSP),
ACCORD_GET_EPHMRL_READ_DEPS_RSP (161, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_EPHMRL_READ_DEPS_REQ (162, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_GET_EPHMRL_READ_DEPS_RSP),
ACCORD_GET_MAX_CONFLICT_RSP (163, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.reply, RESPONSE_HANDLER ),

34
src/java/org/apache/cassandra/service/accord/AccordCachingState.java
View File

@ -25,8 +25,6 @@ import java.util.function.ToLongFunction;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.local.Command.TransientListener;
import accord.local.Listeners;
import accord.utils.IntrusiveLinkedListNode;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@ -69,11 +67,6 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
final byte index;
private boolean shouldUpdateSize;
/**
* Transient listeners aren't meant to survive process restart, but must survive cache eviction.
*/
private Listeners<TransientListener> transientListeners;
AccordCachingState(K key, int index)
{
this.key = key;
@ -157,33 +150,6 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
return complete().status();
}
public void addListener(TransientListener listener)
{
if (transientListeners == null)
transientListeners = new Listeners<>();
transientListeners.add(listener);
}
public boolean removeListener(TransientListener listener)
{
return transientListeners != null && transientListeners.remove(listener);
}
public void listeners(Listeners<TransientListener> listeners)
{
transientListeners = listeners;
}
public Listeners<TransientListener> listeners()
{
return transientListeners == null ? Listeners.EMPTY : transientListeners;
}
public boolean hasListeners()
{
return !listeners().isEmpty();
}
State<K, V> complete()
{
return state.isCompleteable() ? state(state.complete()) : state;

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

@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
@ -114,11 +115,12 @@ public class AccordCommandStore extends CommandStore implements CacheSize
Agent agent,
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
LocalListeners.Factory listenerFactory,
EpochUpdateHolder epochUpdateHolder,
IJournal journal,
AccordStateCacheMetrics cacheMetrics)
{
this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics);
this(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics);
}
private static <K, V> void registerJfrListener(int id, AccordStateCache.Instance<K, V, ?> instance, String name)
@ -194,13 +196,14 @@ public class AccordCommandStore extends CommandStore implements CacheSize
Agent agent,
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
LocalListeners.Factory listenerFactory,
EpochUpdateHolder epochUpdateHolder,
IJournal journal,
ExecutorPlus loadExecutor,
ExecutorPlus saveExecutor,
AccordStateCacheMetrics cacheMetrics)
{
super(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder);
super(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
this.journal = journal;
loggingId = String.format("[%s]", id);
executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']');
@ -257,8 +260,8 @@ public class AccordCommandStore extends CommandStore implements CacheSize
static Factory factory(AccordJournal journal, AccordStateCacheMetrics cacheMetrics)
{
return (id, time, agent, dataStore, progressLogFactory, rangesForEpoch) ->
new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal, cacheMetrics);
return (id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch) ->
new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, cacheMetrics);
}
public CommandsForRangesLoader diskCommandsForRanges()

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

@ -22,6 +22,7 @@ import java.util.function.Supplier;
import accord.api.Agent;
import accord.api.ConfigurationService.EpochReady;
import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.local.CommandStores;
import accord.local.Node;
@ -45,17 +46,19 @@ public class AccordCommandStores extends CommandStores implements CacheSize
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, AccordJournal journal)
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenerFactory,
AccordJournal journal)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
super(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory,
AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
setCapacity(DatabaseDescriptor.getAccordCacheSizeInMiB() << 20);
this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this);
}
static Factory factory(AccordJournal journal)
{
return (time, agent, store, random, shardDistributor, progressLogFactory) ->
new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, journal);
return (time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory) ->
new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory, journal);
}
@Override

10
src/java/org/apache/cassandra/service/accord/AccordConfiguration.java
View File

@ -18,9 +18,7 @@
package org.apache.cassandra.service.accord;
import java.time.Duration;
import accord.config.LocalConfig;
import accord.api.LocalConfig;
import org.apache.cassandra.config.Config;
// TODO (expected): should this be merged with AccordSpec?
@ -32,10 +30,4 @@ public class AccordConfiguration implements LocalConfig
{
this.config = config;
}
@Override
public Duration getProgressLogScheduleDelay()
{
return config.accord.progress_log_schedule_delay.toDuration();
}
}

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

@ -35,7 +35,6 @@ import org.slf4j.LoggerFactory;
import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.Node;
import accord.messages.LocalRequest;
import accord.messages.ReplyContext;
import accord.messages.Request;
import accord.primitives.TxnId;
@ -175,19 +174,6 @@ public class AccordJournal implements IJournal, Shutdownable
delayedRequestProcessor.delay(requestContext);
}
/**
* Accord protocol messages originating from local node, e.g. Propagate.
*/
@SuppressWarnings("rawtypes, unchecked")
public <R> void processLocalRequest(LocalRequest request, BiConsumer<? super R, Throwable> callback)
{
LocalRequestContext requestContext = LocalRequestContext.create(request, callback);
if (node.topology().hasEpoch(request.waitForEpoch()))
request.process(node, requestContext.callback);
else
delayedRequestProcessor.delay(requestContext);
}
@Override
public Command loadCommand(int commandStoreId, TxnId txnId)
{
@ -281,29 +267,6 @@ public class AccordJournal implements IJournal, Shutdownable
public abstract void process(Node node, AccordEndpointMapper endpointMapper);
}
private static class LocalRequestContext<T> extends RequestContext
{
private final BiConsumer<T, Throwable> callback;
private final LocalRequest<T> request;
LocalRequestContext(long waitForEpoch, LocalRequest<T> request, BiConsumer<T, Throwable> callback)
{
super(waitForEpoch);
this.callback = callback;
this.request = request;
}
public void process(Node node, AccordEndpointMapper endpointMapper)
{
request.process(node, callback);
}
static <R> LocalRequestContext<R> create(LocalRequest<R> request, BiConsumer<R, Throwable> callback)
{
return new LocalRequestContext<>(request.waitForEpoch(), request, callback);
}
}
/**
* Barebones response context not holding a reference to the entire message
*/

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

@ -53,7 +53,6 @@ import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.DurableBefore;
import accord.local.Listeners;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.SaveStatus;
@ -146,7 +145,6 @@ import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers;
import org.apache.cassandra.service.accord.serializers.CommandsForKeySerializer;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.ListenerSerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.Clock.Global;
@ -250,7 +248,6 @@ public class AccordKeyspace
public static class LocalVersionedSerializers
{
static final LocalVersionedSerializer<Route<?>> route = localSerializer(KeySerializers.route);
static final LocalVersionedSerializer<Command.DurableAndIdempotentListener> listeners = localSerializer(ListenerSerializers.listener);
static final LocalVersionedSerializer<Topology> topology = localSerializer(TopologySerializers.topology);
static final LocalVersionedSerializer<ReducingRangeMap<Timestamp>> rejectBefore = localSerializer(CommandStoreSerializers.rejectBefore);
static final LocalVersionedSerializer<DurableBefore> durableBefore = localSerializer(CommandStoreSerializers.durableBefore);
@ -284,6 +281,7 @@ public class AccordKeyspace
public static final ColumnMetadata execute_at = getColumn(Commands, "execute_at");
public static final ColumnMetadata[] TRUNCATE_FIELDS = new ColumnMetadata[] { durability, execute_at, route, status };
public static final ColumnMetadata[] INVALIDATE_FIELDS = new ColumnMetadata[] { status };
static
{
@ -368,6 +366,31 @@ public class AccordKeyspace
return newLeaf;
}
public static Row invalidated(SaveStatus newSaveStatus, Row row, long nowInSec)
{
long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
long newTimestamp = oldTimestamp + 1;
Object[] newLeaf = invalidatedLeaf(newTimestamp, newSaveStatus);
// Including a deletion allows future compactions to drop data before it gets to the purger
// but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns
// regardless
Row.Deletion deletion = new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false);
return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), deletion, newLeaf);
}
private static Object[] invalidatedLeaf(long newTimestamp, SaveStatus newSaveStatus)
{
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(INVALIDATE_FIELDS.length);
int colIndex = 0;
// Status always needs to use the new timestamp since we are replacing the existing value
// All the other columns are being retained unmodified with at most updated timestamps to accomdate deletion
//noinspection UnusedAssignment
newLeaf[colIndex++] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
return newLeaf;
}
public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, Cell<?> durabilityCell, Cell<?> executeAtCell, Cell<?> routeCell, boolean withOutcome)
{
checkArgument(durabilityCell.column() == CommandsColumns.durability);
@ -624,7 +647,7 @@ public class AccordKeyspace
if (current == null)
return null;
CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore);
CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore.shardRedundantBefore());
if (current == updated)
return row;
@ -1138,18 +1161,6 @@ public class AccordKeyspace
}
}
private static Listeners.Immutable deserializeListeners(UntypedResultSet.Row row) throws IOException
{
Set<ByteBuffer> serialized = row.getSet("listeners", BytesType.instance);
if (serialized == null || serialized.isEmpty())
return Listeners.Immutable.EMPTY;
Listeners<Command.DurableAndIdempotentListener> result = new Listeners<>();
for (ByteBuffer bytes : serialized)
result.add(deserialize(bytes, LocalVersionedSerializers.listeners));
return new Listeners.Immutable(result);
}
public static PartitionKey deserializeKey(ByteBuffer buffer)
{
List<ByteBuffer> split = KEY_TYPE.unpack(buffer, ByteBufferAccessor.instance);

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

@ -122,8 +122,8 @@ public class AccordMessageSink implements MessageSink
builder.put(MessageType.ACCEPT_REQ, Verb.ACCORD_ACCEPT_REQ);
builder.put(MessageType.ACCEPT_RSP, Verb.ACCORD_ACCEPT_RSP);
builder.put(MessageType.ACCEPT_INVALIDATE_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ);
builder.put(MessageType.GET_DEPS_REQ, Verb.ACCORD_GET_DEPS_REQ);
builder.put(MessageType.GET_DEPS_RSP, Verb.ACCORD_GET_DEPS_RSP);
builder.put(MessageType.CALCULATE_DEPS_REQ, Verb.ACCORD_CALCULATE_DEPS_REQ);
builder.put(MessageType.CALCULATE_DEPS_RSP, Verb.ACCORD_CALCULATE_DEPS_RSP);
builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_REQ, Verb.ACCORD_GET_EPHMRL_READ_DEPS_REQ);
builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_RSP, Verb.ACCORD_GET_EPHMRL_READ_DEPS_RSP);
builder.put(MessageType.GET_MAX_CONFLICT_REQ, Verb.ACCORD_GET_MAX_CONFLICT_REQ);
@ -144,13 +144,12 @@ public class AccordMessageSink implements MessageSink
builder.put(MessageType.BEGIN_RECOVER_RSP, Verb.ACCORD_BEGIN_RECOVER_RSP);
builder.put(MessageType.BEGIN_INVALIDATE_REQ, Verb.ACCORD_BEGIN_INVALIDATE_REQ);
builder.put(MessageType.BEGIN_INVALIDATE_RSP, Verb.ACCORD_BEGIN_INVALIDATE_RSP);
builder.put(MessageType.WAIT_ON_COMMIT_REQ, Verb.ACCORD_WAIT_ON_COMMIT_REQ);
builder.put(MessageType.WAIT_ON_COMMIT_RSP, Verb.ACCORD_WAIT_ON_COMMIT_RSP);
builder.put(MessageType.AWAIT_REQ, Verb.ACCORD_AWAIT_REQ);
builder.put(MessageType.AWAIT_RSP, Verb.ACCORD_AWAIT_RSP);
builder.put(MessageType.ASYNC_AWAIT_COMPLETE_REQ, Verb.ACCORD_AWAIT_ASYNC_RSP_REQ);
builder.put(MessageType.WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_WAIT_UNTIL_APPLIED_REQ);
builder.put(MessageType.APPLY_THEN_WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_APPLY_AND_WAIT_REQ);
builder.put(MessageType.INFORM_OF_TXN_REQ, Verb.ACCORD_INFORM_OF_TXN_REQ);
builder.put(MessageType.INFORM_DURABLE_REQ, Verb.ACCORD_INFORM_DURABLE_REQ);
builder.put(MessageType.INFORM_HOME_DURABLE_REQ, Verb.ACCORD_INFORM_HOME_DURABLE_REQ);
builder.put(MessageType.CHECK_STATUS_REQ, Verb.ACCORD_CHECK_STATUS_REQ);
builder.put(MessageType.CHECK_STATUS_RSP, Verb.ACCORD_CHECK_STATUS_RSP);
builder.put(MessageType.FETCH_DATA_REQ, Verb.ACCORD_FETCH_DATA_REQ);

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

@ -69,7 +69,7 @@ import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.apache.cassandra.utils.ObjectSizes;
import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
import static accord.primitives.TxnId.NO_TXNIDS;
import static org.apache.cassandra.utils.ObjectSizes.measure;
public class AccordObjectSizes
@ -266,14 +266,6 @@ public class AccordObjectSizes
return ((TxnResult) result).estimatedSizeOnHeap();
}
private static final long EMPTY_COMMAND_LISTENER = measure(new Command.ProxyListener(null));
public static long listener(Command.DurableAndIdempotentListener listener)
{
if (listener instanceof Command.ProxyListener)
return EMPTY_COMMAND_LISTENER + timestamp(((Command.ProxyListener) listener).txnId());
throw new IllegalArgumentException("Unhandled listener type: " + listener.getClass());
}
private static class CommandEmptySizes
{
private final static TokenKey EMPTY_KEY = new TokenKey(EMPTY_ID, null);
@ -301,7 +293,7 @@ public class AccordObjectSizes
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_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));
final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID));
private static long emptySize(Command command)
{
@ -343,8 +335,6 @@ public class AccordObjectSizes
long size = CommandEmptySizes.emptySize(command);
size += sizeNullable(command.route(), AccordObjectSizes::route);
size += sizeNullable(command.promised(), AccordObjectSizes::timestamp);
for (Command.DurableAndIdempotentListener listener : command.durableListeners())
size += listener(listener);
size += sizeNullable(command.executeAt(), AccordObjectSizes::timestamp);
size += sizeNullable(command.partialTxn(), AccordObjectSizes::txn);
size += sizeNullable(command.partialDeps(), AccordObjectSizes::dependencies);

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

@ -24,8 +24,6 @@ import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
import accord.local.Command;
import accord.local.Command.TransientListener;
import accord.local.Listeners;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@ -152,27 +150,6 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
return invalidated;
}
@Override
public void addListener(TransientListener listener)
{
checkNotInvalidated();
global.addListener(listener);
}
@Override
public boolean removeListener(TransientListener listener)
{
checkNotInvalidated();
return global.removeListener(listener);
}
@Override
public Listeners<TransientListener> transientListeners()
{
checkNotInvalidated();
return global.listeners();
}
public static Function<AccordCachingState<TxnId, Command>, AccordSafeCommand> safeRefFactory()
{
return Invariants.testParanoia(LINEAR, LINEAR, HIGH) ? DebugAccordSafeCommand::new : AccordSafeCommand::new;

34
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -27,7 +27,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
@ -37,7 +36,6 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.primitives.Ints;
@ -45,8 +43,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.BarrierType;
import accord.api.LocalConfig;
import accord.api.Result;
import accord.config.LocalConfig;
import accord.coordinate.Barrier;
import accord.coordinate.CoordinateSyncPoint;
import accord.coordinate.CoordinationFailed;
@ -57,10 +55,13 @@ import accord.coordinate.Timeout;
import accord.coordinate.TopologyMismatch;
import accord.impl.AbstractConfigurationService;
import accord.impl.CoordinateDurabilityScheduling;
import accord.impl.SimpleProgressLog;
import accord.impl.DefaultLocalListeners;
import accord.impl.DefaultRemoteListeners;
import accord.impl.DefaultRequestTimeouts;
import accord.impl.SizeOfIntersectionSorter;
import accord.local.Command;
import accord.local.CommandStore;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.CommandStores;
import accord.local.DurableBefore;
import accord.local.KeyHistory;
@ -73,7 +74,6 @@ import accord.local.SaveStatus;
import accord.local.ShardDistributor.EvenSplit;
import accord.local.Status;
import accord.local.cfk.CommandsForKey;
import accord.messages.LocalRequest;
import accord.messages.Request;
import accord.primitives.Keys;
import accord.primitives.Ranges;
@ -340,18 +340,11 @@ public class AccordService implements IAccordService, Shutdownable
return i;
}
public static long uniqueNow()
public static long now()
{
// TODO (now, correctness): This is not unique it's just currentTimeMillis as microseconds
return TimeUnit.MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
}
public static long unix(TimeUnit timeUnit)
{
Preconditions.checkArgument(timeUnit != TimeUnit.NANOSECONDS, "Nanoseconds since the epoch doesn't fit in a long");
return timeUnit.convert(Clock.Global.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
private AccordService(Id localId)
{
Invariants.checkState(localId != null, "static localId must be set before instantiating AccordService");
@ -366,9 +359,8 @@ public class AccordService implements IAccordService, Shutdownable
this.journal = new AccordJournal(configService, DatabaseDescriptor.getAccord().journal);
this.node = new Node(localId,
messageSink,
this::handleLocalRequest,
configService,
AccordService::uniqueNow,
AccordService::now,
NodeTimeService.elapsedWrapperFromMonotonicSource(NANOSECONDS, Clock.Global::nanoTime),
() -> dataStore,
new KeyspaceSplitter(new EvenSplit<>(DatabaseDescriptor.getAccordShardCount(), getPartitioner().accordSplitter())),
@ -377,7 +369,10 @@ public class AccordService implements IAccordService, Shutdownable
scheduler,
CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER,
new AccordTopologySorter.Supplier(configService, DatabaseDescriptor.getNodeProximity())),
SimpleProgressLog::new,
DefaultRemoteListeners::new,
DefaultRequestTimeouts::new,
DefaultProgressLogs::new,
DefaultLocalListeners.Factory::new,
AccordCommandStores.factory(journal),
new AccordInteropFactory(agent, configService),
configuration);
@ -756,13 +751,6 @@ public class AccordService implements IAccordService, Shutdownable
}
}
private <R> void handleLocalRequest(LocalRequest<R> request, BiConsumer<? super R, Throwable> callback, Node node)
{
// currently, we only create LocalRequests that have side effects and need to be persisted
Invariants.checkState(request.type().hasSideEffects());
journal.processLocalRequest(request, callback);
}
private static RequestTimeoutException newTimeout(TxnId txnId, boolean isWrite, ConsistencyLevel consistencyLevel)
{
// Client protocol doesn't handle null consistency level so use ANY

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

@ -216,18 +216,11 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
instance.validateLoadEvicted(node);
if (!node.hasListeners())
{
AccordCachingState<?, ?> self = instances.get(node.index).cache.remove(node.key());
Invariants.checkState(self.references == 0);
checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
if (instance.listeners != null)
instance.listeners.forEach(l -> l.onEvict((AccordCachingState) node));
}
else
{
node.markEvicted(); // keep the node in the cache to prevent transient listeners from being GCd
}
AccordCachingState<?, ?> self = instances.get(node.index).cache.remove(node.key());
Invariants.checkState(self.references == 0);
checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
if (instance.listeners != null)
instance.listeners.forEach(l -> l.onEvict((AccordCachingState) node));
}
public ImmutableStats stats()

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

@ -113,7 +113,7 @@ public class AccordSyncPropagator
if (closed.containsAll(addClosed))
return null;
addClosed = addClosed.subtract(closed);
addClosed = addClosed.without(closed);
closed = closed.with(addClosed);
return new Notification(epoch, Collections.emptySet(), addClosed, Ranges.EMPTY);
}
@ -123,7 +123,7 @@ public class AccordSyncPropagator
if (redundant.containsAll(addRedundant))
return null;
addRedundant = addRedundant.subtract(redundant);
addRedundant = addRedundant.without(redundant);
redundant = redundant.with(addRedundant);
return new Notification(epoch, Collections.emptySet(), Ranges.EMPTY, addRedundant);
}
@ -140,8 +140,8 @@ public class AccordSyncPropagator
if (notification.syncComplete.containsAll(syncComplete)) syncComplete = ImmutableSet.of();
else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.syncComplete.contains(v)));
}
closed = closed.subtract(notification.closed);
redundant = redundant.subtract(notification.redundant);
closed = closed.without(notification.closed);
redundant = redundant.without(notification.redundant);
return syncComplete.isEmpty() && closed.isEmpty() && redundant.isEmpty();
}

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

@ -233,7 +233,7 @@ public class CommandsForRangesLoader
accum.add(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end));
return accum;
}, new ArrayList<Range>(), ignore -> false).toArray(Range[]::new));
Ranges newRanges = ranges.subtract(durableAlready);
Ranges newRanges = ranges.without(durableAlready);
if (newRanges.isEmpty())
return null;

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

@ -29,7 +29,6 @@ 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;
@ -69,7 +68,6 @@ public class SavedCommand
ADDITIONAL_KEYS,
WAITING_ON,
WRITES,
LISTENERS
}
public interface Writer<K> extends Journal.Writer
@ -179,13 +177,6 @@ public class SavedCommand
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);
}
}
@VisibleForTesting
@ -210,7 +201,6 @@ public class SavedCommand
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;
}
@ -289,7 +279,6 @@ public class SavedCommand
SavedCommand.WaitingOnProvider waitingOn = (txn, deps) -> null;
Writes writes = null;
Listeners.Immutable<?> listeners = null;
Result result = CommandSerializers.APPLIED;
boolean nextCalled = false;
@ -427,22 +416,6 @@ public class SavedCommand
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)
@ -469,8 +442,6 @@ public class SavedCommand
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)
@ -518,7 +489,7 @@ public class SavedCommand
case Erased:
return Command.Truncated.erased(attrs.txnId(), attrs.durability(), attrs.route());
case Invalidated:
return Command.Truncated.invalidated(attrs.txnId(), attrs.durableListeners());
return Command.Truncated.invalidated(attrs.txnId());
}
}
@ -537,7 +508,6 @@ public class SavedCommand
", additionalKeysOrRanges=" + additionalKeysOrRanges +
", waitingOn=" + waitingOn +
", writes=" + writes +
", listeners=" + listeners +
'}';
}
}

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

@ -21,29 +21,45 @@ package org.apache.cassandra.service.accord.api;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.EventsListener;
import accord.api.ProgressLog.BlockedUntil;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.Node;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.messages.ReplyContext;
import accord.primitives.Ranges;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind;
import accord.primitives.TxnId;
import accord.topology.Shard;
import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.RandomSource;
import accord.utils.SortedList;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.metrics.AccordMetrics;
import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static accord.primitives.Routable.Domain.Key;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getReadRpcTimeout;
import static org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.maybeSaveAccordKeyMigrationLocally;
@ -55,6 +71,7 @@ public class AccordAgent implements Agent
// TODO (required): this should be configurable and have exponential back-off, escaping to operator input past a certain number of retries
private long retryBootstrapDelayMicros = SECONDS.toMicros(1L);
private final RandomSource random = new DefaultRandom();
public void setRetryBootstrapDelay(long delay, TimeUnit units)
{
@ -147,4 +164,70 @@ public class AccordAgent implements Agent
{
return AccordMetrics.Listener.instance;
}
@Override
public long replyTimeout(ReplyContext replyContext, TimeUnit units)
{
return Math.max(1, units.convert(((ResponseContext)replyContext).expiresAtNanos() - Clock.Global.nanoTime(), NANOSECONDS));
}
@Override
public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
{
SafeCommand safeCommand = safeStore.ifInitialised(txnId);
Invariants.nonNull(safeCommand);
Command command = safeCommand.current();
Invariants.nonNull(command);
Timestamp mostRecentAttempt = Timestamp.max(command.txnId(), command.promised());
RoutingKey homeKey = command.route().homeKey();
Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch());
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long oneSecond = SECONDS.toMicros(1L);
long startTime = mostRecentAttempt.hlc() + DatabaseDescriptor.getAccord().recover_delay.to(MICROSECONDS)
+ (retryCount == 0 ? 0 : random.nextLong(oneSecond << Math.min(retryCount, 4)));
startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), oneSecond, random);
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
return units.convert(Math.max(1, startTime - nowMicros), MICROSECONDS);
}
@VisibleForTesting
public static long nonClashingStartTime(long startTime, SortedList<Node.Id> nodes, Node.Id id, long granularity, RandomSource random)
{
long perSecondStartTime;
if (nodes != null)
{
int position = nodes.indexOf(id);
perSecondStartTime = position * (SECONDS.toMicros(1) / nodes.size());
}
else
{
// we've raced with topology update, this should be rare so just pick a random start time
perSecondStartTime = random.nextLong(granularity);
}
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long subSecondRemainder = startTime % granularity;
long newStartTime = startTime - subSecondRemainder + perSecondStartTime;
if (newStartTime < startTime)
newStartTime += granularity;
return newStartTime;
}
@Override
public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
{
// TODO (required): make this configurable and dependent upon normal request latencies, and perhaps offset from txnId.hlc()
return units.convert((1L << Math.min(retryCount, 4)), SECONDS);
}
@Override
public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
{
// TODO (expected): integrate with contention backoff
return units.convert((1L << Math.min(retryCount, 4)), SECONDS);
}
}

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

@ -46,6 +46,12 @@ public class AccordScheduler implements Scheduler, Shutdownable
{
future.cancel(false);
}
@Override
public boolean isDone()
{
return future.isDone();
}
}
@Override

35
src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
View File

@ -21,10 +21,10 @@ package org.apache.cassandra.service.accord.interop;
import java.util.BitSet;
import javax.annotation.Nullable;
import accord.api.LocalListeners;
import accord.api.Result;
import accord.local.Command;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.Status;
@ -34,8 +34,8 @@ import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -47,9 +47,9 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType;
import org.apache.cassandra.service.accord.serializers.ApplySerializers.ApplySerializer;
import org.apache.cassandra.service.accord.txn.AccordUpdate;
import org.jctools.queues.MpscChunkedArrayQueue;
import static accord.utils.Invariants.checkState;
import static accord.utils.MapReduceConsume.forEach;
import static com.google.common.base.Preconditions.checkArgument;
/**
@ -58,7 +58,7 @@ import static com.google.common.base.Preconditions.checkArgument;
* // and these all are a bit copy pasta in terms of managing things like waiting on, obsoletion, cancellation/listeners, insufficient etc. and it would be less fragile
* // in the long run to not duplicate these kind of difficult to get right mechanism and have a single pluggable framework to request each specific behavior
*/
public class AccordInteropApply extends Apply implements Command.TransientListener
public class AccordInteropApply extends Apply implements LocalListeners.ComplexListener
{
public static final Apply.Factory FACTORY = new Apply.Factory()
{
@ -77,7 +77,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
public static final IVersionedSerializer<AccordInteropApply> serializer = new ApplySerializer<AccordInteropApply>()
{
@Override
protected AccordInteropApply deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
protected AccordInteropApply deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
return new AccordInteropApply(kind, txnId, scope, waitForEpoch, keys, executeAt, deps, txn, fullRoute, writes, result);
}
@ -85,8 +85,9 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
transient BitSet waitingOn;
transient int waitingOnCount;
final MpscChunkedArrayQueue<LocalListeners.Registered> listeners = new MpscChunkedArrayQueue<>(4, 1 << 30);
private AccordInteropApply(Kind kind, TxnId txnId, PartialRoute<?> route, long waitForEpoch, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
private AccordInteropApply(Kind kind, TxnId txnId, Route<?> route, long waitForEpoch, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
super(kind, txnId, route, waitForEpoch, keys, executeAt, deps, txn, fullRoute, writes, result);
}
@ -137,7 +138,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
waitingOn.set(safeStore.commandStore().id());
++waitingOnCount;
}
safeCommand.addListener(this);
listeners.add(safeStore.register(txnId, this));
break;
case Applied:
@ -190,11 +191,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
private void cancel()
{
node.commandStores().mapReduceConsume(this, waitingOn.stream(), forEach(safeStore -> {
SafeCommand safeCommand = safeStore.ifInitialised(txnId);
if (safeCommand != null)
safeCommand.removeListener(this);
}, node.agent()));
listeners.drain(LocalListeners.Registered::cancel);
}
@Override
@ -234,7 +231,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
}
@Override
public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand)
{
Command command = safeCommand.current();
@ -248,20 +245,14 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
case PreCommitted:
case Committed:
case PreApplied:
return;
return true;
case Applied:
case Invalidated:
case Truncated:
}
if (safeCommand.removeListener(this))
ack();
}
@Override
public PreLoadContext listenerPreLoadContext(TxnId caller)
{
return PreLoadContext.contextFor(txnId);
ack();
return false;
}
}

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

@ -29,8 +29,8 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -46,13 +46,13 @@ public class AccordInteropCommit extends Commit
public static final IVersionedSerializer<AccordInteropCommit> serializer = new CommitSerializer<AccordInteropCommit, AccordInteropRead>(AccordInteropRead.class, AccordInteropRead.requestSerializer)
{
@Override
protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
protected AccordInteropCommit deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, read);
}
};
public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
public AccordInteropCommit(Kind kind, TxnId txnId, Route<?> scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
{
super(kind, txnId, scope, waitForEpoch, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, readData);
}

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

@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.Data;
import accord.api.Result;
import accord.coordinate.CoordinationAdapter;
import accord.local.AgentExecutor;
import accord.local.CommandStore;
import accord.local.Node;
@ -93,7 +94,6 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.Dispatcher;
import static accord.coordinate.CoordinationAdapter.Factory.Step.Continue;
import static accord.coordinate.CoordinationAdapter.Invoke.persist;
import static accord.utils.Invariants.checkArgument;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteMetrics;
@ -345,7 +345,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
CommandStore cs = node.commandStores().select(route.homeKey());
result.beginAsResult().withExecutor(cs).begin((data, failure) -> {
if (failure == null)
persist(node.coordinationAdapter(txnId, Continue), node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
((CoordinationAdapter)node.coordinationAdapter(txnId, Continue)).persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
else
callback.accept(null, failure);
});

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

@ -22,7 +22,7 @@ import java.util.function.BiConsumer;
import accord.api.Result;
import accord.coordinate.Persist;
import accord.coordinate.tracking.AppliedTracker;
import accord.coordinate.tracking.AllTracker;
import accord.coordinate.tracking.QuorumTracker;
import accord.coordinate.tracking.RequestStatus;
import accord.coordinate.tracking.ResponseTracker;
@ -110,7 +110,7 @@ public class AccordInteropPersist extends Persist
callback = new CallbackHolder(new QuorumTracker(topologies), result, clientCallback);
break;
case ALL:
callback = new CallbackHolder(new AppliedTracker(topologies), result, clientCallback);
callback = new CallbackHolder(new AllTracker(topologies), result, clientCallback);
break;
default:
throw new IllegalArgumentException("Unhandled consistency level: " + consistencyLevel);

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

@ -108,7 +108,7 @@ public class AccordInteropReadRepair extends ReadData
private final Mutation mutation;
private static final IVersionedSerializer<Data> noop_data_serializer = new IVersionedSerializer<Data>()
private static final IVersionedSerializer<Data> noop_data_serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(Data t, DataOutputPlus out, int version) throws IOException {}

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

@ -22,7 +22,7 @@ import java.io.IOException;
import accord.messages.Accept;
import accord.messages.Accept.AcceptReply;
import accord.primitives.PartialRoute;
import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@ -47,9 +47,9 @@ public class AcceptSerializers
}
@Override
public Accept deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
public Accept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
return create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey,
return create(txnId, scope, waitForEpoch, minEpoch,
CommandSerializers.ballot.deserialize(in, version),
CommandSerializers.timestamp.deserialize(in, version),
KeySerializers.seekables.deserialize(in, version),

9
src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
View File

@ -24,8 +24,8 @@ import accord.api.Result;
import accord.messages.Apply;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@ -58,7 +58,6 @@ public class ApplySerializers
}
};
// public static final IVersionedSerializer<Apply> request = new TxnRequestSerializer<Apply>()
public abstract static class ApplySerializer<A extends Apply> extends TxnRequestSerializer<A>
{
@Override
@ -73,11 +72,11 @@ public class ApplySerializers
CommandSerializers.writes.serialize(apply.writes, out, version);
}
protected abstract A deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
protected abstract A deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result);
@Override
public A deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
public A deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
return deserializeApply(txnId, scope, waitForEpoch,
kind.deserialize(in, version),
@ -106,7 +105,7 @@ public class ApplySerializers
public static final IVersionedSerializer<Apply> request = new ApplySerializer<Apply>()
{
@Override
protected Apply deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
protected Apply deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result)
{
return Apply.SerializationSupport.create(txnId, scope, waitForEpoch, kind, keys, executeAt, deps, txn, fullRoute, writes, result);

105
src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java Normal file
View File

@ -0,0 +1,105 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.ProgressLog.BlockedUntil;
import accord.local.SaveStatus;
import accord.messages.Await;
import accord.messages.Await.AsyncAwaitComplete;
import accord.messages.Await.AwaitOk;
import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.vint.VIntCoding;
public class AwaitSerializer
{
public static final IVersionedSerializer<Await> request = new IVersionedSerializer<>()
{
@Override
public void serialize(Await await, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(await.txnId, out, version);
KeySerializers.participants.serialize(await.scope, out, version);
out.writeByte(await.blockedUntil.ordinal());
out.writeUnsignedVInt32(await.callbackId + 1);
Invariants.checkState(await.callbackId >= -1);
}
@Override
public Await deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
Participants<?> scope = KeySerializers.participants.deserialize(in, version);
BlockedUntil blockedUntil = BlockedUntil.forOrdinal(in.readByte());
int callbackId = in.readUnsignedVInt32() - 1;
Invariants.checkState(callbackId >= -1);
return Await.SerializerSupport.create(txnId, scope, blockedUntil, callbackId);
}
@Override
public long serializedSize(Await await, int version)
{
return CommandSerializers.txnId.serializedSize(await.txnId, version)
+ KeySerializers.participants.serializedSize(await.scope, version)
+ TypeSizes.BYTE_SIZE
+ VIntCoding.computeUnsignedVIntSize(await.callbackId + 1);
}
};
public static final IVersionedSerializer<AwaitOk> syncReply = new EnumSerializer<>(AwaitOk.class);
public static final IVersionedSerializer<AsyncAwaitComplete> asyncReply = new IVersionedSerializer<>()
{
@Override
public void serialize(AsyncAwaitComplete ok, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(ok.txnId, out, version);
KeySerializers.route.serialize(ok.route, out, version);
out.writeByte(ok.newStatus.ordinal());
out.writeUnsignedVInt32(ok.callbackId);
}
@Override
public AsyncAwaitComplete deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
Route<?> scope = KeySerializers.route.deserialize(in, version);
SaveStatus newStatus = SaveStatus.forOrdinal(in.readByte());
int callbackId = in.readUnsignedVInt32();
return new AsyncAwaitComplete(txnId, scope, newStatus, callbackId);
}
@Override
public long serializedSize(AsyncAwaitComplete ok, int version)
{
return CommandSerializers.txnId.serializedSize(ok.txnId, version)
+ KeySerializers.route.serializedSize(ok.route, version)
+ TypeSizes.BYTE_SIZE
+ VIntCoding.computeVIntSize(ok.callbackId);
}
};
}

13
src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
View File

@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.RoutingKey;
import accord.local.Status;
import accord.local.SaveStatus;
import accord.messages.BeginInvalidation;
import accord.messages.BeginInvalidation.InvalidateReply;
import accord.primitives.Ballot;
@ -67,7 +67,8 @@ public class BeginInvalidationSerializers
{
CommandSerializers.nullableBallot.serialize(reply.supersededBy, out, version);
CommandSerializers.ballot.serialize(reply.accepted, out, version);
CommandSerializers.status.serialize(reply.status, out, version);
CommandSerializers.saveStatus.serialize(reply.maxStatus, out, version);
CommandSerializers.saveStatus.serialize(reply.maxKnowledgeStatus, out, version);
out.writeBoolean(reply.acceptedFastPath);
KeySerializers.nullableRoute.serialize(reply.route, out, version);
KeySerializers.nullableRoutingKey.serialize(reply.homeKey, out, version);
@ -78,11 +79,12 @@ public class BeginInvalidationSerializers
{
Ballot supersededBy = CommandSerializers.nullableBallot.deserialize(in, version);
Ballot accepted = CommandSerializers.ballot.deserialize(in, version);
Status status = CommandSerializers.status.deserialize(in, version);
SaveStatus maxStatus = CommandSerializers.saveStatus.deserialize(in, version);
SaveStatus maxKnowledgeStatus = CommandSerializers.saveStatus.deserialize(in, version);
boolean acceptedFastPath = in.readBoolean();
Route<?> route = KeySerializers.nullableRoute.deserialize(in, version);
RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
return new InvalidateReply(supersededBy, accepted, status, acceptedFastPath, route, homeKey);
return new InvalidateReply(supersededBy, accepted, maxStatus, maxKnowledgeStatus, acceptedFastPath, route, homeKey);
}
@Override
@ -90,7 +92,8 @@ public class BeginInvalidationSerializers
{
return CommandSerializers.nullableBallot.serializedSize(reply.supersededBy, version)
+ CommandSerializers.ballot.serializedSize(reply.accepted, version)
+ CommandSerializers.status.serializedSize(reply.status, version)
+ CommandSerializers.saveStatus.serializedSize(reply.maxStatus, version)
+ CommandSerializers.saveStatus.serializedSize(reply.maxKnowledgeStatus, version)
+ TypeSizes.BOOL_SIZE
+ KeySerializers.nullableRoute.serializedSize(reply.route, version)
+ KeySerializers.nullableRoutingKey.serializedSize(reply.homeKey, version);

28
src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java → src/java/org/apache/cassandra/service/accord/serializers/CalculateDepsSerializers.java
View File

@ -20,9 +20,9 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.GetDeps;
import accord.messages.GetDeps.GetDepsOk;
import accord.primitives.PartialRoute;
import accord.messages.CalculateDeps;
import accord.messages.CalculateDeps.CalculateDepsOk;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@ -30,49 +30,49 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class GetDepsSerializers
public class CalculateDepsSerializers
{
public static final IVersionedSerializer<GetDeps> request = new TxnRequestSerializer.WithUnsyncedSerializer<GetDeps>()
public static final IVersionedSerializer<CalculateDeps> request = new TxnRequestSerializer.WithUnsyncedSerializer<CalculateDeps>()
{
@Override
public void serializeBody(GetDeps msg, DataOutputPlus out, int version) throws IOException
public void serializeBody(CalculateDeps msg, DataOutputPlus out, int version) throws IOException
{
KeySerializers.seekables.serialize(msg.keys, out, version);
CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
}
@Override
public GetDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
public CalculateDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version);
return GetDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, keys, executeAt);
return CalculateDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, keys, executeAt);
}
@Override
public long serializedBodySize(GetDeps msg, int version)
public long serializedBodySize(CalculateDeps msg, int version)
{
return KeySerializers.seekables.serializedSize(msg.keys, version)
+ CommandSerializers.timestamp.serializedSize(msg.executeAt, version);
}
};
public static final IVersionedSerializer<GetDepsOk> reply = new IVersionedSerializer<GetDepsOk>()
public static final IVersionedSerializer<CalculateDepsOk> reply = new IVersionedSerializer<CalculateDepsOk>()
{
@Override
public void serialize(GetDepsOk reply, DataOutputPlus out, int version) throws IOException
public void serialize(CalculateDepsOk reply, DataOutputPlus out, int version) throws IOException
{
DepsSerializer.partialDeps.serialize(reply.deps, out, version);
}
@Override
public GetDepsOk deserialize(DataInputPlus in, int version) throws IOException
public CalculateDepsOk deserialize(DataInputPlus in, int version) throws IOException
{
return new GetDepsOk(DepsSerializer.partialDeps.deserialize(in, version));
return new CalculateDepsOk(DepsSerializer.partialDeps.deserialize(in, version));
}
@Override
public long serializedSize(GetDepsOk reply, int version)
public long serializedSize(CalculateDepsOk reply, int version)
{
return DepsSerializer.partialDeps.serializedSize(reply.deps, version);
}

33
src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
View File

@ -41,7 +41,7 @@ import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.utils.vint.VIntCoding;
import static accord.local.cfk.CommandsForKey.NO_PENDING_UNMANAGED;
import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
import static accord.primitives.TxnId.NO_TXNIDS;
import static accord.primitives.Txn.Kind.Read;
import static accord.primitives.Txn.Kind.SyncPoint;
import static accord.primitives.Txn.Kind.Write;
@ -117,19 +117,21 @@ public class CommandsForKeySerializer
if (commandCount == 0)
return ByteBuffer.allocate(1);
int[] nodeIds = cachedInts().getInts(Math.min(64, commandCount));
int[] nodeIds = cachedInts().getInts(Math.min(64, Math.max(4, commandCount)));
try
{
// first compute the unique Node Ids and some basic characteristics of the data, such as
// whether we have any missing transactions to encode, any executeAt that are not equal to their TxnId
// and whether there are any non-standard flag bits to encode
boolean hasNonStandardFlags = false;
int nodeIdCount = 0, missingIdCount = 0, executeAtCount = 0, ballotCount = 0;
int nodeIdCount, missingIdCount = 0, executeAtCount = 0, ballotCount = 0;
int bitsPerExecuteAtEpoch = 0, bitsPerExecuteAtFlags = 0, bitsPerExecuteAtHlc = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
{
nodeIds[0] = cfk.redundantBefore().node.id;
nodeIdCount = 1;
for (int i = 0 ; i < commandCount ; ++i)
{
if (nodeIdCount + 2 >= nodeIds.length)
if (nodeIdCount + 3 >= nodeIds.length)
{
nodeIdCount = compact(nodeIds);
if (nodeIdCount > nodeIds.length/2 || nodeIdCount + 2 >= nodeIds.length)
@ -176,10 +178,10 @@ public class CommandsForKeySerializer
int maxHeaderBits = minHeaderBits;
int totalBytes = 0;
long prevEpoch = cfk.get(0).epoch();
long prevHlc = cfk.get(0).hlc();
int prunedBeforeIndex = cfk.prunedBefore().equals(TxnId.NONE) ? -1 : cfk.indexOf(cfk.prunedBefore());
long prevEpoch = cfk.redundantBefore().epoch();
long prevHlc = cfk.redundantBefore().hlc();
int[] bytesHistogram = cachedInts().getInts(12);
Arrays.fill(bytesHistogram, 0);
for (int i = 0 ; i < commandCount ; ++i)
@ -277,10 +279,12 @@ public class CommandsForKeySerializer
cachedInts().forceDiscard(bytesHistogram);
prevEpoch = cfk.get(0).epoch();
prevHlc = cfk.get(0).hlc();
prevEpoch = cfk.redundantBefore().epoch();
prevHlc = cfk.redundantBefore().hlc();
totalBytes += TypeSizes.sizeofUnsignedVInt(prevEpoch);
totalBytes += TypeSizes.sizeofUnsignedVInt(prevHlc);
totalBytes += TypeSizes.sizeofUnsignedVInt(cfk.redundantBefore().flags());
totalBytes += TypeSizes.sizeofUnsignedVInt(Arrays.binarySearch(nodeIds, 0, nodeIdCount, cfk.redundantBefore().node.id));
totalBytes += TypeSizes.sizeofUnsignedVInt(prunedBeforeIndex + 1);
int bitsPerBallotEpoch = 0, bitsPerBallotHlc = 1, bitsPerBallotFlags = 0;
@ -348,8 +352,11 @@ public class CommandsForKeySerializer
VIntCoding.writeUnsignedVInt32(nodeIds[i] - nodeIds[i-1], out);
out.putShort((short)flags);
VIntCoding.writeUnsignedVInt(prevEpoch, out);
VIntCoding.writeUnsignedVInt(prevHlc, out);
VIntCoding.writeUnsignedVInt32(cfk.redundantBefore().flags(), out);
VIntCoding.writeUnsignedVInt32(Arrays.binarySearch(nodeIds, 0, nodeIdCount, cfk.redundantBefore().node.id), out);
VIntCoding.writeUnsignedVInt32(prunedBeforeIndex + 1, out);
int executeAtMask = executeAtCount > 0 ? 1 : 0;
@ -372,7 +379,7 @@ public class CommandsForKeySerializer
bits |= hasExecuteAt << bitIndex;
bitIndex += statusHasInfo & executeAtMask;
long hasMissingIds = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).missing != CommandsForKey.NO_TXNIDS ? 1 : 0;
long hasMissingIds = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).missing != NO_TXNIDS ? 1 : 0;
bits |= hasMissingIds << bitIndex;
bitIndex += statusHasInfo & missingDepsMask;
@ -639,6 +646,12 @@ public class CommandsForKeySerializer
long prevEpoch = VIntCoding.readUnsignedVInt(in);
long prevHlc = VIntCoding.readUnsignedVInt(in);
TxnId redundantBefore;
{
int flags = VIntCoding.readUnsignedVInt32(in);
Node.Id node = nodeIds[VIntCoding.readUnsignedVInt32(in)];
redundantBefore = TxnId.fromValues(prevEpoch, prevHlc, flags, node);
}
int prunedBeforeIndex = VIntCoding.readUnsignedVInt32(in) - 1;
for (int i = 0 ; i < commandCount ; ++i)
@ -879,7 +892,7 @@ public class CommandsForKeySerializer
}
cachedTxnIds().forceDiscard(txnIds, commandCount);
return CommandsForKey.SerializerSupport.create(key, txns, unmanageds, prunedBeforeIndex == -1 ? TxnId.NONE : txns[prunedBeforeIndex]);
return CommandsForKey.SerializerSupport.create(key, txns, unmanageds, redundantBefore, prunedBeforeIndex == -1 ? TxnId.NONE : txns[prunedBeforeIndex]);
}
private static int getHlcBytes(int lookup, int index)

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

@ -26,8 +26,8 @@ import accord.messages.ReadData;
import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@ -68,13 +68,13 @@ public class CommitSerializers
serializeNullable(msg.readData, out, version, read);
}
protected abstract C deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind,
protected abstract C deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Commit.Kind kind,
Ballot ballot, Timestamp executeAt,
Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps,
@Nullable FullRoute<?> fullRoute, @Nullable ReadData read);
@Override
public C deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
public C deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
Commit.Kind kind = CommitSerializers.kind.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
@ -104,7 +104,7 @@ public class CommitSerializers
public static final IVersionedSerializer<Commit> request = new CommitSerializer<Commit, ReadData>(ReadData.class, ReadDataSerializers.readData)
{
@Override
protected Commit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
protected Commit deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, read);
}

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

@ -21,25 +21,11 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.Data;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.impl.AbstractFetchCoordinator.FetchRequest;
import accord.impl.AbstractFetchCoordinator.FetchResponse;
import accord.local.SaveStatus;
import accord.local.Status.Durability;
import accord.local.Status.Known;
import accord.messages.CheckStatus;
import accord.messages.Propagate;
import accord.messages.ReadData.CommitOrReadNack;
import accord.messages.ReadData.ReadReply;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@ -134,92 +120,4 @@ public class FetchSerializers
}
};
public static final IVersionedSerializer<Propagate> propagate = new IVersionedSerializer<Propagate>()
{
@Override
public void serialize(Propagate p, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(p.txnId, out, version);
KeySerializers.route.serialize(p.route, out, version);
CommandSerializers.saveStatus.serialize(p.maxKnowledgeSaveStatus, out, version);
CommandSerializers.saveStatus.serialize(p.maxSaveStatus, out, version);
CommandSerializers.ballot.serialize(p.ballot, out, version);
CommandSerializers.durability.serialize(p.durability, out, version);
KeySerializers.nullableRoutingKey.serialize(p.homeKey, out, version);
KeySerializers.nullableRoutingKey.serialize(p.progressKey, out, version);
CommandSerializers.known.serialize(p.achieved, out, version);
CheckStatusSerializers.foundKnownMap.serialize(p.known, out, version);
out.writeBoolean(p.isShardTruncated);
CommandSerializers.nullablePartialTxn.serialize(p.partialTxn, out, version);
DepsSerializer.nullablePartialDeps.serialize(p.stableDeps, out, version);
out.writeLong(p.toEpoch);
CommandSerializers.nullableTimestamp.serialize(p.committedExecuteAt, out, version);
CommandSerializers.nullableWrites.serialize(p.writes, out, version);
}
@Override
public Propagate deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
Route<?> route = KeySerializers.route.deserialize(in, version);
SaveStatus maxKnowledgeSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
SaveStatus maxSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
Durability durability = CommandSerializers.durability.deserialize(in, version);
RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
RoutingKey progressKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
Known achieved = CommandSerializers.known.deserialize(in, version);
CheckStatus.FoundKnownMap known = CheckStatusSerializers.foundKnownMap.deserialize(in, version);
boolean isTruncated = in.readBoolean();
PartialTxn partialTxn = CommandSerializers.nullablePartialTxn.deserialize(in, version);
PartialDeps committedDeps = DepsSerializer.nullablePartialDeps.deserialize(in, version);
long toEpoch = in.readLong();
Timestamp committedExecuteAt = CommandSerializers.nullableTimestamp.deserialize(in, version);
Writes writes = CommandSerializers.nullableWrites.deserialize(in, version);
Result result = null;
if (achieved.outcome.isOrWasApply())
result = CommandSerializers.APPLIED;
return Propagate.SerializerSupport.create(txnId,
route,
maxKnowledgeSaveStatus,
maxSaveStatus,
ballot,
durability,
homeKey,
progressKey,
achieved,
known,
isTruncated,
partialTxn,
committedDeps,
toEpoch,
committedExecuteAt,
writes,
result);
}
@Override
public long serializedSize(Propagate p, int version)
{
return CommandSerializers.txnId.serializedSize(p.txnId, version)
+ KeySerializers.route.serializedSize(p.route, version)
+ CommandSerializers.saveStatus.serializedSize(p.maxKnowledgeSaveStatus, version)
+ CommandSerializers.saveStatus.serializedSize(p.maxSaveStatus, version)
+ CommandSerializers.ballot.serializedSize(p.ballot, version)
+ CommandSerializers.durability.serializedSize(p.durability, version)
+ KeySerializers.nullableRoutingKey.serializedSize(p.homeKey, version)
+ KeySerializers.nullableRoutingKey.serializedSize(p.progressKey, version)
+ CommandSerializers.known.serializedSize(p.achieved, version)
+ CheckStatusSerializers.foundKnownMap.serializedSize(p.known, version)
+ TypeSizes.BOOL_SIZE
+ CommandSerializers.nullablePartialTxn.serializedSize(p.partialTxn, version)
+ DepsSerializer.nullablePartialDeps.serializedSize(p.stableDeps, version)
+ TypeSizes.sizeof(p.toEpoch)
+ CommandSerializers.nullableTimestamp.serializedSize(p.committedExecuteAt, version)
+ CommandSerializers.nullableWrites.serializedSize(p.writes, version)
;
}
};
}

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

@ -23,7 +23,7 @@ import java.io.IOException;
import accord.messages.GetEphemeralReadDeps;
import accord.messages.GetEphemeralReadDeps.GetEphemeralReadDepsOk;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
@ -43,7 +43,7 @@ public class GetEphmrlReadDepsSerializers
}
@Override
public GetEphemeralReadDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
public GetEphemeralReadDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
long executionEpoch = in.readUnsignedVInt();

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

@ -22,7 +22,7 @@ import java.io.IOException;
import accord.messages.GetMaxConflict;
import accord.messages.GetMaxConflict.GetMaxConflictOk;
import accord.primitives.PartialRoute;
import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@ -43,7 +43,7 @@ public class GetMaxConflictSerializers
}
@Override
public GetMaxConflict deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
public GetMaxConflict deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
long executionEpoch = in.readUnsignedVInt();

10
src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java
View File

@ -22,7 +22,7 @@ import java.io.IOException;
import accord.local.Status;
import accord.messages.InformDurable;
import accord.primitives.PartialRoute;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.io.IVersionedSerializer;
@ -36,14 +36,14 @@ public class InformDurableSerializers
@Override
public void serializeBody(InformDurable msg, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
CommandSerializers.nullableTimestamp.serialize(msg.executeAt, out, version);
CommandSerializers.durability.serialize(msg.durability, out, version);
}
@Override
public InformDurable deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute scope, long waitForEpoch) throws IOException
public InformDurable deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version);
Timestamp executeAt = CommandSerializers.nullableTimestamp.deserialize(in, version);
Status.Durability durability = CommandSerializers.durability.deserialize(in, version);
return InformDurable.SerializationSupport.create(txnId, scope, waitForEpoch, executeAt, durability);
}
@ -51,7 +51,7 @@ public class InformDurableSerializers
@Override
public long serializedBodySize(InformDurable msg, int version)
{
return CommandSerializers.timestamp.serializedSize(msg.executeAt, version)
return CommandSerializers.nullableTimestamp.serializedSize(msg.executeAt, version)
+ CommandSerializers.durability.serializedSize(msg.durability, version);
}
};

60
src/java/org/apache/cassandra/service/accord/serializers/InformHomeDurableSerializers.java
View File

@ -1,60 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.InformHomeDurable;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class InformHomeDurableSerializers
{
public static final IVersionedSerializer<InformHomeDurable> request = new IVersionedSerializer<>()
{
@Override
public void serialize(InformHomeDurable inform, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(inform.txnId, out, version);
KeySerializers.route.serialize(inform.route, out, version);
CommandSerializers.timestamp.serialize(inform.executeAt, out, version);
CommandSerializers.durability.serialize(inform.durability, out, version);
}
@Override
public InformHomeDurable deserialize(DataInputPlus in, int version) throws IOException
{
return new InformHomeDurable(CommandSerializers.txnId.deserialize(in, version),
KeySerializers.route.deserialize(in, version),
CommandSerializers.timestamp.deserialize(in, version),
CommandSerializers.durability.deserialize(in, version));
}
@Override
public long serializedSize(InformHomeDurable inform, int version)
{
return CommandSerializers.txnId.serializedSize(inform.txnId, version)
+ KeySerializers.route.serializedSize(inform.route, version)
+ CommandSerializers.timestamp.serializedSize(inform.executeAt, version)
+ CommandSerializers.durability.serializedSize(inform.durability, version);
}
};
}

53
src/java/org/apache/cassandra/service/accord/serializers/InformOfTxnIdSerializers.java
View File

@ -1,53 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.InformOfTxnId;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class InformOfTxnIdSerializers
{
public static final IVersionedSerializer<InformOfTxnId> request = new IVersionedSerializer<InformOfTxnId>()
{
@Override
public void serialize(InformOfTxnId inform, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(inform.txnId, out, version);
KeySerializers.route.serialize(inform.someRoute, out, version);
}
@Override
public InformOfTxnId deserialize(DataInputPlus in, int version) throws IOException
{
return new InformOfTxnId(CommandSerializers.txnId.deserialize(in, version),
KeySerializers.route.deserialize(in, version));
}
@Override
public long serializedSize(InformOfTxnId inform, int version)
{
return CommandSerializers.txnId.serializedSize(inform.txnId, version)
+ KeySerializers.route.serializedSize(inform.someRoute, version);
}
};
}

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

@ -1,113 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.local.Command;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class ListenerSerializers
{
public enum Kind
{
COMMAND;
private static Kind of(Command.DurableAndIdempotentListener listener)
{
if (listener instanceof Command.ProxyListener)
return COMMAND;
throw new IllegalArgumentException("Unsupported listener type: " + listener.getClass().getName());
}
}
private static final IVersionedSerializer<Command.ProxyListener> commandListener = new IVersionedSerializer<Command.ProxyListener>()
{
@Override
public void serialize(Command.ProxyListener listener, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(listener.txnId(), out, version);
}
@Override
public Command.ProxyListener deserialize(DataInputPlus in, int version) throws IOException
{
return new Command.ProxyListener(CommandSerializers.txnId.deserialize(in, version));
}
@Override
public long serializedSize(Command.ProxyListener listener, int version)
{
return CommandSerializers.txnId.serializedSize(listener.txnId(), version);
}
};
public static final IVersionedSerializer<Command.DurableAndIdempotentListener> listener = new IVersionedSerializer<Command.DurableAndIdempotentListener>()
{
@Override
public void serialize(Command.DurableAndIdempotentListener listener, DataOutputPlus out, int version) throws IOException
{
Kind kind = Kind.of(listener);
out.write(kind.ordinal());
switch (kind)
{
case COMMAND:
commandListener.serialize((Command.ProxyListener) listener, out, version);
break;
default:
throw new IllegalArgumentException();
}
}
@Override
public Command.DurableAndIdempotentListener deserialize(DataInputPlus in, int version) throws IOException
{
Kind kind = Kind.values()[in.readByte()];
switch (kind)
{
case COMMAND:
return commandListener.deserialize(in, version);
default:
throw new IllegalArgumentException();
}
}
@Override
public long serializedSize(Command.DurableAndIdempotentListener listener, int version)
{
Kind kind = Kind.of(listener);
long size = TypeSizes.BYTE_SIZE;
switch (kind)
{
case COMMAND:
size += commandListener.serializedSize((Command.ProxyListener) listener, version);
break;
default:
throw new IllegalArgumentException();
}
return size;
}
};
}

10
src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
View File

@ -25,8 +25,8 @@ import accord.messages.PreAccept;
import accord.messages.PreAccept.PreAcceptOk;
import accord.messages.PreAccept.PreAcceptReply;
import accord.primitives.FullRoute;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@ -49,16 +49,16 @@ public class PreacceptSerializers
{
CommandSerializers.partialTxn.serialize(msg.partialTxn, out, version);
serializeNullable(msg.route, out, version, KeySerializers.fullRoute);
out.writeUnsignedVInt(msg.maxEpoch - msg.minUnsyncedEpoch);
out.writeUnsignedVInt(msg.maxEpoch - msg.minEpoch);
}
@Override
public PreAccept deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
public PreAccept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version);
@Nullable FullRoute<?> fullRoute = deserializeNullable(in, version, KeySerializers.fullRoute);
long maxEpoch = in.readUnsignedVInt() + minEpoch;
return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey,
return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch,
maxEpoch, partialTxn, fullRoute);
}
@ -67,7 +67,7 @@ public class PreacceptSerializers
{
return CommandSerializers.partialTxn.serializedSize(msg.partialTxn, version)
+ serializedNullableSize(msg.route, version, KeySerializers.fullRoute)
+ TypeSizes.sizeofUnsignedVInt(msg.maxEpoch - msg.minUnsyncedEpoch);
+ TypeSizes.sizeofUnsignedVInt(msg.maxEpoch - msg.minEpoch);
}
};

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

@ -33,8 +33,8 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.LatestDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
@ -60,7 +60,7 @@ public class RecoverySerializers
}
@Override
public BeginRecovery deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
public BeginRecovery deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);

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

@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.TxnRequest;
import accord.primitives.PartialRoute;
import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@ -33,7 +33,7 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
void serializeHeader(T msg, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(msg.txnId, out, version);
KeySerializers.partialRoute.serialize(msg.scope, out, version);
KeySerializers.route.serialize(msg.scope, out, version);
out.writeUnsignedVInt(msg.waitForEpoch);
}
@ -46,13 +46,13 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
serializeBody(msg, out, version);
}
public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException;
public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException;
@Override
public final T deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
PartialRoute<?> scope = KeySerializers.partialRoute.deserialize(in, version);
Route<?> scope = KeySerializers.route.deserialize(in, version);
// TODO: there should be a base epoch
long waitForEpoch = in.readUnsignedVInt();
return deserializeBody(in, version, txnId, scope, waitForEpoch);
@ -61,7 +61,7 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
long serializedHeaderSize(T msg, int version)
{
return CommandSerializers.txnId.serializedSize(msg.txnId, version)
+ KeySerializers.partialRoute.serializedSize(msg.scope(), version) +
+ KeySerializers.route.serializedSize(msg.scope(), version) +
TypeSizes.sizeofUnsignedVInt(msg.waitForEpoch);
}
@ -79,26 +79,23 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
void serializeHeader(T msg, DataOutputPlus out, int version) throws IOException
{
super.serializeHeader(msg, out, version);
out.writeUnsignedVInt(msg.minUnsyncedEpoch);
out.writeBoolean(msg.doNotComputeProgressKey);
out.writeUnsignedVInt(msg.minEpoch);
}
public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException;
public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException;
@Override
public final T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
public final T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
long minEpoch = in.readUnsignedVInt();
boolean doNotComputeProgressKey = in.readBoolean();
return deserializeBody(in, version, txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey);
return deserializeBody(in, version, txnId, scope, waitForEpoch, minEpoch);
}
@Override
long serializedHeaderSize(T msg, int version)
{
long size = super.serializedHeaderSize(msg, version);
size += TypeSizes.sizeofUnsignedVInt(msg.minUnsyncedEpoch);
size += TypeSizes.BOOL_SIZE;
size += TypeSizes.sizeofUnsignedVInt(msg.minEpoch);
return size;
}
}

77
src/java/org/apache/cassandra/service/accord/serializers/WaitOnCommitSerializer.java
View File

@ -1,77 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.WaitOnCommit;
import accord.messages.WaitOnCommit.WaitOnCommitOk;
import accord.primitives.Participants;
import accord.primitives.TxnId;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
public class WaitOnCommitSerializer
{
public static final IVersionedSerializer<WaitOnCommit> request = new IVersionedSerializer<WaitOnCommit>()
{
@Override
public void serialize(WaitOnCommit wait, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(wait.txnId, out, version);
KeySerializers.participants.serialize(wait.scope, out, version);
}
@Override
public WaitOnCommit deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
Participants<?> scope = KeySerializers.participants.deserialize(in, version);
return WaitOnCommit.SerializerSupport.create(txnId, scope);
}
@Override
public long serializedSize(WaitOnCommit wait, int version)
{
return CommandSerializers.txnId.serializedSize(wait.txnId, version)
+ KeySerializers.participants.serializedSize(wait.scope, version);
}
};
public static final IVersionedSerializer<WaitOnCommitOk> reply = new IVersionedSerializer<WaitOnCommitOk>()
{
@Override
public void serialize(WaitOnCommitOk ok, DataOutputPlus out, int version) throws IOException
{
}
@Override
public WaitOnCommitOk deserialize(DataInputPlus in, int version) throws IOException
{
return WaitOnCommitOk.INSTANCE;
}
@Override
public long serializedSize(WaitOnCommitOk ok, int version)
{
return 0;
}
};
}

5
src/java/org/apache/cassandra/triggers/TriggerExecutor.java
View File

@ -19,11 +19,6 @@
package org.apache.cassandra.triggers;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.TimeUnit;

1
test/conf/cassandra.yaml
View File

@ -119,4 +119,3 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
shard_count: 4
progress_log_schedule_delay: 1s

2
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -447,7 +447,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
byte[] bytes = out.toByteArray();
if (messageOut.serializedSize(toVersion) != bytes.length)
throw new AssertionError(String.format("Message serializedSize(%s) does not match what was written with serialize(out, %s) for verb %s and serializer %s; " +
"expected %s, actual %s ", toVersion, toVersion, messageOut.verb(), Message.serializer.getClass(),
"expected %s, actual %s ", toVersion, toVersion, messageOut.verb(), messageOut.verb().serializer().getClass(),
messageOut.serializedSize(toVersion), bytes.length));
return new MessageImpl(messageOut.verb().id, bytes, messageOut.id(), toVersion, messageOut.expiresAtNanos(), fromCassandraInetAddressAndPort(from));
}

2
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
View File

@ -100,7 +100,7 @@ public class InstanceConfig implements IInstanceConfig
.set("accord.enabled", accord.enabled)
.set("accord.journal_directory", accord.journal_directory)
.set("accord.shard_count", accord.shard_count.toString())
.set("accord.progress_log_schedule_delay", accord.progress_log_schedule_delay.toString())
.set("accord.recover_delay", accord.recover_delay.toString())
.set("partitioner", "org.apache.cassandra.dht.Murmur3Partitioner")
.set("start_native_transport", true)
.set("concurrent_writes", 2)

4
test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java
View File

@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import accord.impl.progresslog.DefaultProgressLogs;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
@ -31,7 +32,6 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.impl.SimpleProgressLog;
import com.datastax.driver.core.Session;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
@ -177,7 +177,7 @@ public class QueriesTableTest extends TestBaseImpl
// Disable recovery to make sure only one local read occurs:
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = true);
instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(true));
String update = "BEGIN TRANSACTION\n" +
" LET row1 = (SELECT * FROM " + KEYSPACE + ".accord_tbl WHERE k = 0);\n" +

6
test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
View File

@ -36,7 +36,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.impl.SimpleProgressLog;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@ -156,7 +156,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
public void tearDown()
{
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = false);
instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(false));
SHARED_CLUSTER.filters().reset();
}
@ -292,7 +292,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
// heal partition and wait for node 1 to see node 3 again
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> {
SimpleProgressLog.PAUSE_FOR_TEST = true;
DefaultProgressLogs.unsafePauseForTesting(true);
Assert.assertTrue(agent().executedBarriers().isEmpty());
});
SHARED_CLUSTER.filters().reset();

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

@ -26,7 +26,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.impl.SimpleProgressLog;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.messages.Commit;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IMessageFilters;
@ -131,6 +131,6 @@ public class AccordIntegrationTest extends AccordTestBase
private void pauseSimpleProgressLog()
{
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = true);
instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(true));
}
}

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

@ -42,8 +42,8 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.messages.PreAccept;
import accord.primitives.PartialKeyRoute;
import accord.primitives.PartialRoute;
import accord.primitives.Routable.Domain;
import accord.primitives.Route;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.Util;
import org.apache.cassandra.batchlog.BatchlogManager;
@ -458,7 +458,7 @@ public abstract class AccordMigrationRaceTestBase extends AccordTestBase
{
boolean drop = cluster.get(to).callsOnInstance(() -> {
PreAccept preAccept = (PreAccept)Instance.deserializeMessage(message).payload;
PartialRoute<?> route = preAccept.scope;
Route<?> route = preAccept.scope;
if (route.domain() == Domain.Key)
for (RoutingKey key : (PartialKeyRoute)route)
{

130
test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java Normal file
View File

@ -0,0 +1,130 @@
/*
* 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.distributed.test.accord;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.consensus.TransactionalMode;
public class AccordProgressLogTest extends TestBaseImpl
{
private static final Logger logger = LoggerFactory.getLogger(AccordProgressLogTest.class);
@Test
public void testRecoveryTimeWindow() throws Throwable
{
try (Cluster cluster = init(Cluster.build(3)
.withoutVNodes()
.withConfig(c -> c.with(Feature.NETWORK).set("accord.enabled", "true"))
.start()))
{
cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}");
cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key (k, c)) WITH " + TransactionalMode.full.asCqlParam());
String query = "BEGIN TRANSACTION\n" +
" SELECT * FROM ks.tbl WHERE k=0 AND c=0;\n" +
"COMMIT TRANSACTION";
IMessageFilters.Filter dropCommit = cluster.filters().outbound().from(1).verbs(Verb.ACCORD_COMMIT_REQ.id).drop();
AtomicLong recoveryStartedAt = new AtomicLong();
Semaphore waitForRecovery = new Semaphore(0);
IMessageFilters.Filter recovery = cluster.filters().outbound().messagesMatching((from, to, message) -> {
if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_RSP.id)
{
recoveryStartedAt.compareAndSet(0, System.nanoTime());
waitForRecovery.release();
}
return false;
}).drop();
long coordinationStartedAt = System.nanoTime();
boolean failed = false;
try
{
cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.ANY);
}
catch (Throwable e)
{
failed = true;
}
Assert.assertTrue(failed);
waitForRecovery.acquire();
long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(recoveryStartedAt.get() - coordinationStartedAt);
Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 1000);
Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis <= 3000);
}
}
@Test
public void testFetchTimeWindow() throws Throwable
{
try (Cluster cluster = init(Cluster.build(3)
.withoutVNodes()
.withConfig(c -> c.with(Feature.NETWORK).set("accord.enabled", "true"))
.start()))
{
cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}");
cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key (k, c)) WITH " + TransactionalMode.full.asCqlParam());
String query = "BEGIN TRANSACTION\n" +
" SELECT * FROM ks.tbl WHERE k=0 AND c=0;\n" +
"COMMIT TRANSACTION";
IMessageFilters.Filter dropApply = cluster.filters().outbound().from(1).verbs(Verb.ACCORD_APPLY_REQ.id).drop();
AtomicLong fetchStartedAt = new AtomicLong();
Semaphore waitForFetch = new Semaphore(0);
IMessageFilters.Filter fetch = cluster.filters().outbound().messagesMatching((from, to, message) -> {
if (message.verb() == Verb.ACCORD_AWAIT_REQ.id)
{
fetchStartedAt.compareAndSet(0, System.nanoTime());
waitForFetch.release();
}
return false;
}).drop();
long coordinationStartedAt = System.nanoTime();
try
{
cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.ANY);
}
catch (Throwable e)
{
}
waitForFetch.acquire();
logger.info("Coordinated at {}", coordinationStartedAt);
logger.info("Awaited at {}", fetchStartedAt.get());
long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(fetchStartedAt.get() - coordinationStartedAt);
Assert.assertTrue("Fetch started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 100);
Assert.assertTrue("Fetch started in " + timeDeltaMillis + "ms", timeDeltaMillis <= 2000);
}
}
}

8
test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
View File

@ -44,11 +44,11 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.coordinate.Invalidated;
import accord.impl.SimpleProgressLog;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.messages.PreAccept;
import accord.primitives.PartialKeyRoute;
import accord.primitives.PartialRoute;
import accord.primitives.Routable.Domain;
import accord.primitives.Route;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
@ -142,7 +142,7 @@ public abstract class AccordTestBase extends TestBaseImpl
public void tearDown() throws Exception
{
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = false);
instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(false));
}
protected static void assertRowSerial(Cluster cluster, String query, int k, int c, int v, int s)
@ -558,7 +558,7 @@ public abstract class AccordTestBase extends TestBaseImpl
{
boolean drop = cluster.get(to).callsOnInstance(() -> {
PreAccept preAccept = (PreAccept)Instance.deserializeMessage(message).payload;
PartialRoute<?> route = preAccept.scope;
Route<?> route = preAccept.scope;
if (route.domain() == Domain.Key)
for (RoutingKey key : (PartialKeyRoute)route)
{

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

@ -32,8 +32,8 @@ import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
import accord.primitives.Routable;
import accord.local.CommandStores;
import accord.primitives.Route;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.shared.WithProperties;
import org.apache.cassandra.service.accord.*;
@ -58,7 +58,6 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Seekable;
@ -95,6 +94,7 @@ import org.assertj.core.api.Assertions;
import static accord.impl.TimestampsForKey.NO_LAST_EXECUTED_HLC;
import static accord.local.KeyHistory.COMMANDS;
import static accord.local.PreLoadContext.contextFor;
import static accord.primitives.Routable.Domain.Range;
import static accord.utils.async.AsyncChains.getUninterruptibly;
import static org.apache.cassandra.Util.spinAssertEquals;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
@ -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() + 2, NODE, Kind.Read, Routable.Domain.Range);
private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 2, NODE, Kind.Read, 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 };
@ -383,8 +383,8 @@ 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);
txnId = txnId.as(Kind.Read, Range);
return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, LT_TXN_ID.as(Range));
}
enum DurableBeforeType
@ -468,25 +468,22 @@ public class CompactionAccordIteratorsTest
Txn txn = txnId.kind().isWrite() ? writeTxn : readTxn;
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true);
PartialRoute<?> partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
Route<?> partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToKeyspace(commandStore));
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, txnId, partialDeps, appendDiffToKeyspace(commandStore));
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), txnId, partialDeps, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, txnId, partialDeps, appendDiffToKeyspace(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
Pair<Writes, Result> result = AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
CheckedCommands.apply(safe, txnId, route, null, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
}).beginAsResult());
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
safe.get(txnId, txnId, route).addListener(new Command.ProxyListener(RANGE_TXN_ID)); // add a junk listener just to test it in compaction
CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied

4
test/unit/org/apache/cassandra/dht/AccordSplitterTest.java
View File

@ -98,7 +98,7 @@ public class AccordSplitterTest
Assertions.assertThat(ranges).describedAs("num splits not as expected for partitioner %s", partitioner).hasSizeBetween(numSplits, numSplits + 1);
Ranges split = Ranges.of(ranges.toArray(new Range[0])).mergeTouching();
Ranges missing = Ranges.of(range).subtract(split);
Ranges missing = Ranges.of(range).without(split);
Assertions.assertThat(missing).isEmpty();
testEventSplit(partitioner, range, rs, numSplits);
@ -115,7 +115,7 @@ public class AccordSplitterTest
Assertions.assertThat(ranges).describedAs("num splits not as expected for partitioner %s", partitioner).hasSize(numSplits);
Ranges split = ranges.stream().reduce(Ranges.EMPTY, Ranges::with).mergeTouching();
Ranges missing = topLevel.subtract(split);
Ranges missing = topLevel.without(split);
Assertions.assertThat(missing).isEmpty();
}

70
test/unit/org/apache/cassandra/service/accord/AccordAgentTest.java Normal file
View File

@ -0,0 +1,70 @@
/*
* 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.Arrays;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import accord.local.Node;
import accord.utils.RandomTestRunner;
import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.service.accord.api.AccordAgent;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.junit.Assert.assertTrue;
public class AccordAgentTest
{
@Test
public void testNonClashingStartTimes()
{
RandomTestRunner.test().check(rnd -> {
SortedArrayList<Node.Id> nodes; {
Node.Id[] ids = new Node.Id[rnd.nextInt(4, 16)];
for (int i = 0 ; i < ids.length ; ++i)
ids[i] = new Node.Id(i);
nodes = new SortedArrayList<>(ids);
}
long[] startTimes = new long[nodes.size()];
long oneSecond = SECONDS.toMicros(1);
long targetDelta = oneSecond / nodes.size();
for (int i = 0 ; i < 10000 ; ++i)
{
long startTime = rnd.nextLong(1, TimeUnit.DAYS.toMicros(100L));
for (int j = 0 ; j < startTimes.length ; ++j)
{
long nonClashingStartTime = AccordAgent.nonClashingStartTime(startTime, nodes, nodes.get(j), oneSecond, rnd);
assertTrue(nonClashingStartTime >= startTime);
startTimes[j] = nonClashingStartTime;
}
Arrays.sort(startTimes);
for (int j = 1 ; j < startTimes.length ; ++j)
{
long actualDelta = startTimes[j] - startTimes[j - 1];
assertTrue(Math.abs(targetDelta - actualDelta) <= startTimes.length);
}
}
});
}
}

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

@ -135,7 +135,6 @@ public class AccordCommandStoreTest
SimpleBitSet waitingOnApply = new SimpleBitSet(3);
waitingOnApply.set(1);
Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps, dependencies.directKeyDeps, new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt);
Command expected = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted,

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

@ -41,7 +41,6 @@ import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Timestamp;
@ -100,9 +99,9 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
PartialRoute<?> route = fullRoute.slice(fullRange(txn));
Route<?> route = fullRoute.slice(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, false, 1, partialTxn, fullRoute);
PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, fullRoute);
// Check preaccept
getUninterruptibly(commandStore.execute(preAccept, safeStore -> {
@ -141,7 +140,7 @@ public class AccordCommandTest
builder.add(key, txnId2);
deps = builder.build();
}
Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, false, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
getUninterruptibly(commandStore.execute(accept, safeStore -> {
Command before = safeStore.ifInitialised(txnId).current();
@ -192,9 +191,9 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
PartialRoute<?> route = fullRoute.slice(fullRange(txn));
Route<?> route = fullRoute.slice(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, false, 1, partialTxn, fullRoute);
PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, fullRoute);
getUninterruptibly(commandStore.execute(preAccept1, safeStore -> {
persistDiff(commandStore, safeStore, txnId1, route, () -> {
@ -204,7 +203,7 @@ public class AccordCommandTest
// second preaccept should identify txnId1 as a dependency
TxnId txnId2 = txnId(1, clock.incrementAndGet(), 1);
PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, false, 1, partialTxn, fullRoute);
PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, 1, partialTxn, fullRoute);
getUninterruptibly(commandStore.execute(preAccept2, safeStore -> {
persistDiff(commandStore, safeStore, txnId2, route, () -> {
PreAccept.PreAcceptReply reply = preAccept2.apply(safeStore);

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

@ -32,12 +32,9 @@ import accord.api.Agent;
import accord.impl.AbstractFetchCoordinator;
import accord.impl.IntKey;
import accord.local.Node;
import accord.messages.InformOfTxnId;
import accord.messages.MessageType;
import accord.messages.ReadTxnData;
import accord.messages.Reply;
import accord.messages.Request;
import accord.messages.SimpleReply;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
@ -72,16 +69,6 @@ public class AccordMessageSinkTest
ClusterMetadataService.initializeForClients();
}
@Test
public void informOfTxn()
{
// There was an issue where the reply was the wrong verb
// see CASSANDRA-18375
InformOfTxnId request = Mockito.mock(InformOfTxnId.class);
Mockito.when(request.type()).thenReturn(MessageType.INFORM_OF_TXN_REQ);
checkRequestReplies(request, SimpleReply.Ok);
}
@Test
public void bootstrapRead()
{

6
test/unit/org/apache/cassandra/service/accord/AccordServiceTest.java
View File

@ -74,7 +74,7 @@ public class AccordServiceTest
throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);
case 4:
attempts++;
throw new Exhausted(null, null);
throw new Exhausted(null, null, null);
case 5:
attempts++;
throw AccordService.newBarrierExhausted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);
@ -124,7 +124,7 @@ public class AccordServiceTest
timeoutFailures.add(() -> {throw AccordService.newBarrierTimeout(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
timeoutFailures.add(() -> {throw new Preempted(null, null);});
timeoutFailures.add(() -> {throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
timeoutFailures.add(() -> {throw new Exhausted(null, null);});
timeoutFailures.add(() -> {throw new Exhausted(null, null, null);});
Collections.shuffle(timeoutFailures, rs.asJdkRandom());
Iterator<Runnable> it = timeoutFailures.iterator();
Supplier<Seekables> failing = () -> {
@ -162,7 +162,7 @@ public class AccordServiceTest
failures.add(() -> {throw AccordService.newBarrierTimeout(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
failures.add(() -> {throw new Preempted(null, null);});
failures.add(() -> {throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
failures.add(() -> {throw new Exhausted(null, null);});
failures.add(() -> {throw new Exhausted(null, null, null);});
boolean isError = rs.nextBoolean();
failures.add(new Unexpected(isError));
Collections.shuffle(failures, rs.asJdkRandom());

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

@ -30,10 +30,13 @@ import java.util.function.LongSupplier;
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.annotation.Nullable;
import com.google.common.collect.Sets;
import accord.api.LocalListeners;
import accord.api.ProgressLog.NoOpProgressLog;
import accord.api.RemoteListeners;
import accord.impl.DefaultLocalListeners;
import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -41,7 +44,6 @@ import org.apache.cassandra.io.util.File;
import org.junit.Assert;
import accord.api.Data;
import accord.api.ProgressLog;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.impl.InMemoryCommandStore;
@ -56,17 +58,14 @@ import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.SaveStatus;
import accord.local.SaveStatus.LocalExecution;
import accord.primitives.Ballot;
import accord.primitives.FullKeyRoute;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Route;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
@ -202,21 +201,6 @@ public class AccordTestUtils
Assert.assertEquals(val, safeState.current());
}
public static final ProgressLog NOOP_PROGRESS_LOG = new ProgressLog()
{
@Override public void unwitnessed(TxnId txnId, ProgressShard progressShard) {}
@Override public void preaccepted(Command command, ProgressShard progressShard) {}
@Override public void accepted(Command command, ProgressShard progressShard) {}
@Override public void precommitted(Command command) {}
@Override public void stable(Command command, ProgressShard progressShard) {}
@Override public void readyToExecute(Command command) {}
@Override public void executed(Command command, ProgressShard progressShard) {}
@Override public void clear(TxnId txnId) {}
@Override public void durable(Command command) {}
@Override public void waiting(SafeCommand blockedBy, LocalExecution blockedUntil, Route<?> blockedOnRoute, Participants<?> blockedOnParticipants) {}
@Override public void waiting(TxnId blockedBy, LocalExecution blockedUntil, @Nullable Route<?> blockedOnRoute, @Nullable Participants<?> blockedOnParticipants) {}
};
public static TxnId txnId(long epoch, long hlc, int node)
{
return txnId(epoch, hlc, node, Txn.Kind.Write);
@ -390,11 +374,8 @@ public class AccordTestUtils
};
SingleEpochRanges holder = new SingleEpochRanges(Ranges.of(range));
InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0,
time,
new AccordAgent(),
null,
cs -> null, holder);
InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0, time, new AccordAgent(),
null, null, cs -> null, holder);
holder.set(result);
return result;
}
@ -425,7 +406,12 @@ public class AccordTestUtils
time,
new AccordAgent(),
null,
cs -> NOOP_PROGRESS_LOG,
cs -> new NoOpProgressLog(),
cs -> new DefaultLocalListeners(new RemoteListeners.NoOpRemoteListeners(), new DefaultLocalListeners.NotifySink()
{
@Override public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener) {}
@Override public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, LocalListeners.ComplexListener listener) { return false; }
}),
holder,
journal,
loadExecutor,

6
test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
View File

@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.ConfigurationService.EpochReady;
import accord.api.Scheduler;
import accord.config.LocalConfig;
import accord.api.LocalConfig;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Node;
@ -352,7 +352,7 @@ public class EpochSyncTest
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Ranges ranges = tm.globalForEpoch(epoch).ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
Assertions.assertThat(actual).describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.subtract(actual)).isEqualTo(ranges);
Assertions.assertThat(actual).describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual)).isEqualTo(ranges);
}
else
{
@ -369,7 +369,7 @@ public class EpochSyncTest
if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.syncComplete(epoch - 1).mergeTouching()))
continue;
Assertions.assertThat(actual)
.describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", id, epoch, ranges.subtract(actual), topology.nodes(),
.describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", id, epoch, ranges.without(actual), topology.nodes(),
LongStream.range(inst.epoch.getEpoch(), epoch + 1).mapToObj(e -> e + " -> " + conf.getEpochSnapshot(e).syncStatus + "(synced=" + globalSynced(e) + "): " + tm.syncComplete(e)).collect(Collectors.joining("\n")))
.isEqualTo(ranges);
}

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

@ -26,7 +26,6 @@ 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;
@ -97,8 +96,7 @@ public class MockJournal implements IJournal
ifNotEqual(before, after, Command::additionalKeysOrRanges, false),
new NewValue<>((k, deps) -> waitingOn),
ifNotEqual(before, after, Command::writes, false),
ifNotEqual(before, after, Command::durableListeners, true));
ifNotEqual(before, after, Command::writes, false));
}
static Command reconstructFromDiff(List<LoadedDiff> diffs)
@ -131,7 +129,6 @@ public class MockJournal implements IJournal
SavedCommand.WaitingOnProvider waitingOnProvider = null;
Writes writes = null;
Listeners.Immutable listeners = null;
for (LoadedDiff diff : diffs)
{
@ -162,8 +159,6 @@ public class MockJournal implements IJournal
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);
@ -180,8 +175,6 @@ public class MockJournal implements IJournal
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
if (listeners != null && !listeners.isEmpty())
attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (waitingOnProvider != null)
@ -292,7 +285,6 @@ public class MockJournal implements IJournal
public final NewValue<Seekables<?, ?>> additionalKeysOrRanges;
public final NewValue<Writes> writes;
public final NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners;
public final NewValue<WaitingOnProvider> waitingOn;
public LoadedDiff(TxnId txnId,
@ -309,8 +301,7 @@ public class MockJournal implements IJournal
NewValue<Seekables<?, ?>> additionalKeysOrRanges,
NewValue<SavedCommand.WaitingOnProvider> waitingOn,
NewValue<Writes> writes,
NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners)
NewValue<Writes> writes)
{
this.txnId = txnId;
this.executeAt = executeAt;
@ -326,7 +317,6 @@ public class MockJournal implements IJournal
this.additionalKeysOrRanges = additionalKeysOrRanges;
this.writes = writes;
this.listeners = listeners;
this.waitingOn = waitingOn;
}

12
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -28,6 +28,10 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.ToLongFunction;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.api.RemoteListeners;
import accord.impl.DefaultLocalListeners;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Command;
@ -36,6 +40,7 @@ import accord.local.CommandStores;
import accord.local.Node;
import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.messages.BeginRecovery;
import accord.messages.PreAccept;
@ -164,7 +169,12 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
},
null,
ignore -> AccordTestUtils.NOOP_PROGRESS_LOG,
ignore -> new ProgressLog.NoOpProgressLog(),
cs -> new DefaultLocalListeners(new RemoteListeners.NoOpRemoteListeners(), new DefaultLocalListeners.NotifySink()
{
@Override public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener) {}
@Override public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, LocalListeners.ComplexListener listener) { return false; }
}),
updateHolder,
journal,
new AccordStateCacheMetrics("test"));

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

@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import accord.primitives.Route;
import accord.utils.DefaultRandom;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
@ -52,7 +53,6 @@ import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
@ -212,8 +212,8 @@ public class AsyncOperationTest
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys(), COMMANDS), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
}).beginAsResult());
@ -253,16 +253,16 @@ public class AsyncOperationTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
PartialRoute<?> partialRoute = route.slice(ranges);
Route<?> partialRoute = route.slice(ranges);
PartialDeps deps = PartialDeps.builder(ranges).build();
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys(), COMMANDS), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToLog(commandStore));
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
}).beginAsResult());

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

@ -46,7 +46,6 @@ import accord.local.cfk.CommandsForKey.TxnInfo;
import accord.local.cfk.CommandsForKey.Unmanaged;
import accord.local.CommonAttributes;
import accord.local.CommonAttributes.Mutable;
import accord.local.Listeners;
import accord.local.Node;
import accord.local.SaveStatus;
import accord.local.Status;
@ -151,8 +150,9 @@ public class CommandsForKeySerializerTest
return Command.SerializerSupport.notDefined(attributes(), Ballot.ZERO);
case PreAccepted:
return Command.SerializerSupport.preaccepted(attributes(), executeAt, Ballot.ZERO);
case Accepted:
case AcceptedInvalidate:
return Command.SerializerSupport.acceptedInvalidateWithoutDefinition(attributes(), ballot, ballot);
case Accepted:
case AcceptedWithDefinition:
case AcceptedInvalidateWithDefinition:
case PreCommittedWithDefinition:
@ -185,7 +185,7 @@ public class CommandsForKeySerializerTest
case Erased:
case ErasedOrInvalidOrVestigial:
case Invalidated:
return Command.SerializerSupport.invalidated(txnId, Listeners.Immutable.EMPTY);
return Command.SerializerSupport.invalidated(txnId);
}
}
@ -356,7 +356,7 @@ public class CommandsForKeySerializerTest
@Test
public void serde()
{
testOne(-669467611022826851L);
testOne(-8928257345122888710L);
Random random = new Random();
for (int i = 0 ; i < 10000 ; ++i)
{
@ -508,7 +508,7 @@ public class CommandsForKeySerializerTest
for (int i = 0; i < info.length; i++)
{
InternalStatus status = rs.pick(InternalStatus.values());
info[i] = TxnInfo.create(ids[i], status, ids[i], CommandsForKey.NO_TXNIDS, Ballot.ZERO);
info[i] = TxnInfo.create(ids[i], status, ids[i], TxnId.NO_TXNIDS, Ballot.ZERO);
}
Gen<Unmanaged.Pending> pendingGen = Gens.enums().allMixedDistribution(Unmanaged.Pending.class).next(rs);
@ -537,7 +537,7 @@ public class CommandsForKeySerializerTest
}
else unmanaged = CommandsForKey.NO_PENDING_UNMANAGED;
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
@ -553,8 +553,8 @@ public class CommandsForKeySerializerTest
PartitionKey pk = new PartitionKey(TableId.fromString("1b255f4d-ef25-40a6-0000-000000000009"), key);
TxnId txnId = TxnId.fromValues(11,34052499,2,1);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk,
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, txnId, CommandsForKey.NO_TXNIDS, Ballot.ZERO) },
CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE);
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) },
CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);

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

@ -29,7 +29,6 @@ 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;
@ -262,7 +261,7 @@ public class AccordGenerators
case Erased:
case ErasedOrInvalidOrVestigial:
case Invalidated:
return Command.SerializerSupport.invalidated(txnId, Listeners.Immutable.EMPTY);
return Command.SerializerSupport.invalidated(txnId);
}
}
}