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

perf improvements

This commit is contained in:
Benedict Elliott Smith 2024-02-28 15:50:04 +00:00 committed by David Capwell
parent 599cd59736
commit 7dadc080cb
40 changed files with 654 additions and 320 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit ef36616441bd4ff4fec5379d986c75ad5a62ff7d
Subproject commit 3562bb3c9ce4e9eecdf65e236e968ef3ee9e0a86

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

@ -47,9 +47,9 @@ public class AccordSpec
public volatile DurationSpec fast_path_update_delay = new DurationSpec.IntSecondsBound(5);
public volatile DurationSpec schedule_durability_frequency = new DurationSpec.IntSecondsBound(15);
public volatile DurationSpec schedule_durability_frequency = new DurationSpec.IntSecondsBound(5);
public volatile DurationSpec durability_txnid_lag = new DurationSpec.IntSecondsBound(5);
public volatile DurationSpec shard_durability_cycle = new DurationSpec.IntMinutesBound(2);
public volatile DurationSpec shard_durability_cycle = new DurationSpec.IntMinutesBound(1);
public volatile DurationSpec global_durability_cycle = new DurationSpec.IntMinutesBound(10);
public enum TransactionalRangeMigration

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -500,6 +500,7 @@ public class Config
public volatile int counter_cache_keys_to_save = Integer.MAX_VALUE;
public DataStorageSpec.LongMebibytesBound paxos_cache_size = null;
public DataStorageSpec.LongMebibytesBound accord_cache_size = null;
public DataStorageSpec.LongMebibytesBound consensus_migration_cache_size = null;

22
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -227,6 +227,7 @@ public class DatabaseDescriptor
private static long keyCacheSizeInMiB;
private static long paxosCacheSizeInMiB;
private static long accordCacheSizeInMiB;
private static long consensusMigrationCacheSizeInMiB;
private static long counterCacheSizeInMiB;
private static long indexSummaryCapacityInMiB;
@ -963,6 +964,22 @@ public class DatabaseDescriptor
+ conf.paxos_cache_size + "', supported values are <integer> >= 0.", false);
}
try
{
// if paxosCacheSizeInMiB option was set to "auto" then size of the cache should be "max(10% of Heap (in MB), 1MB)
accordCacheSizeInMiB = (conf.accord_cache_size == null)
? Math.max(1, (int) ((Runtime.getRuntime().totalMemory() * 0.10) / 1024 / 1024))
: conf.accord_cache_size.toMebibytes();
if (accordCacheSizeInMiB < 0)
throw new NumberFormatException(); // to escape duplicating error message
}
catch (NumberFormatException e)
{
throw new ConfigurationException("paxos_cache_size option was set incorrectly to '"
+ conf.paxos_cache_size + "', supported values are <integer> >= 0.", false);
}
try
{
// if consensusMigrationCacheSizeInMiB option was set to "auto" then size of the cache should be "min(1% of Heap (in MB), 50MB)
@ -4266,6 +4283,11 @@ public class DatabaseDescriptor
return paxosCacheSizeInMiB;
}
public static long getAccordCacheSizeInMiB()
{
return accordCacheSizeInMiB;
}
public static long getConsensusMigrationCacheSizeInMiB()
{
return consensusMigrationCacheSizeInMiB;

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

@ -978,11 +978,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
if (redundantBeforeEntry == null)
return row;
TxnId redundantBeforeTxnId = redundantBeforeEntry.shardRedundantBefore();
if (redundantBeforeTxnId.equals(TxnId.NONE))
return row;
return CommandsForKeysAccessor.withoutRedundantCommands(partitionKey, row, redundantBeforeTxnId);
return CommandsForKeysAccessor.withoutRedundantCommands(partitionKey, row, redundantBeforeEntry);
}
@Override

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

@ -268,7 +268,7 @@ public enum Verb
PAXOS2_PREPARE_REQ (40, P2, writeTimeout, MUTATION, () -> PaxosPrepare.requestSerializer, () -> PaxosPrepare.requestHandler, PAXOS2_PREPARE_RSP ),
PAXOS2_PREPARE_REFRESH_RSP (51, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepareRefresh.responseSerializer, RESPONSE_HANDLER ),
PAXOS2_PREPARE_REFRESH_REQ (41, P2, writeTimeout, MUTATION, () -> PaxosPrepareRefresh.requestSerializer, () -> PaxosPrepareRefresh.requestHandler, PAXOS2_PREPARE_REFRESH_RSP ),
PAXOS2_PROPOSE_RSP (52, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPropose.ACCEPT_RESULT_SERIALIZER, RESPONSE_HANDLER ),
PAXOS2_PROPOSE_RSP (52, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPropose.ACCEPT_RESULT_SERIALIZER, RESPONSE_HANDLER ),
PAXOS2_PROPOSE_REQ (42, P2, writeTimeout, MUTATION, () -> PaxosPropose.requestSerializer, () -> PaxosPropose.requestHandler, PAXOS2_PROPOSE_RSP ),
PAXOS2_COMMIT_AND_PREPARE_RSP (53, P2, writeTimeout, REQUEST_RESPONSE, () -> PaxosPrepare.responseSerializer, RESPONSE_HANDLER ),
PAXOS2_COMMIT_AND_PREPARE_REQ (43, P2, writeTimeout, MUTATION, () -> PaxosCommitAndPrepare.requestSerializer, () -> PaxosCommitAndPrepare.requestHandler, PAXOS2_COMMIT_AND_PREPARE_RSP ),
@ -307,41 +307,41 @@ public enum Verb
DATA_MOVEMENT_EXECUTED_REQ (817, P1, rpcTimeout, MISC, () -> DataMovement.Status.serializer, () -> DataMovements.instance, DATA_MOVEMENT_EXECUTED_RSP ),
// accord
ACCORD_SIMPLE_RSP (119, P2, writeTimeout, REQUEST_RESPONSE, () -> EnumSerializer.simpleReply, RESPONSE_HANDLER ),
ACCORD_PRE_ACCEPT_RSP (120, P2, writeTimeout, REQUEST_RESPONSE, () -> PreacceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_SIMPLE_RSP (119, P2, writeTimeout, IMMEDIATE, () -> EnumSerializer.simpleReply, RESPONSE_HANDLER ),
ACCORD_PRE_ACCEPT_RSP (120, P2, writeTimeout, IMMEDIATE, () -> PreacceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_PRE_ACCEPT_REQ (121, P2, writeTimeout, IMMEDIATE, () -> PreacceptSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_PRE_ACCEPT_RSP ),
ACCORD_ACCEPT_RSP (122, P2, writeTimeout, REQUEST_RESPONSE, () -> AcceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_ACCEPT_RSP (122, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_ACCEPT_REQ (123, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_ACCEPT_RSP ),
ACCORD_ACCEPT_INVALIDATE_REQ (124, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.invalidate, AccordService::verbHandlerOrNoop, ACCORD_ACCEPT_RSP ),
ACCORD_READ_RSP (125, P2, writeTimeout, REQUEST_RESPONSE, () -> ReadDataSerializers.reply, RESPONSE_HANDLER ),
ACCORD_READ_RSP (125, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.reply, RESPONSE_HANDLER ),
ACCORD_READ_REQ (126, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.readData, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
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, REQUEST_RESPONSE, () -> ApplySerializers.reply, RESPONSE_HANDLER ),
ACCORD_APPLY_REQ (130, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_APPLY_RSP ),
ACCORD_BEGIN_RECOVER_RSP (131, P2, writeTimeout, REQUEST_RESPONSE, () -> RecoverySerializers.reply, RESPONSE_HANDLER ),
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_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, REQUEST_RESPONSE, () -> BeginInvalidationSerializers.reply, RESPONSE_HANDLER ),
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, REQUEST_RESPONSE, () -> WaitOnCommitSerializer.reply, RESPONSE_HANDLER ),
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_INFORM_DURABLE_REQ (140, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, REQUEST_RESPONSE, () -> CheckStatusSerializers.reply, RESPONSE_HANDLER ),
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, REQUEST_RESPONSE, () -> GetDepsSerializers.reply, RESPONSE_HANDLER ),
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_GET_EPHMRL_READ_DEPS_RSP (161, P2, writeTimeout, REQUEST_RESPONSE, () -> GetEphmrlReadDepsSerializers.reply, RESPONSE_HANDLER ),
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, REQUEST_RESPONSE, () -> GetMaxConflictSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_MAX_CONFLICT_RSP (163, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_MAX_CONFLICT_REQ (164, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP),
ACCORD_FETCH_DATA_RSP (145, P2, repairTimeout,REQUEST_RESPONSE, () -> FetchSerializers.reply, RESPONSE_HANDLER ),
ACCORD_FETCH_DATA_RSP (145, P2, repairTimeout,IMMEDIATE, () -> FetchSerializers.reply, RESPONSE_HANDLER ),
ACCORD_FETCH_DATA_REQ (146, P2, repairTimeout,IMMEDIATE, () -> FetchSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_FETCH_DATA_RSP ),
ACCORD_SET_SHARD_DURABLE_REQ (147, P2, writeTimeout, IMMEDIATE, () -> SetDurableSerializers.shardDurable, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_SET_GLOBALLY_DURABLE_REQ (148, P2, writeTimeout, IMMEDIATE, () -> SetDurableSerializers.globallyDurable,AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_QUERY_DURABLE_BEFORE_RSP (149, P2, writeTimeout, REQUEST_RESPONSE, () -> QueryDurableBeforeSerializers.reply, RESPONSE_HANDLER ),
ACCORD_QUERY_DURABLE_BEFORE_RSP (149, P2, writeTimeout, IMMEDIATE, () -> QueryDurableBeforeSerializers.reply, RESPONSE_HANDLER ),
ACCORD_QUERY_DURABLE_BEFORE_REQ (150, P2, writeTimeout, IMMEDIATE, () -> QueryDurableBeforeSerializers.request,AccordService::verbHandlerOrNoop, ACCORD_QUERY_DURABLE_BEFORE_RSP ),
ACCORD_SYNC_NOTIFY_REQ (151, P2, writeTimeout, IMMEDIATE, () -> Notification.listSerializer, () -> AccordSyncPropagator.verbHandler, ACCORD_SIMPLE_RSP ),

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

@ -123,7 +123,7 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
int estimatedSizeOnHeap(ToLongFunction<V> estimator)
{
shouldUpdateSize = false;
shouldUpdateSize = false; // TODO (expected): probably not the safest place to clear need to compute size
return lastQueriedEstimatedSizeOnHeap = Ints.checkedCast(EMPTY_SIZE + estimateStateOnHeapSize(estimator));
}
@ -204,7 +204,7 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
protected State<K, V> state(State<K, V> next)
{
State<K, V> prev = state;
if (prev != next)
if (prev != next) // TODO (expected): we change state to transition the cache state machine but often keep payload the same - so shouldn't recompute
shouldUpdateSize = true;
return state = next;
}

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

@ -42,7 +42,7 @@ import accord.api.Agent;
import accord.api.DataStore;
import accord.api.Key;
import accord.api.ProgressLog;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.impl.CommandsSummary;
import accord.impl.TimestampsForKey;
import accord.local.Command;
@ -114,8 +114,8 @@ public class AccordCommandStore extends CommandStore implements CacheSize
private final ExecutionOrder executionOrder;
private final AccordStateCache stateCache;
private final AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache;
private final AccordStateCache.Instance<RoutableKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKeyCache;
private final AccordStateCache.Instance<RoutableKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKeyCache;
private final AccordStateCache.Instance<Key, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKeyCache;
private final AccordStateCache.Instance<Key, CommandsForKey, AccordSafeCommandsForKey> commandsForKeyCache;
private AsyncOperation<?> currentOperation = null;
private AccordSafeCommandStore current = null;
private long lastSystemTimestampMicros = Long.MIN_VALUE;
@ -161,7 +161,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
this::validateCommand,
AccordObjectSizes::command);
timestampsForKeyCache =
stateCache.instance(RoutableKey.class,
stateCache.instance(Key.class,
AccordSafeTimestampsForKey.class,
AccordSafeTimestampsForKey::new,
this::loadTimestampsForKey,
@ -169,7 +169,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
this::validateTimestampsForKey,
AccordObjectSizes::timestampsForKey);
commandsForKeyCache =
stateCache.instance(RoutableKey.class,
stateCache.instance(Key.class,
AccordSafeCommandsForKey.class,
AccordSafeCommandsForKey::new,
this::loadCommandsForKey,
@ -220,7 +220,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
MessageProvider messageProvider = journal.makeMessageProvider(txnId);
SerializerSupport.TxnAndDeps txnAndDeps = SerializerSupport.extractTxnAndDeps(status, accepted, messageProvider);
SerializerSupport.TxnAndDeps txnAndDeps = SerializerSupport.extractTxnAndDeps(unsafeRangesForEpoch(), status, accepted, messageProvider);
Seekables<?, ?> keys = txnAndDeps.txn.keys();
if (keys.domain() != Routable.Domain.Range)
throw new AssertionError(String.format("Txn keys are not range for %s", txnAndDeps.txn));
@ -311,12 +311,12 @@ public class AccordCommandStore extends CommandStore implements CacheSize
return commandCache;
}
public AccordStateCache.Instance<RoutableKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKeyCache()
public AccordStateCache.Instance<Key, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKeyCache()
{
return timestampsForKeyCache;
}
public AccordStateCache.Instance<RoutableKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKeyCache()
public AccordStateCache.Instance<Key, CommandsForKey, AccordSafeCommandsForKey> commandsForKeyCache()
{
return commandsForKeyCache;
}
@ -466,8 +466,8 @@ public class AccordCommandStore extends CommandStore implements CacheSize
public AccordSafeCommandStore beginOperation(PreLoadContext preLoadContext,
Map<TxnId, AccordSafeCommand> commands,
NavigableMap<RoutableKey, AccordSafeTimestampsForKey> timestampsForKeys,
NavigableMap<RoutableKey, AccordSafeCommandsForKey> commandsForKeys)
NavigableMap<Key, AccordSafeTimestampsForKey> timestampsForKeys,
NavigableMap<Key, AccordSafeCommandsForKey> commandsForKeys)
{
Invariants.checkState(current == null);
commands.values().forEach(AccordSafeState::preExecute);

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

@ -31,6 +31,7 @@ import accord.primitives.Range;
import accord.topology.Topology;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.metrics.CacheSizeMetrics;
import org.apache.cassandra.schema.TableId;
@ -47,7 +48,7 @@ public class AccordCommandStores extends CommandStores implements CacheSize
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, AccordJournal journal)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
setCapacity(maxCacheSize());
setCapacity(DatabaseDescriptor.getAccordCacheSizeInMiB() << 20);
this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this);
}
@ -110,11 +111,6 @@ public class AccordCommandStores extends CommandStores implements CacheSize
forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCapacity(perStore));
}
private static long maxCacheSize()
{
return 5 << 20; // TODO (required): make configurable
}
@Override
public synchronized Supplier<EpochReady> updateTopology(Node node, Topology newTopology, boolean startSync)
{

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

@ -23,6 +23,7 @@ import java.time.Duration;
import accord.config.LocalConfig;
import org.apache.cassandra.config.Config;
// TODO (expected): should this be merged with AccordSpec?
public class AccordConfiguration implements LocalConfig
{
private final Config config;

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

@ -21,6 +21,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -570,6 +571,7 @@ public class AccordJournal implements Shutdownable
Key(Timestamp timestamp, Type type)
{
if (timestamp == null) throw new NullPointerException("Null timestamp for type " + type);
this.timestamp = timestamp;
this.type = type;
}
@ -1361,6 +1363,19 @@ public class AccordJournal implements Shutdownable
return presentMessages;
}
public Set<MessageType> all()
{
Set<Type> types = EnumSet.allOf(Type.class);
Set<Key> keys = new ObjectHashSet<>(types.size() + 1, 0.9f);
for (Type type : types)
keys.add(new Key(txnId, type));
Set<Key> presentKeys = journal.test(keys);
Set<MessageType> presentMessages = new ObjectHashSet<>(presentKeys.size() + 1, 0.9f);
for (Key key : presentKeys)
presentMessages.add(key.type.outgoingType);
return presentMessages;
}
@Override
public PreAccept preAccept()
{

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

@ -45,7 +45,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.Command.WaitingOn;
@ -593,7 +593,7 @@ public class AccordKeyspace
}
// TODO (expected): garbage-free filtering, reusing encoding
public Row withoutRedundantCommands(PartitionKey key, Row row, TxnId redundantBefore)
public Row withoutRedundantCommands(PartitionKey key, Row row, RedundantBefore.Entry redundantBefore)
{
Invariants.checkState(row.columnCount() == 1);
Cell<?> cell = row.getCell(data);
@ -601,7 +601,10 @@ public class AccordKeyspace
return row;
CommandsForKey current = CommandsForKeySerializer.fromBytes(key, cell.buffer());
CommandsForKey updated = current.withoutRedundant(redundantBefore);
if (current == null)
return null;
CommandsForKey updated = current.withRedundantBefore(redundantBefore);
if (current == updated)
return row;
@ -823,7 +826,7 @@ public class AccordKeyspace
Command.Committed committed = command.asCommitted();
Command.Committed originalCommitted = original != null && original.isCommitted() ? original.asCommitted() : null;
if (originalCommitted == null || committed.waitingOn != originalCommitted.waitingOn)
builder.addCell(live(CommandsColumns.waiting_on, timestampMicros, WaitingOnSerializer.serialize(committed.waitingOn)));
builder.addCell(live(CommandsColumns.waiting_on, timestampMicros, WaitingOnSerializer.serialize(committed.txnId(), committed.waitingOn)));
}
Row row = builder.build();
@ -1190,10 +1193,10 @@ public class AccordKeyspace
Ballot promised = deserializePromisedOrNull(row);
Ballot accepted = deserializeAcceptedOrNull(row);
WaitingOnProvider waitingOn = deserializeWaitingOn(row);
WaitingOnProvider waitingOn = deserializeWaitingOn(txnId, row);
MessageProvider messages = commandStore.makeMessageProvider(txnId);
return SerializerSupport.reconstruct(attrs, status, executeAt, promised, accepted, waitingOn, messages);
return SerializerSupport.reconstruct(commandStore.unsafeRangesForEpoch(), attrs, status, executeAt, promised, accepted, waitingOn, messages);
}
catch (Throwable t)
{
@ -1271,7 +1274,7 @@ public class AccordKeyspace
return deserializeTimestampOrNull(row.getBlob("accepted_ballot"), Ballot::fromBits);
}
private static WaitingOnProvider deserializeWaitingOn(UntypedResultSet.Row row)
private static WaitingOnProvider deserializeWaitingOn(TxnId txnId, UntypedResultSet.Row row)
{
ByteBuffer bytes = row.getBlob("waiting_on");
@ -1285,7 +1288,7 @@ public class AccordKeyspace
try
{
return WaitingOnSerializer.deserialize(deps, bytes);
return WaitingOnSerializer.deserialize(txnId, deps.keyDeps.keys(), deps.rangeDeps.txnIds(), bytes);
}
catch (IOException e)
{

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

@ -24,11 +24,12 @@ import java.util.function.ToLongFunction;
import accord.api.Key;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.impl.CommandsForKey;
import accord.impl.CommandsForKey.Info;
import accord.local.CommandsForKey;
import accord.local.CommandsForKey.TxnInfo;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.Command.WaitingOn;
import accord.local.CommandsForKey.TxnInfoWithMissing;
import accord.local.CommonAttributes;
import accord.local.Node;
import accord.local.SaveStatus;
@ -305,7 +306,6 @@ public class AccordObjectSizes
return ACCEPTED;
case Committed:
case Stable:
case ReadyToExecute:
return COMMITTED;
case PreApplied:
case Applied:
@ -363,22 +363,25 @@ public class AccordObjectSizes
}
private static long EMPTY_CFK_SIZE = measure(new CommandsForKey(null));
private static long EMPTY_INFO_SIZE = measure(CommandsForKey.Info.createMock(null, null, null));
private static long EMPTY_INFO_SIZE = measure(TxnInfo.createMock(TxnId.NONE, null, null, null));
private static long EMPTY_INFO_WITH_MISSING_ADDITIONAL_SIZE = measure(TxnInfo.createMock(TxnId.NONE, null, null, null)) - EMPTY_INFO_SIZE;
public static long commandsForKey(CommandsForKey cfk)
{
long size = EMPTY_CFK_SIZE;
size += key(cfk.key());
size += 2 * ObjectSizes.sizeOfReferenceArray(cfk.size());
size += cfk.size() * TIMESTAMP_SIZE;
size += ObjectSizes.sizeOfReferenceArray(cfk.size());
size += cfk.size() * EMPTY_INFO_SIZE;
for (int i = 0 ; i < cfk.size() ; ++i)
{
Info info = cfk.info(i);
if (info.getClass() == CommandsForKey.NoInfo.class)
continue;
size += EMPTY_INFO_SIZE;
if (info.missing.length > 0)
size += ObjectSizes.sizeOfReferenceArray(info.missing.length);
TxnInfo info = cfk.get(i);
if (info.getClass() != TxnInfoWithMissing.class) continue;
TxnInfoWithMissing infoWithMissing = (TxnInfoWithMissing) info;
if (infoWithMissing.missing.length > 0)
{
size += EMPTY_INFO_WITH_MISSING_ADDITIONAL_SIZE;
size += ObjectSizes.sizeOfReferenceArray(infoWithMissing.missing.length);
size += infoWithMissing.missing.length * TIMESTAMP_SIZE;
}
}
return size;
}

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

@ -31,8 +31,7 @@ import accord.api.DataStore;
import accord.api.Key;
import accord.api.ProgressLog;
import accord.impl.AbstractSafeCommandStore;
import accord.impl.CommandsForKey;
import accord.impl.CommandsForKeys;
import accord.local.CommandsForKey;
import accord.impl.CommandsSummary;
import accord.local.Command;
import accord.local.CommandStores.RangesForEpoch;
@ -41,7 +40,6 @@ import accord.local.PreLoadContext;
import accord.primitives.AbstractKeys;
import accord.primitives.Deps;
import accord.primitives.Ranges;
import accord.primitives.RoutableKey;
import accord.primitives.Routables;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
@ -53,16 +51,16 @@ import static accord.primitives.Routable.Domain.Range;
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey>
{
private final Map<TxnId, AccordSafeCommand> commands;
private final NavigableMap<RoutableKey, AccordSafeCommandsForKey> commandsForKeys;
private final NavigableMap<RoutableKey, AccordSafeTimestampsForKey> timestampsForKeys;
private final NavigableMap<Key, AccordSafeCommandsForKey> commandsForKeys;
private final NavigableMap<Key, AccordSafeTimestampsForKey> timestampsForKeys;
private final AccordCommandStore commandStore;
private final RangesForEpoch ranges;
CommandsForRanges.Updater rangeUpdates = null;
public AccordSafeCommandStore(PreLoadContext context,
Map<TxnId, AccordSafeCommand> commands,
NavigableMap<RoutableKey, AccordSafeTimestampsForKey> timestampsForKey,
NavigableMap<RoutableKey, AccordSafeCommandsForKey> commandsForKey,
NavigableMap<Key, AccordSafeTimestampsForKey> timestampsForKey,
NavigableMap<Key, AccordSafeCommandsForKey> commandsForKey,
AccordCommandStore commandStore)
{
super(context);
@ -94,7 +92,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected AccordSafeCommandsForKey getCommandsForKeyInternal(RoutableKey key)
protected AccordSafeCommandsForKey getCommandsForKeyInternal(Key key)
{
return commandsForKeys.get(key);
}
@ -106,7 +104,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected AccordSafeCommandsForKey getCommandsForKeyIfLoaded(RoutableKey key)
protected AccordSafeCommandsForKey getCommandsForKeyIfLoaded(Key key)
{
AccordSafeCommandsForKey cfk = commandStore.commandsForKeyCache().acquireIfLoaded(key);
if (cfk != null) cfk.preExecute();
@ -114,7 +112,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected AccordSafeTimestampsForKey getTimestampsForKeyInternal(RoutableKey key)
protected AccordSafeTimestampsForKey getTimestampsForKeyInternal(Key key)
{
return timestampsForKeys.get(key);
}
@ -126,7 +124,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected AccordSafeTimestampsForKey getTimestampsForKeyIfLoaded(RoutableKey key)
protected AccordSafeTimestampsForKey getTimestampsForKeyIfLoaded(Key key)
{
AccordSafeTimestampsForKey cfk = commandStore.timestampsForKeyCache().acquireIfLoaded(key);
if (cfk != null) cfk.preExecute();
@ -177,14 +175,14 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
Ranges allRanges = ranges.all();
deps.keyDeps.keys().forEach(allRanges, key -> {
// TODO (now): batch register to minimise GC
deps.keyDeps.forEach(key, txnId -> {
deps.keyDeps.forEach(key, (txnId, txnIdx) -> {
// TODO (desired, efficiency): this can be made more efficient by batching by epoch
if (ranges.coordinates(txnId).contains(key))
return; // already coordinates, no need to replicate
if (!ranges.allBefore(txnId.epoch()).contains(key))
return;
CommandsForKeys.registerNotWitnessed(this, key, txnId);
get(key).registerHistorical(this, txnId);
});
});
CommandsForRanges commandsForRanges = commandStore.commandsForRanges();
@ -221,7 +219,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
for (Key key : keys)
{
if (!slice.contains(key)) continue;
CommandsForKey commands = commandsForKey(key).current();
CommandsForKey commands = get(key).current();
accumulate = map.apply(commands, accumulate);
}
}
@ -233,11 +231,11 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
Routables<?> sliced = keysOrRanges.slice(slice, Routables.Slice.Minimal);
if (!context.keys().slice(slice, Routables.Slice.Minimal).containsAll(sliced))
throw new AssertionError("Range(s) detected not present in the PreLoadContext: expected " + context.keys() + " but given " + keysOrRanges);
for (RoutableKey key : timestampsForKeys.keySet())
for (Key key : commandsForKeys.keySet())
{
//TODO (duplicate code): this is a repeat of Key... only change is checking contains in range
if (!sliced.contains(key)) continue;
CommandsForKey commands = commandsForKey(key).current();
CommandsForKey commands = get(key).current();
accumulate = map.apply(commands, accumulate);
}
}
@ -263,22 +261,20 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
@Override
protected void update(Command prev, Command updated, @Nullable Seekables<?, ?> keysOrRanges)
protected void update(Command prev, Command updated)
{
super.update(prev, updated, keysOrRanges);
super.update(prev, updated);
if (updated.txnId().domain() == Range && CommandsForKey.needsUpdate(prev, updated))
{
Seekables<?, ?> keysOrRanges = updated.keysOrRanges();
if (keysOrRanges == null) keysOrRanges = prev.keysOrRanges();
if (keysOrRanges == null)
{
if (updated.known().isDefinitionKnown()) keysOrRanges = updated.partialTxn().keys();
else if (prev.known().isDefinitionKnown()) keysOrRanges = prev.partialTxn().keys();
else return;
}
List<TxnId> waitingOn;
return;
if (updated.partialDeps() == null) waitingOn = Collections.emptyList();
List<TxnId> waitingOn;
// TODO (required): this is faulty: we cannot simply save the raw transaction ids, as they may be for other ranges
if (updated.partialDeps() == null) waitingOn = Collections.emptyList();
else waitingOn = updated.partialDeps().txnIds();
updateRanges().put(updated.txnId(), (Ranges)keysOrRanges, updated.saveStatus(), updated.executeAt(), waitingOn);
}
@ -300,8 +296,8 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
public void postExecute(Map<TxnId, AccordSafeCommand> commands,
Map<RoutableKey, AccordSafeTimestampsForKey> timestampsForKey,
Map<RoutableKey, AccordSafeCommandsForKey> commandsForKeys
Map<Key, AccordSafeTimestampsForKey> timestampsForKey,
Map<Key, AccordSafeCommandsForKey> commandsForKeys
)
{
postExecute();

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

@ -23,18 +23,17 @@ import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Key;
import accord.impl.CommandsForKey;
import accord.impl.SafeCommandsForKey;
import accord.primitives.RoutableKey;
import accord.local.CommandsForKey;
import accord.local.SafeCommandsForKey;
public class AccordSafeCommandsForKey extends SafeCommandsForKey implements AccordSafeState<RoutableKey, CommandsForKey>
public class AccordSafeCommandsForKey extends SafeCommandsForKey implements AccordSafeState<Key, CommandsForKey>
{
private boolean invalidated;
private final AccordCachingState<RoutableKey, CommandsForKey> global;
private final AccordCachingState<Key, CommandsForKey> global;
private CommandsForKey original;
private CommandsForKey current;
public AccordSafeCommandsForKey(AccordCachingState<RoutableKey, CommandsForKey> global)
public AccordSafeCommandsForKey(AccordCachingState<Key, CommandsForKey> global)
{
super((Key) global.key());
this.global = global;
@ -83,7 +82,7 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
}
@Override
public AccordCachingState<RoutableKey, CommandsForKey> global()
public AccordCachingState<Key, CommandsForKey> global()
{
checkNotInvalidated();
return global;

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

@ -25,6 +25,8 @@ public interface AccordSafeState<K, V> extends SafeState<V>
{
void set(V update);
V original();
void invalidate();
boolean invalidated();
void preExecute();
void postExecute();
AccordCachingState<K, V> global();

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

@ -26,17 +26,16 @@ import com.google.common.annotations.VisibleForTesting;
import accord.api.Key;
import accord.impl.SafeTimestampsForKey;
import accord.impl.TimestampsForKey;
import accord.primitives.RoutableKey;
import accord.primitives.Timestamp;
public class AccordSafeTimestampsForKey extends SafeTimestampsForKey implements AccordSafeState<RoutableKey, TimestampsForKey>
public class AccordSafeTimestampsForKey extends SafeTimestampsForKey implements AccordSafeState<Key, TimestampsForKey>
{
private boolean invalidated;
private final AccordCachingState<RoutableKey, TimestampsForKey> global;
private final AccordCachingState<Key, TimestampsForKey> global;
private TimestampsForKey original;
private TimestampsForKey current;
public AccordSafeTimestampsForKey(AccordCachingState<RoutableKey, TimestampsForKey> global)
public AccordSafeTimestampsForKey(AccordCachingState<Key, TimestampsForKey> global)
{
super((Key) global.key());
this.global = global;
@ -71,7 +70,7 @@ public class AccordSafeTimestampsForKey extends SafeTimestampsForKey implements
}
@Override
public AccordCachingState<RoutableKey, TimestampsForKey> global()
public AccordCachingState<Key, TimestampsForKey> global()
{
checkNotInvalidated();
return global;

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

@ -127,7 +127,7 @@ public class AccordService implements IAccordService, Shutdownable
private final AccordDataStore dataStore;
private final AccordJournal journal;
private final CoordinateDurabilityScheduling durabilityScheduling;
private final AccordVerbHandler<? extends Request> verbHandler;
private final AccordVerbHandler<? extends Request> requestHandler;
private final LocalConfig configuration;
@GuardedBy("this")
private State state = State.INIT;
@ -296,7 +296,7 @@ public class AccordService implements IAccordService, Shutdownable
configuration);
this.nodeShutdown = toShutdownable(node);
this.durabilityScheduling = new CoordinateDurabilityScheduling(node);
this.verbHandler = new AccordVerbHandler<>(node, configService, journal);
this.requestHandler = new AccordVerbHandler<>(node, configService, journal);
}
@Override
@ -313,14 +313,14 @@ public class AccordService implements IAccordService, Shutdownable
durabilityScheduling.setShardCycleTime(Ints.checkedCast(DatabaseDescriptor.getAccordShardDurabilityCycle(SECONDS)), SECONDS);
durabilityScheduling.setTxnIdLag(Ints.checkedCast(DatabaseDescriptor.getAccordScheduleDurabilityTxnIdLag(SECONDS)), TimeUnit.SECONDS);
durabilityScheduling.setFrequency(Ints.checkedCast(DatabaseDescriptor.getAccordScheduleDurabilityFrequency(SECONDS)), SECONDS);
durabilityScheduling.start();
// durabilityScheduling.start();
state = State.STARTED;
}
@Override
public IVerbHandler<? extends Request> verbHandler()
{
return verbHandler;
return requestHandler;
}
@Override

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

@ -41,7 +41,7 @@ import com.google.common.collect.ImmutableSortedMap;
import accord.api.Key;
import accord.api.RoutingKey;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.impl.CommandsSummary;
import accord.local.Command;
import accord.local.SaveStatus;
@ -64,7 +64,6 @@ import org.apache.cassandra.utils.IntervalTree;
import static accord.local.SafeCommandStore.*;
import static accord.local.SafeCommandStore.TestDep.ANY_DEPS;
import static accord.local.SafeCommandStore.TestDep.WITH;
import static accord.local.SafeCommandStore.TestStartedAt.ANY;
import static accord.local.SafeCommandStore.TestStartedAt.STARTED_BEFORE;
import static accord.local.SafeCommandStore.TestStatus.ANY_STATUS;
import static accord.local.Status.Stable;
@ -379,11 +378,13 @@ public class CommandsForRanges
private static Range toRange(Interval<RoutableKey, RangeCommandSummary> interval)
{
TokenKey start = (TokenKey) interval.min;
TokenKey end = (TokenKey) interval.max;
AccordRoutingKey start = (AccordRoutingKey) interval.min;
if (!(start instanceof AccordRoutingKey.SentinelKey))
start = new TokenKey(start.table(), start.token().decreaseSlightly());
AccordRoutingKey end = (AccordRoutingKey) interval.max;
// TODO (required, correctness) : accord doesn't support wrap around, so decreaseSlightly may fail in some cases
// TODO (required, correctness) : this logic is mostly used for testing, so is it actually safe for all partitioners?
return new TokenRange(start.withToken(start.token().decreaseSlightly()), end);
return new TokenRange(start, end);
}
@Nullable

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

@ -117,6 +117,9 @@ public class AccordAgent implements Agent
public boolean isExpired(TxnId initiated, long now)
{
// TODO: should distinguish between reads and writes
if (initiated.kind().isSyncPoint())
return false;
return now - initiated.hlc() > getReadRpcTimeout(MICROSECONDS);
}

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

@ -33,12 +33,12 @@ import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.RoutingKey;
import accord.local.KeyHistory;
import accord.local.PreLoadContext;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.RoutableKey;
import accord.primitives.Seekables;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@ -121,7 +121,7 @@ public class AsyncLoader
}
}
private void referenceAndAssembleReadsForKey(RoutableKey key,
private void referenceAndAssembleReadsForKey(Key key,
AsyncOperation.Context context,
List<AsyncChain<?>> listenChains)
{
@ -157,7 +157,7 @@ public class AsyncLoader
{
case Key:
// cast to Keys fails...
Iterable<RoutableKey> keys = (Iterable<RoutableKey>) keysOrRanges;
Iterable<Key> keys = (Iterable<Key>) keysOrRanges;
keys.forEach(key -> referenceAndAssembleReadsForKey(key, context, chains));
break;
case Range:
@ -172,7 +172,7 @@ public class AsyncLoader
private AsyncChain<?> referenceAndDispatchReadsForRange(AsyncOperation.Context context)
{
AsyncChain<Set<? extends RoutableKey>> overlappingKeys = findOverlappingKeys((Ranges) keysOrRanges);
AsyncChain<Set<? extends Key>> overlappingKeys = findOverlappingKeys((Ranges) keysOrRanges);
return overlappingKeys.flatMap(keys -> {
if (keys.isEmpty())
@ -183,14 +183,14 @@ public class AsyncLoader
}, commandStore);
}
private AsyncChain<Set<? extends RoutableKey>> findOverlappingKeys(Ranges ranges)
private AsyncChain<Set<? extends Key>> findOverlappingKeys(Ranges ranges)
{
Invariants.checkArgument(!ranges.isEmpty());
List<AsyncChain<Set<PartitionKey>>> chains = new ArrayList<>(ranges.size());
for (Range range : ranges)
chains.add(findOverlappingKeys(range));
return AsyncChains.reduce(chains, (a, b) -> ImmutableSet.<RoutableKey>builder().addAll(a).addAll(b).build());
return AsyncChains.reduce(chains, (a, b) -> ImmutableSet.<Key>builder().addAll(a).addAll(b).build());
}
private AsyncChain<Set<PartitionKey>> findOverlappingKeys(Range range)

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

@ -30,6 +30,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import accord.api.Key;
import accord.local.CommandStore;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
@ -67,8 +68,8 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
static class Context
{
final HashMap<TxnId, AccordSafeCommand> commands = new HashMap<>();
final TreeMap<RoutableKey, AccordSafeTimestampsForKey> timestampsForKey = new TreeMap<>();
final TreeMap<RoutableKey, AccordSafeCommandsForKey> commandsForKey = new TreeMap<>();
final TreeMap<Key, AccordSafeTimestampsForKey> timestampsForKey = new TreeMap<>();
final TreeMap<Key, AccordSafeCommandsForKey> commandsForKey = new TreeMap<>();
void releaseResources(AccordCommandStore commandStore)
{

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

@ -132,7 +132,6 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
case PreCommitted:
case Committed:
case PreApplied:
case ReadyToExecute:
synchronized (this)
{
waitingOn.set(safeStore.commandStore().id());
@ -249,7 +248,6 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
case PreCommitted:
case Committed:
case PreApplied:
case ReadyToExecute:
return;
case Applied:

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

@ -24,10 +24,10 @@ import java.util.Arrays;
import com.google.common.primitives.Ints;
import accord.api.Key;
import accord.impl.CommandsForKey;
import accord.impl.CommandsForKey.Info;
import accord.impl.CommandsForKey.InternalStatus;
import accord.impl.CommandsForKey.NoInfo;
import accord.local.CommandsForKey;
import accord.local.CommandsForKey.TxnInfo;
import accord.local.CommandsForKey.InternalStatus;
import accord.local.CommandsForKey.Unmanaged;
import accord.local.Node;
import accord.primitives.Routable.Domain;
import accord.primitives.Timestamp;
@ -35,8 +35,10 @@ import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.utils.vint.VIntCoding;
import static accord.local.CommandsForKey.NO_PENDING_UNMANAGED;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Read;
import static accord.primitives.Txn.Kind.Write;
@ -104,23 +106,7 @@ public class CommandsForKeySerializer
{
int commandCount = cfk.size();
if (commandCount == 0)
{
// TODO (expected): we should not need to special-case, but best solution here is not to store redundantBefore;
// but this requires some modest deeper changes, so for now special-case serialization when empty
Timestamp redundantBefore = cfk.redundantBefore();
ByteBuffer out = ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(0) +
TypeSizes.sizeofUnsignedVInt(redundantBefore.epoch()) +
TypeSizes.sizeofUnsignedVInt(redundantBefore.hlc()) +
TypeSizes.sizeofUnsignedVInt(redundantBefore.flags()) +
TypeSizes.sizeofUnsignedVInt(redundantBefore.node.id));
VIntCoding.writeUnsignedVInt32(0, out);
VIntCoding.writeUnsignedVInt(redundantBefore.epoch(), out);
VIntCoding.writeUnsignedVInt(redundantBefore.hlc(), out);
VIntCoding.writeUnsignedVInt32(redundantBefore.flags(), out);
VIntCoding.writeUnsignedVInt32(redundantBefore.node.id, out);
out.flip();
return out;
}
return ByteBuffer.allocate(1);
int[] nodeIds = cachedInts().getInts(Math.min(64, commandCount));
try
@ -129,11 +115,9 @@ public class CommandsForKeySerializer
// 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, missingIdCount = 0, executeAtCount = 0, bitsPerExecuteAtFlags = 0;
int nodeIdCount = 0, missingIdCount = 0, executeAtCount = 0, bitsPerExecuteAtFlags = 0;
int bitsPerExecuteAtEpochDelta = 0, bitsPerExecuteAtHlcDelta = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
{
nodeIdCount = 1;
nodeIds[0] = cfk.redundantBefore().node.id;
for (int i = 0 ; i < commandCount ; ++i)
{
if (nodeIdCount + 1 >= nodeIds.length)
@ -143,24 +127,19 @@ public class CommandsForKeySerializer
nodeIds = cachedInts().resize(nodeIds, nodeIds.length, nodeIds.length * 2);
}
TxnId txnId = cfk.txnId(i);
Info info = cfk.info(i);
TxnInfo txn = cfk.get(i);
hasNonStandardFlags |= txnIdFlags(txnId) != STANDARD;
nodeIds[nodeIdCount++] = txnId.node.id;
hasNonStandardFlags |= txnIdFlags(txn) != STANDARD;
nodeIds[nodeIdCount++] = txn.node.id;
if (info.getClass() == NoInfo.class)
missingIdCount += txn.missing().length;
if (txn.executeAt == txn)
continue;
missingIdCount += info.missing.length;
if (info.executeAt == txnId)
continue;
nodeIds[nodeIdCount++] = info.executeAt.node.id;
bitsPerExecuteAtEpochDelta = Math.max(bitsPerExecuteAtEpochDelta, numberOfBitsToRepresent(info.executeAt.epoch() - txnId.epoch()));
bitsPerExecuteAtHlcDelta = Math.max(bitsPerExecuteAtHlcDelta, numberOfBitsToRepresent(info.executeAt.hlc() - txnId.hlc()));
bitsPerExecuteAtFlags = Math.max(bitsPerExecuteAtFlags, numberOfBitsToRepresent(info.executeAt.flags()));
nodeIds[nodeIdCount++] = txn.executeAt.node.id;
bitsPerExecuteAtEpochDelta = Math.max(bitsPerExecuteAtEpochDelta, numberOfBitsToRepresent(txn.executeAt.epoch() - txn.epoch()));
bitsPerExecuteAtHlcDelta = Math.max(bitsPerExecuteAtHlcDelta, numberOfBitsToRepresent(txn.executeAt.hlc() - txn.hlc()));
bitsPerExecuteAtFlags = Math.max(bitsPerExecuteAtFlags, numberOfBitsToRepresent(txn.executeAt.flags()));
executeAtCount += 1;
}
nodeIdCount = compact(nodeIds);
@ -175,8 +154,8 @@ public class CommandsForKeySerializer
int maxHeaderBits = minHeaderBits;
int totalBytes = 0;
long prevEpoch = cfk.redundantBefore().epoch();
long prevHlc = cfk.redundantBefore().hlc();
long prevEpoch = cfk.get(0).epoch();
long prevHlc = cfk.get(0).hlc();
int[] bytesHistogram = cachedInts().getInts(12);
Arrays.fill(bytesHistogram, 0);
for (int i = 0 ; i < commandCount ; ++i)
@ -214,7 +193,7 @@ public class CommandsForKeySerializer
if (hasNonStandardFlags && txnIdFlags(txnId) == RAW)
totalBytes += 2;
Info info = cfk.info(i);
TxnInfo info = cfk.get(i);
if (info.status.hasInfo)
headerBits += infoHeaderBits;
maxHeaderBits = Math.max(headerBits, maxHeaderBits);
@ -247,9 +226,11 @@ public class CommandsForKeySerializer
// then pick third number as 75th %ile, but at least 1 less than highest, and one more than second
// finally, ensure third then second are distributed so that there is no more than a gap of 4 between them and the next
int l0 = Math.max(0, Math.min(3, minBasicBytes - headerBytes));
int l1 = Math.max(l0+1, Math.min(l0+4,Arrays.binarySearch(bytesHistogram, commandCount/4) - headerBytes));
int l1 = Arrays.binarySearch(bytesHistogram, minBasicBytes, maxBasicBytes, commandCount/4);
l1 = Math.max(l0+1, Math.min(l0+4, (l1 < 0 ? -1 - l1 : l1) - headerBytes));
int l3 = Math.max(l1+2, maxBasicBytes - headerBytes);
int l2 = Math.max(l1+1, Math.min(l3-1, Arrays.binarySearch(bytesHistogram, (3*commandCount)/4) - headerBytes));
int l2 = Arrays.binarySearch(bytesHistogram, minBasicBytes, maxBasicBytes,(3*commandCount)/4);
l2 = Math.max(l1+1, Math.min(l3-1, (l2 < 0 ? -1 -l2 : l2) - headerBytes));
while (l3-l2 > 4) ++l2;
while (l2-l1 > 4) ++l1;
hlcBytesLookup = setHlcBytes(l0, l1, l2, l3);
@ -267,16 +248,13 @@ public class CommandsForKeySerializer
totalBytes += TypeSizes.sizeofUnsignedVInt(nodeIds[i] - nodeIds[i-1]);
totalBytes += 2;
Arrays.fill(bytesHistogram, minBasicBytes, maxBasicBytes + 1, 0);
cachedInts().forceDiscard(bytesHistogram);
prevEpoch = cfk.redundantBefore().epoch();
prevHlc = cfk.redundantBefore().hlc();
prevEpoch = cfk.get(0).epoch();
prevHlc = cfk.get(0).hlc();
// account for encoding redundantBefore
totalBytes += TypeSizes.sizeofUnsignedVInt(prevEpoch);
totalBytes += TypeSizes.sizeofUnsignedVInt(prevHlc);
totalBytes += 2; // flags TODO (expected): pack this along with uniqueIdBits, as usually zero bits should be needed
totalBytes += (bitsPerNodeId+7)/8;
if (missingIdCount + executeAtCount > 0)
{
@ -291,6 +269,20 @@ public class CommandsForKeySerializer
totalBytes += 2;
}
// count unmanaged bytes
int unmanagedPendingCommitCount = 0;
for (int i = 0 ; i < cfk.unmanagedCount() ; ++i)
{
Unmanaged unmanaged = cfk.getUnmanaged(i);
if (unmanaged.pending == Unmanaged.Pending.COMMIT)
++unmanagedPendingCommitCount;
totalBytes += CommandSerializers.txnId.serializedSize();
// TODO (desired): this could be more efficient, e.g. referencing one of the TxnInfo indexes for timestamp
totalBytes += CommandSerializers.timestamp.serializedSize();
}
totalBytes += TypeSizes.sizeofUnsignedVInt(unmanagedPendingCommitCount);
totalBytes += TypeSizes.sizeofUnsignedVInt(cfk.unmanagedCount() - unmanagedPendingCommitCount);
ByteBuffer out = ByteBuffer.allocate(totalBytes);
VIntCoding.writeUnsignedVInt32(commandCount, out);
VIntCoding.writeUnsignedVInt32(nodeIdCount, out);
@ -301,8 +293,6 @@ public class CommandsForKeySerializer
VIntCoding.writeUnsignedVInt(prevEpoch, out);
VIntCoding.writeUnsignedVInt(prevHlc, out);
out.putShort((short) cfk.redundantBefore().flags());
writeLeastSignificantBytes(Arrays.binarySearch(nodeIds, 0, nodeIdCount, cfk.redundantBefore().node.id), (bitsPerNodeId+7)/8, out);
int executeAtMask = executeAtCount > 0 ? 1 : 0;
int missingDepsMask = missingIdCount > 0 ? 1 : 0;
@ -311,7 +301,7 @@ public class CommandsForKeySerializer
for (int i = 0 ; i < commandCount ; ++i)
{
TxnId txnId = cfk.txnId(i);
Info info = cfk.info(i);
TxnInfo info = cfk.get(i);
InternalStatus status = info.status;
long bits = status.ordinal();
@ -322,7 +312,7 @@ public class CommandsForKeySerializer
bits |= hasExecuteAt << bitIndex;
bitIndex += statusHasInfo & executeAtMask;
long hasMissingIds = info.missing != CommandsForKey.NO_TXNIDS ? 1 : 0;
long hasMissingIds = info.missing() != CommandsForKey.NO_TXNIDS ? 1 : 0;
bits |= hasMissingIds << bitIndex;
bitIndex += statusHasInfo & missingDepsMask;
@ -392,6 +382,20 @@ public class CommandsForKeySerializer
}
}
VIntCoding.writeUnsignedVInt32(unmanagedPendingCommitCount, out);
VIntCoding.writeUnsignedVInt32(cfk.unmanagedCount() - unmanagedPendingCommitCount, out);
Unmanaged.Pending pending = unmanagedPendingCommitCount == 0 ? Unmanaged.Pending.APPLY : Unmanaged.Pending.COMMIT;
for (int i = 0 ; i < cfk.unmanagedCount() ; ++i)
{
Unmanaged unmanaged = cfk.getUnmanaged(i);
Invariants.checkState(unmanaged.pending == pending);
CommandSerializers.txnId.serialize(unmanaged.txnId, out, ByteBufferAccessor.instance, out.position());
out.position(out.position() + CommandSerializers.txnId.serializedSize());
CommandSerializers.timestamp.serialize(unmanaged.waitingUntil, out, ByteBufferAccessor.instance, out.position());
out.position(out.position() + CommandSerializers.timestamp.serializedSize());
if (--unmanagedPendingCommitCount == 0) pending = Unmanaged.Pending.APPLY;
}
if ((executeAtCount | missingIdCount) > 0)
{
int bitsPerCommandId = numberOfBitsToRepresent(commandCount);
@ -407,21 +411,17 @@ public class CommandsForKeySerializer
for (int i = 0 ; i < commandCount ; ++i)
{
Info info = cfk.info(i);
if (info.getClass() == NoInfo.class)
continue;
TxnId txnId = cfk.txnId(i);
if (info.executeAt != txnId)
TxnInfo txn = cfk.get(i);
if (txn.executeAt != txn)
{
Timestamp executeAt = info.executeAt;
Timestamp executeAt = txn.executeAt;
int nodeIdx = Arrays.binarySearch(nodeIds, 0, nodeIdCount, executeAt.node.id);
if (bitsPerExecuteAt <= 64)
{
Invariants.checkState(executeAt.epoch() >= txnId.epoch());
long executeAtBits = executeAt.epoch() - txnId.epoch();
Invariants.checkState(executeAt.epoch() >= txn.epoch());
long executeAtBits = executeAt.epoch() - txn.epoch();
int offset = bitsPerExecuteAtEpochDelta;
executeAtBits |= (executeAt.hlc() - txnId.hlc()) << offset ;
executeAtBits |= (executeAt.hlc() - txn.hlc()) << offset ;
offset += bitsPerExecuteAtHlcDelta;
executeAtBits |= ((long)executeAt.flags()) << offset;
offset += bitsPerExecuteAtFlags;
@ -431,9 +431,9 @@ public class CommandsForKeySerializer
}
else
{
buffer = flushBits(buffer, bufferCount, executeAt.epoch() - txnId.epoch(), bitsPerExecuteAtEpochDelta, out);
buffer = flushBits(buffer, bufferCount, executeAt.epoch() - txn.epoch(), bitsPerExecuteAtEpochDelta, out);
bufferCount = (bufferCount + bitsPerExecuteAtEpochDelta) & 63;
buffer = flushBits(buffer, bufferCount, executeAt.hlc() - txnId.hlc(), bitsPerExecuteAtHlcDelta, out);
buffer = flushBits(buffer, bufferCount, executeAt.hlc() - txn.hlc(), bitsPerExecuteAtHlcDelta, out);
bufferCount = (bufferCount + bitsPerExecuteAtHlcDelta) & 63;
buffer = flushBits(buffer, bufferCount, executeAt.flags(), bitsPerExecuteAtFlags, out);
bufferCount = (bufferCount + bitsPerExecuteAtFlags) & 63;
@ -442,16 +442,17 @@ public class CommandsForKeySerializer
}
}
if (info.missing.length > 0)
TxnId[] missing = txn.missing();
if (missing.length > 0)
{
int j = 0;
while (j < info.missing.length - 1)
while (j < missing.length - 1)
{
int missingId = cfk.indexOf(info.missing[j++]);
int missingId = cfk.indexOf(missing[j++]);
buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
bufferCount = (bufferCount + bitsPerMissingId) & 63;
}
int missingId = cfk.indexOf(info.missing[info.missing.length - 1]);
int missingId = cfk.indexOf(missing[missing.length - 1]);
missingId |= 1L << bitsPerCommandId;
buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
bufferCount = (bufferCount + bitsPerMissingId) & 63;
@ -461,6 +462,7 @@ public class CommandsForKeySerializer
writeMostSignificantBytes(buffer, (bufferCount + 7)/8, out);
}
Invariants.checkState(!out.hasRemaining());
out.flip();
return out;
}
@ -494,16 +496,10 @@ public class CommandsForKeySerializer
in = in.duplicate();
int commandCount = VIntCoding.readUnsignedVInt32(in);
if (commandCount == 0)
{
long epoch = VIntCoding.readUnsignedVInt(in);
long hlc = VIntCoding.readUnsignedVInt(in);
int flags = VIntCoding.readUnsignedVInt32(in);
Node.Id id = new Node.Id(VIntCoding.readUnsignedVInt32(in));
return new CommandsForKey(key).withoutRedundant(TxnId.fromValues(epoch, hlc, flags, id));
}
return new CommandsForKey(key);
TxnId[] txnIds = new TxnId[commandCount];
Info[] infos = new Info[commandCount];
TxnId[] txnIds = cachedTxnIds().get(commandCount);
TxnInfo[] txns = new TxnInfo[commandCount];
int nodeIdCount = VIntCoding.readUnsignedVInt32(in);
int bitsPerNodeId = numberOfBitsToRepresent(nodeIdCount);
long nodeIdMask = (1L << bitsPerNodeId) - 1;
@ -526,12 +522,8 @@ public class CommandsForKeySerializer
hlcBytesLookup = setHlcByteDeltas((flags >>> 5) & 0x3, (flags >>> 7) & 0x3, (flags >>> 9) & 0x3, (flags >>> 11) & 0x3);
}
long prevEpoch = VIntCoding.readUnsignedVInt32(in);
long prevHlc = VIntCoding.readUnsignedVInt32(in);
TxnId redundantBefore = TxnId.fromValues(prevEpoch, prevHlc, in.getShort(),
nodeIds[(int)readLeastSignificantBytes((bitsPerNodeId+7)/8, in)]);
long prevEpoch = VIntCoding.readUnsignedVInt(in);
long prevHlc = VIntCoding.readUnsignedVInt(in);
for (int i = 0 ; i < commandCount ; ++i)
{
long header = readLeastSignificantBytes(headerByteCount, in);
@ -601,12 +593,34 @@ public class CommandsForKeySerializer
: TxnId.fromValues(epoch, hlc, flags, node);
txnIds[i] = txnId;
infos[i] = DECODE_INFOS[(executeAtInfoOffset | missingDepsInfoOffset)*STATUS_COUNT + status.ordinal()];
txns[i] = DECODE_INFOS[(executeAtInfoOffset | missingDepsInfoOffset)*STATUS_COUNT + status.ordinal()];
prevEpoch = epoch;
prevHlc = hlc;
}
int unmanagedPendingCommitCount = VIntCoding.readUnsignedVInt32(in);
int unmanagedCount = unmanagedPendingCommitCount + VIntCoding.readUnsignedVInt32(in);
Unmanaged[] unmanageds;
if (unmanagedCount == 0)
{
unmanageds = NO_PENDING_UNMANAGED;
}
else
{
unmanageds = new Unmanaged[unmanagedCount];
Unmanaged.Pending pending = unmanagedPendingCommitCount == 0 ? Unmanaged.Pending.APPLY : Unmanaged.Pending.COMMIT;
for (int i = 0 ; i < unmanagedCount ; ++i)
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, ByteBufferAccessor.instance, in.position());
in.position(in.position() + CommandSerializers.txnId.serializedSize());
Timestamp waitingUntil = CommandSerializers.timestamp.deserialize(in, ByteBufferAccessor.instance, in.position());
in.position(in.position() + CommandSerializers.timestamp.serializedSize());
unmanageds[i] = new Unmanaged(pending, txnId, waitingUntil);
if (--unmanagedPendingCommitCount == 0) pending = Unmanaged.Pending.APPLY;
}
}
if (executeAtMasks + missingDepsMasks > 0)
{
TxnId[] missingIdBuffer = cachedTxnIds().get(8);
@ -630,13 +644,10 @@ public class CommandsForKeySerializer
for (int i = 0 ; i < commandCount ; ++i)
{
Info info = infos[i];
if (info.getClass() == NoInfo.class)
continue;
TxnId txnId = txnIds[i];
Timestamp executeAt = txnId;
if (info.executeAt == null)
TxnInfo placeholder = txns[i];
Timestamp executeAt;
if (placeholder.executeAt == null)
{
long epoch, hlc;
int flags;
@ -661,8 +672,12 @@ public class CommandsForKeySerializer
}
executeAt = Timestamp.fromValues(epoch, hlc, flags, id);
}
else
{
executeAt = txnId;
}
TxnId[] missing = info.missing;
TxnId[] missing = placeholder.missing();
if (missing == null)
{
int prev = -1;
@ -684,13 +699,19 @@ public class CommandsForKeySerializer
missingIdCount = 0;
}
infos[i] = Info.create(txnId, info.status, executeAt, missing);
txns[i] = TxnInfo.create(txnId, placeholder.status, executeAt, missing);
}
cachedTxnIds().forceDiscard(missingIdBuffer, maxIdBufferCount);
}
else
{
for (int i = 0 ; i < commandCount ; ++i)
txns[i] = TxnInfo.create(txnIds[i], txns[i].status, txnIds[i]);
}
cachedTxnIds().forceDiscard(txnIds, commandCount);
return CommandsForKey.SerializerSupport.create(key, redundantBefore, txnIds, infos);
return CommandsForKey.SerializerSupport.create(key, txns, unmanageds);
}
private static int getHlcBytes(int lookup, int index)
@ -833,16 +854,16 @@ public class CommandsForKeySerializer
private static final Txn.Kind[] TXN_ID_FLAG_BITS_KIND_LOOKUP = new Txn.Kind[] { Read, Write, ExclusiveSyncPoint, null };
private static final int STATUS_COUNT = InternalStatus.values().length;
private static final Info[] DECODE_INFOS = new Info[4 * STATUS_COUNT];
private static final TxnInfo[] DECODE_INFOS = new TxnInfo[4 * STATUS_COUNT];
static
{
for (InternalStatus status : InternalStatus.values())
{
int ordinal = status.ordinal();
DECODE_INFOS[ordinal] = status.asNoInfo;
DECODE_INFOS[STATUS_COUNT+ordinal] = Info.createMock(status, Timestamp.NONE, null);
DECODE_INFOS[2*STATUS_COUNT+ordinal] = Info.createMock(status, null, CommandsForKey.NO_TXNIDS);
DECODE_INFOS[3*STATUS_COUNT+ordinal] = Info.createMock(status, null, null);
DECODE_INFOS[ordinal] = TxnInfo.createMock(TxnId.NONE, status, TxnId.NONE, CommandsForKey.NO_TXNIDS);
DECODE_INFOS[STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, TxnId.NONE, null);
DECODE_INFOS[2*STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, null, CommandsForKey.NO_TXNIDS);
DECODE_INFOS[3*STATUS_COUNT+ordinal] = TxnInfo.createMock(TxnId.NONE, status, null, null);
}
}

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

@ -22,40 +22,63 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import accord.local.Command.WaitingOn;
import accord.primitives.Deps;
import accord.primitives.Keys;
import accord.primitives.Routable;
import accord.primitives.TxnId;
import accord.utils.ImmutableBitSet;
import accord.utils.Invariants;
import accord.utils.SimpleBitSet;
import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.vint.VIntCoding;
public class WaitingOnSerializer
{
public static void serialize(WaitingOn waitingOn, DataOutputPlus out) throws IOException
public static void serialize(TxnId txnId, WaitingOn waitingOn, DataOutputPlus out) throws IOException
{
// TODO (expected): use run length encoding; we know that at most 1/3rd of bits will be set between the three bitsets
int length = (waitingOn.deps.txnIdCount() + 63) / 64;
serialize(length, waitingOn.waitingOnCommit, out);
serialize(length, waitingOn.waitingOnApply, out);
serialize(length, waitingOn.appliedOrInvalidated, out);
out.writeUnsignedVInt32(waitingOn.keys.size());
out.writeUnsignedVInt32(waitingOn.txnIds.size());
int keyCount = waitingOn.keys.size();
int txnIdCount = waitingOn.txnIds.size();
int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
serialize(waitingOnLength, waitingOn.waitingOn, out);
if (txnId.domain() == Routable.Domain.Range)
{
int appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
serialize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated, out);
}
}
public static WaitingOn deserialize(Deps deps, DataInputPlus in) throws IOException
public static WaitingOn deserialize(TxnId txnId, Keys keys, SortedArrayList<TxnId> txnIds, DataInputPlus in) throws IOException
{
int length = (deps.txnIdCount() + 63) / 64;
ImmutableBitSet waitingOnCommit = deserialize(length, in);
ImmutableBitSet waitingOnApply = deserialize(length, in);
ImmutableBitSet appliedOrInvalidated = deserialize(length, in);
return new WaitingOn(deps, waitingOnCommit, waitingOnApply, appliedOrInvalidated);
int a = in.readUnsignedVInt32();
int b = in.readUnsignedVInt32();
int waitingOnLength = (txnIds.size() + keys.size() + 63) / 64;
ImmutableBitSet waitingOn = deserialize(waitingOnLength, in);
ImmutableBitSet appliedOrInvalidated = null;
if (txnId.domain() == Routable.Domain.Range)
{
int appliedOrInvalidatedLength = (txnIds.size() + 63) / 64;
appliedOrInvalidated = deserialize(appliedOrInvalidatedLength, in);
}
return new WaitingOn(keys, txnIds, waitingOn, appliedOrInvalidated);
}
public static long serializedSize(WaitingOn waitingOn)
{
int length = (waitingOn.deps.txnIdCount() + 63) / 64;
return serializedSize(length, waitingOn.waitingOnCommit)
+ serializedSize(length, waitingOn.waitingOnApply)
+ serializedSize(length, waitingOn.appliedOrInvalidated);
int keyCount = waitingOn.keys.size();
int txnIdCount = waitingOn.txnIds.size();
int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
long size = serializedSize(waitingOnLength, waitingOn.waitingOn);
size += TypeSizes.sizeofUnsignedVInt(keyCount);
size += TypeSizes.sizeofUnsignedVInt(txnIdCount);
if (waitingOn.appliedOrInvalidated == null)
return size;
int appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
return size + serializedSize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated);
}
private static void serialize(int length, SimpleBitSet write, DataOutputPlus out) throws IOException
@ -81,14 +104,23 @@ public class WaitingOnSerializer
return (long) TypeSizes.LONG_SIZE * length;
}
public static ByteBuffer serialize(WaitingOn waitingOn) throws IOException
public static ByteBuffer serialize(TxnId txnId, WaitingOn waitingOn) throws IOException
{
int length = (waitingOn.deps.txnIdCount() + 63) / 64;
ByteBuffer out = ByteBuffer.allocate(TypeSizes.LONG_SIZE * length * 3);
serialize(length, waitingOn.waitingOnCommit, out);
serialize(length, waitingOn.waitingOnApply, out);
serialize(length, waitingOn.appliedOrInvalidated, out);
return (ByteBuffer) out.flip();
int keyCount = waitingOn.keys.size();
int txnIdCount = waitingOn.txnIds.size();
int waitingOnLength = (txnIdCount + keyCount + 63) / 64;
int appliedOrInvalidatedLength = 0;
if (txnId.domain() == Routable.Domain.Range)
appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
ByteBuffer out = ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(keyCount) + TypeSizes.sizeofUnsignedVInt(txnIdCount)
+ TypeSizes.LONG_SIZE * (waitingOnLength + appliedOrInvalidatedLength));
VIntCoding.writeUnsignedVInt32(keyCount, out);
VIntCoding.writeUnsignedVInt32(txnIdCount, out);
serialize(waitingOnLength, waitingOn.waitingOn, out);
if (appliedOrInvalidatedLength > 0)
serialize(appliedOrInvalidatedLength, waitingOn.appliedOrInvalidated, out);
return out.flip();
}
private static void serialize(int length, SimpleBitSet write, ByteBuffer out)
@ -99,16 +131,23 @@ public class WaitingOnSerializer
out.putLong(bits[i]);
}
public static WaitingOn deserialize(Deps deps, ByteBuffer in) throws IOException
public static WaitingOn deserialize(TxnId txnId, Keys keys, SortedArrayList<TxnId> txnIds, ByteBuffer in) throws IOException
{
int length = (deps.txnIdCount() + 63) / 64;
int waitingOnLength = (txnIds.size() + keys.size() + 63) / 64;
int position = in.position();
ImmutableBitSet waitingOnCommit = deserialize(position, length, in);
position += length*8;
ImmutableBitSet waitingOnApply = deserialize(position, length, in);
position += length*8;
ImmutableBitSet appliedOrInvalidated = deserialize(position, length, in);
return new WaitingOn(deps, waitingOnCommit, waitingOnApply, appliedOrInvalidated);
int a = VIntCoding.readUnsignedVInt32(in, position);
position += TypeSizes.sizeofUnsignedVInt(a);
int b = VIntCoding.readUnsignedVInt32(in, position);
position += TypeSizes.sizeofUnsignedVInt(a);
ImmutableBitSet waitingOn = deserialize(position, waitingOnLength, in);
ImmutableBitSet appliedOrInvalidated = null;
if (txnId.domain() == Routable.Domain.Range)
{
position += waitingOnLength*8;
int appliedOrInvalidatedLength = (txnIds.size() + 63) / 64;
appliedOrInvalidated = deserialize(position, appliedOrInvalidatedLength, in);
}
return new WaitingOn(keys, txnIds, waitingOn, appliedOrInvalidated);
}
private static ImmutableBitSet deserialize(int position, int length, ByteBuffer in)

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

@ -35,9 +35,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.DataStore;
import accord.api.Key;
import accord.api.Write;
import accord.impl.AbstractSafeCommandStore;
import accord.impl.CommandsForKeys;
import accord.impl.TimestampsForKeys;
import accord.impl.TimestampsForKey;
import accord.local.SafeCommandStore;
import accord.primitives.PartialTxn;
@ -375,7 +376,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
// TODO (expected, efficiency): 99.9999% of the time we can just use executeAt.hlc(), so can avoid bringing
// cfk into memory by retaining at all times in memory key ranges that are dirty and must use this logic;
// any that aren't can just use executeAt.hlc
TimestampsForKey cfk = CommandsForKeys.updateLastExecutionTimestamps((AbstractSafeCommandStore<?,?,?>) safeStore, (RoutableKey) key, executeAt, true);
TimestampsForKey cfk = TimestampsForKeys.updateLastExecutionTimestamps((AbstractSafeCommandStore<?,?,?>) safeStore, (Key) key, executeAt, true);
long timestamp = AccordSafeTimestampsForKey.timestampMicrosFor(cfk, executeAt, true);
// TODO (low priority - do we need to compute nowInSeconds, or can we just use executeAt?)
int nowInSeconds = AccordSafeTimestampsForKey.nowInSecondsFor(cfk, executeAt, true);

47
src/java/org/apache/cassandra/utils/vint/VIntCoding.java
View File

@ -126,13 +126,27 @@ public class VIntCoding
return retval;
}
@DontInline
private static long readUnsignedVIntSlow(ByteBuffer in, int position, byte firstByte)
{
int size = numberOfExtraBytesToRead(firstByte);
long retval = firstByte & firstByteValueMask(size);
for (int ii = 0; ii < size; ii++)
{
byte b = in.get(position++);
retval <<= 8;
retval |= b & 0xff;
}
return retval;
}
public static long readUnsignedVInt(ByteBuffer in)
{
byte firstByte = in.get();
if (firstByte >= 0)
return firstByte;
int position = in.position();
int limit = in.limit();
if (limit - position < 8)
@ -155,6 +169,32 @@ public class VIntCoding
return retval;
}
public static long readUnsignedVInt(ByteBuffer in, int position)
{
byte firstByte = in.get(position++);
if (firstByte >= 0)
return firstByte;
int limit = in.limit();
if (limit - position < 8)
return readUnsignedVIntSlow(in, position, firstByte);
int extraBytes = VIntCoding.numberOfExtraBytesToRead(firstByte);
int extraBits = extraBytes * 8;
long retval = in.getLong(position);
if (in.order() == ByteOrder.LITTLE_ENDIAN)
retval = Long.reverseBytes(retval);
// truncate the bytes we read in excess of those we needed
retval >>>= 64 - extraBits;
// remove the non-value bits from the first byte
firstByte &= VIntCoding.firstByteValueMask(extraBytes);
// shift the first byte up to its correct position
retval |= (long) firstByte << extraBits;
return retval;
}
public static void skipUnsignedVInt(DataInputPlus input) throws IOException
{
int firstByte = input.readByte();
@ -330,6 +370,11 @@ public class VIntCoding
return checkedCast(readUnsignedVInt(input));
}
public static int readUnsignedVInt32(ByteBuffer input, int position)
{
return checkedCast(readUnsignedVInt(input, position));
}
// & this with the first byte to give the value part for a given extraBytesToRead encoded in the byte
public static int firstByteValueMask(int extraBytesToRead)
{

56
test/conf/logback-dtest-quiet.xml Normal file
View File

@ -0,0 +1,56 @@
<!--
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.
-->
<configuration debug="false" scan="true" scanPeriod="60 seconds">
<define name="cluster_id" class="org.apache.cassandra.distributed.impl.ClusterIDDefiner" />
<define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
<!-- Shutdown hook ensures that async appender flushes -->
<shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
<appender name="INSTANCEFILE" class="ch.qos.logback.core.FileAppender">
<file>./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log</file>
<encoder>
<pattern>%-5level [%thread] ${instance_id} %date{ISO8601} %F:%L - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
<immediateFlush>true</immediateFlush>
</appender>
<appender name="INSTANCESTDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>ERROR</level>
</filter>
</appender>
<!-- Un-comment to enable TRACE logging for Accord transactions.
<logger name="accord" level="TRACE" />
<logger name="org.apache.cassandra.service.accord" level="TRACE" />
-->
<root level="DEBUG">
<appender-ref ref="INSTANCEFILE" /> <!-- use blocking to avoid race conditions with appending and searching -->
<appender-ref ref="INSTANCESTDERR" />
</root>
</configuration>

3
test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.api;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.Future;
import java.util.function.BiConsumer;
import org.apache.cassandra.distributed.shared.FutureUtils;
@ -60,6 +61,8 @@ public interface ICoordinator
}
SimpleQueryResult executeWithResult(String query, ConsistencyLevel consistencyLevel, Object... boundValues);
Future<?> executeWithResult(BiConsumer<SimpleQueryResult, Throwable> callback, String query, ConsistencyLevel consistencyLevel, Object... boundValues);
Future<?> executeWithResult(BiConsumer<SimpleQueryResult, Throwable> callback, String query, ConsistencyLevel serialConsistencyLevel, ConsistencyLevel commitConsistencyLevel, Object... boundValues);
default SimpleQueryResult executeWithResult(String query, ConsistencyLevel serialConsistencyLevel, ConsistencyLevel commitConsistencyLevel, Object... boundValues)
{

25
test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
View File

@ -25,6 +25,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Future;
import java.util.function.BiConsumer;
import com.google.common.collect.Iterators;
@ -62,6 +63,30 @@ public class Coordinator implements ICoordinator
return instance().sync(() -> unsafeExecuteInternal(query, consistencyLevel, boundValues)).call();
}
@Override
public Future<?> executeWithResult(BiConsumer<SimpleQueryResult, Throwable> callback, String query, ConsistencyLevel consistencyLevel, Object... boundValues)
{
return executeWithResult(callback, query, null, consistencyLevel, boundValues);
}
@Override
public Future<?> executeWithResult(BiConsumer<SimpleQueryResult, Throwable> callback, String query, ConsistencyLevel serialConsistencyLevel, ConsistencyLevel commitConsistencyLevel, Object... boundValues)
{
return instance().async(cb -> {
SimpleQueryResult result;
try
{
result = CoordinatorHelper.unsafeExecuteInternal(query, serialConsistencyLevel, commitConsistencyLevel, boundValues);
}
catch (Throwable t)
{
callback.accept(null, t);
return;
}
callback.accept(result, null);
}).apply(callback);
}
public Future<SimpleQueryResult> asyncExecuteWithTracingWithResult(UUID sessionId, String query, ConsistencyLevel consistencyLevelOrigin, Object... boundValues)
{
return instance.async(() -> {

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

@ -507,6 +507,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
public void receiveMessage(IMessage message)
{
sync(receiveMessageRunnable(message)).accept(false);
// async(receiveMessageRunnable(message)).apply(false);
}
@Override

96
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java Normal file
View File

@ -0,0 +1,96 @@
/*
* 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.io.IOException;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.Semaphore;
import com.google.common.util.concurrent.RateLimiter;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.utils.EstimatedHistogram;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
public class AccordLoadTest extends AccordTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordLoadTest.class);
@BeforeClass
public static void setUp() throws IOException
{
AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config.set("lwt_strategy", "accord").set("non_serial_write_strategy", "accord")), 2);
}
@Ignore
@Test
public void testLoad() throws Exception
{
test("CREATE TABLE " + qualifiedTableName + " (k int, v int, PRIMARY KEY(k))",
cluster -> {
ICoordinator coordinator = cluster.coordinator(1);
final int batchSize = 1000;
final int concurrency = 100;
final int ratePerSecond = 1000;
final int keyCount = 10;
for (int i = 1; i <= keyCount; i++)
coordinator.execute("INSERT INTO " + qualifiedTableName + " (k, v) VALUES (0, 0) USING TIMESTAMP 0;", ConsistencyLevel.ALL, i);
Random random = new Random();
// CopyOnWriteArrayList<Throwable> exceptions = new CopyOnWriteArrayList<>();
final Semaphore inFlight = new Semaphore(concurrency);
final RateLimiter rateLimiter = RateLimiter.create(ratePerSecond);
long testStart = System.nanoTime();
// while (NANOSECONDS.toMinutes(System.nanoTime() - testStart) < 10 && exceptions.size() < 10000)
while (true)
{
final EstimatedHistogram histogram = new EstimatedHistogram(200);
long batchStart = System.nanoTime();
for (int i = 0 ; i < batchSize ; ++i)
{
inFlight.acquire();
rateLimiter.acquire();
long commandStart = System.nanoTime();
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
// else exceptions.add(fail);
}, "UPDATE " + qualifiedTableName + " SET v += 1 WHERE k = ? IF EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(keyCount));
}
System.out.printf("%tT rate: %.2f/s\n", new Date(), (((float)batchSize * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStart)));
System.out.printf("%tT percentiles: %d %d %d %d\n", new Date(), histogram.percentile(.25)/1000, histogram.percentile(.5)/1000, histogram.percentile(.75)/1000, histogram.percentile(1)/1000);
}
}
);
}
@Override
protected Logger logger()
{
return logger;
}
}

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

@ -38,7 +38,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.Result;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.CommandStore;

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

@ -31,8 +31,8 @@ import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.Result;
import accord.impl.CommandsForKey;
import accord.impl.CommandsForKeys;
import accord.local.CommandsForKey;
import accord.impl.TimestampsForKeys;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommonAttributes;
@ -41,10 +41,13 @@ import accord.messages.Apply;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Route;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.ImmutableBitSet;
@ -102,20 +105,20 @@ public class AccordCommandStoreTest
AtomicLong clock = new AtomicLong(0);
PartialTxn depTxn = createPartialTxn(0);
Key key = (Key)depTxn.keys().get(0);
Range range = key.toUnseekable().asRange();
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
QueryProcessor.executeInternal("INSERT INTO ks.tbl (k, c, v) VALUES (0, 0, 1)");
TableId tableId = Schema.instance.getTableMetadata("ks", "tbl").id;
TxnId oldTxnId1 = txnId(1, clock.incrementAndGet(), 1);
TxnId oldTxnId2 = txnId(1, clock.incrementAndGet(), 1);
TxnId oldTimestamp = txnId(1, clock.incrementAndGet(), 1);
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
TxnId oldTxnId1 = txnId(1, clock.incrementAndGet(), 1, Txn.Kind.Write, Routable.Domain.Range);
TxnId oldTxnId2 = txnId(1, clock.incrementAndGet(), 1, Txn.Kind.Write, Routable.Domain.Range);
TxnId txnId = txnId(1, clock.incrementAndGet(), 1, Txn.Kind.Write, Routable.Domain.Range);
PartialDeps dependencies;
try (PartialDeps.Builder builder = PartialDeps.builder(depTxn.covering()))
{
builder.add(key, oldTxnId1);
builder.add(key, oldTxnId2);
builder.add(range, oldTxnId1);
builder.add(range, oldTxnId2);
dependencies = builder.build();
}
@ -130,11 +133,9 @@ public class AccordCommandStoreTest
Ballot accepted = ballot(1, clock.incrementAndGet(), 1);
Timestamp executeAt = timestamp(1, clock.incrementAndGet(), 1);
attrs.partialDeps(dependencies);
SimpleBitSet waitingOnCommit = new SimpleBitSet(2);
waitingOnCommit.set(0);
SimpleBitSet waitingOnApply = new SimpleBitSet(2);
SimpleBitSet waitingOnApply = new SimpleBitSet(3);
waitingOnApply.set(1);
Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies, new ImmutableBitSet(waitingOnCommit), new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps.txnIds(), new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt);
@ -183,10 +184,10 @@ public class AccordCommandStoreTest
AccordSafeTimestampsForKey tfk = new AccordSafeTimestampsForKey(loaded(key, null));
tfk.initialize();
CommandsForKeys.updateLastExecutionTimestamps(commandStore, tfk, txnId1, true);
TimestampsForKeys.updateLastExecutionTimestamps(commandStore, tfk, txnId1, true);
Assert.assertEquals(txnId1.hlc(), AccordSafeTimestampsForKey.timestampMicrosFor(tfk.current(), txnId1, true));
CommandsForKeys.updateLastExecutionTimestamps(commandStore, tfk, txnId2, true);
TimestampsForKeys.updateLastExecutionTimestamps(commandStore, tfk, txnId2, true);
Assert.assertEquals(txnId2.hlc(), AccordSafeTimestampsForKey.timestampMicrosFor(tfk.current(), txnId2, true));
Assert.assertEquals(txnId2, tfk.current().lastExecutedTimestamp());

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

@ -26,7 +26,7 @@ import org.junit.Test;
import accord.api.Key;
import accord.api.RoutingKey;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.local.Node;
@ -117,7 +117,7 @@ public class AccordCommandTest
Assert.assertEquals(Status.PreAccepted, command.status());
Assert.assertTrue(command.partialDeps() == null || command.partialDeps().isEmpty());
CommandsForKey cfk = ((AccordSafeCommandStore) instance).commandsForKey(key(1)).current();
CommandsForKey cfk = ((AccordSafeCommandStore) instance).get(key(1)).current();
Assert.assertTrue(cfk.indexOf(txnId) >= 0);
}));
@ -145,7 +145,7 @@ public class AccordCommandTest
Assert.assertEquals(Status.Accepted, command.status());
Assert.assertEquals(deps, command.partialDeps());
CommandsForKey cfk = ((AccordSafeCommandStore) instance).commandsForKey(key(1)).current();
CommandsForKey cfk = ((AccordSafeCommandStore) instance).get(key(1)).current();
Assert.assertTrue(cfk.indexOf(txnId) >= 0);
}));
@ -160,7 +160,7 @@ public class AccordCommandTest
Assert.assertTrue(command.hasBeen(Status.Committed));
Assert.assertEquals(commit.partialDeps, command.partialDeps());
CommandsForKey cfk = ((AccordSafeCommandStore) instance).commandsForKey(key(1)).current();
CommandsForKey cfk = ((AccordSafeCommandStore) instance).get(key(1)).current();
Assert.assertTrue(cfk.indexOf(txnId) >= 0);
}));
}

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

@ -30,6 +30,8 @@ import java.util.function.LongSupplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.annotation.Nullable;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.junit.Assert;
@ -59,6 +61,7 @@ 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;
@ -199,8 +202,8 @@ public class AccordTestUtils
@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(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)
@ -213,6 +216,11 @@ public class AccordTestUtils
return new TxnId(epoch, hlc, kind, Key, new Node.Id(node));
}
public static TxnId txnId(long epoch, long hlc, int node, Txn.Kind kind, Routable.Domain domain)
{
return new TxnId(epoch, hlc, kind, domain, new Node.Id(node));
}
public static Timestamp timestamp(long epoch, long hlc, int node)
{
return Timestamp.fromValues(epoch, hlc, new Node.Id(node));

9
test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
View File

@ -32,13 +32,12 @@ import org.junit.BeforeClass;
import org.junit.Test;
import accord.api.Key;
import accord.impl.CommandsForKey;
import accord.local.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.primitives.Keys;
import accord.primitives.PartialTxn;
import accord.primitives.RoutableKey;
import accord.primitives.TxnId;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
@ -96,7 +95,7 @@ public class AsyncLoaderTest
AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache = commandStore.commandCache();
commandStore.executeBlocking(() -> commandStore.setCapacity(1024));
AccordStateCache.Instance<RoutableKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsCache = commandStore.timestampsForKeyCache();
AccordStateCache.Instance<Key, TimestampsForKey, AccordSafeTimestampsForKey> timestampsCache = commandStore.timestampsForKeyCache();
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
PartialTxn txn = createPartialTxn(0);
PartitionKey key = (PartitionKey) Iterables.getOnlyElement(txn.keys());
@ -375,7 +374,7 @@ public class AsyncLoaderTest
@Test
public void inProgressCFKSaveTest()
{
inProgressCFKSaveTest(COMMANDS, AccordCommandStore::commandsForKeyCache, context -> context.commandsForKey, CommandsForKey::new, (cfk, u) -> cfk.update(null, u));
this.inProgressCFKSaveTest(COMMANDS, AccordCommandStore::commandsForKeyCache, context -> context.commandsForKey, CommandsForKey::new, (cfk, u) -> cfk.update(null, u));
}
@Test
@ -384,7 +383,7 @@ public class AsyncLoaderTest
inProgressCFKSaveTest(TIMESTAMPS, AccordCommandStore::timestampsForKeyCache, context -> context.timestampsForKey, TimestampsForKey::new, (tfk, c) -> new TimestampsForKey(tfk.key(), c.executeAt(), c.executeAt().hlc(), c.executeAt()));
}
private <T1, T2 extends AccordSafeState<RoutableKey, T1>, C extends AccordStateCache.Instance<RoutableKey, T1, T2>> void inProgressCFKSaveTest(KeyHistory history, Function<AccordCommandStore, C> getter, Function<Context, TreeMap<?, ?>> inContext, Function<Key, T1> initialiser, BiFunction<T1, Command, T1> update)
private <T1, T2 extends AccordSafeState<Key, T1>, C extends AccordStateCache.Instance<Key, T1, T2>> void inProgressCFKSaveTest(KeyHistory history, Function<AccordCommandStore, C> getter, Function<Context, TreeMap<?, ?>> inContext, Function<Key, T1> initialiser, BiFunction<T1, Command, T1> update)
{
AtomicLong clock = new AtomicLong(0);
ManualExecutor executor = new ManualExecutor();

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

@ -36,7 +36,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.impl.SafeCommandsForKey;
import accord.local.SafeCommandsForKey;
import accord.local.CheckedCommands;
import accord.local.Command;
import accord.local.PreLoadContext;
@ -53,7 +53,6 @@ import accord.primitives.PartialDeps;
import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.RoutableKey;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -259,7 +258,7 @@ public class AsyncOperationTest
Accept accept =
Accept.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), txnId.epoch(), false, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
Commit commit =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.Commit, Ballot.ZERO, executeAt, partialTxn.keys(), partialTxn, deps, route, null);
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.CommitSlowPath, Ballot.ZERO, executeAt, partialTxn.keys(), partialTxn, deps, route, null);
Commit stable =
Commit.SerializerSupport.create(txnId, partialRoute, txnId.epoch(), Commit.Kind.StableSlowPath, Ballot.ZERO, executeAt, partialTxn.keys(), partialTxn, deps, route, null);
@ -482,8 +481,7 @@ public class AsyncOperationTest
}
try
{
//TODO this is due to bad typing for Instance, it doesn't use ? extends RoutableKey
assertNoReferences(commandStore.commandsForKeyCache(), (Iterable<RoutableKey>) (Iterable<?>) keys);
assertNoReferences(commandStore.commandsForKeyCache(), keys);
}
catch (AssertionError e)
{
@ -524,8 +522,7 @@ public class AsyncOperationTest
private static void awaitDone(AccordCommandStore commandStore, List<TxnId> ids, Keys keys)
{
awaitDone(commandStore.commandCache(), ids);
//TODO this is due to bad typing for Instance, it doesn't use ? extends RoutableKey
awaitDone(commandStore.commandsForKeyCache(), (Iterable<RoutableKey>) (Iterable<?>) keys);
awaitDone(commandStore.commandsForKeyCache(), keys);
}
private static <T> void awaitDone(AccordStateCache.Instance<T, ?, ?> cache, Iterable<T> keys)

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

@ -38,9 +38,10 @@ import org.junit.BeforeClass;
import org.junit.Test;
import accord.api.Key;
import accord.impl.CommandsForKey;
import accord.impl.CommandsForKey.InternalStatus;
import accord.local.CommandsForKey;
import accord.local.CommandsForKey.InternalStatus;
import accord.local.Command;
import accord.local.CommandsForKey.TxnInfo;
import accord.local.CommonAttributes;
import accord.local.CommonAttributes.Mutable;
import accord.local.Listeners;
@ -241,9 +242,12 @@ public class CommandsForKeySerializerTest
List<TxnId> deps = cmds[i].deps;
List<TxnId> missing = cmds[i].missing;
for (int j = 0 ; j < limit ; ++j)
if (i != j) deps.add(cmds[j].txnId);
{
if (i != j && cmds[i].txnId.kind().witnesses(cmds[j].txnId))
deps.add(cmds[j].txnId);
}
int missingCount = Math.min(limit - (limit > i ? 1 : 0), missingCountSupplier.getAsInt());
int missingCount = Math.min(deps.size(), missingCountSupplier.getAsInt());
while (missingCount > 0)
{
int remove = source.nextInt(deps.size());
@ -267,14 +271,14 @@ public class CommandsForKeySerializerTest
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
if (status == null || !status.hasInfo) continue;
if (status.depsKnownBefore(cmds[j].txnId, cmds[j].executeAt).compareTo(cmds[i].txnId) > 0 && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && status.depsKnownBefore(cmds[j].txnId, cmds[j].executeAt).compareTo(cmds[i].txnId) > 0 && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
for (int j = i + 1 ; j < cmds.length ; ++j)
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
if (status == null || !status.hasInfo) continue;
if (Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
cmds[i].invisible = true;
@ -322,7 +326,7 @@ public class CommandsForKeySerializerTest
@Test
public void serde()
{
// testOne(1821931462020409370L);
testOne(-6946067792202944553L);
Random random = new Random();
for (int i = 0 ; i < 10000 ; ++i)
{
@ -426,13 +430,13 @@ public class CommandsForKeySerializerTest
Assert.assertTrue(cmd.invisible);
continue;
}
CommandsForKey.Info info = cfk.info(i);
TxnInfo info = cfk.get(i);
InternalStatus expectStatus = InternalStatus.from(cmd.saveStatus);
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVELY_KNOWN;
if (expectStatus.hasInfo)
Assert.assertEquals(cmd.executeAt, info.executeAt(cfk.txnId(i)));
Assert.assertEquals(cmd.executeAt, info.executeAt);
Assert.assertEquals(expectStatus, info.status);
Assert.assertArrayEquals(cmd.missing.toArray(TxnId[]::new), info.missing);
Assert.assertArrayEquals(cmd.missing.toArray(TxnId[]::new), info.missing());
++i;
}
@ -461,11 +465,12 @@ public class CommandsForKeySerializerTest
next = txnIdGen.next(rs0);
return next;
}).unique().ofSizeBetween(0, 10).next(rs);
CommandsForKey.Info[] info = new CommandsForKey.Info[ids.length];
TxnInfo[] info = new TxnInfo[ids.length];
for (int i = 0; i < info.length; i++)
info[i] = rs.pick(InternalStatus.values()).asNoInfo;
Arrays.sort(ids, Comparator.naturalOrder());
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, redudentBefore, ids, info);
info[i] = TxnInfo.create(ids[i], rs.pick(InternalStatus.values()), ids[i], CommandsForKey.NO_TXNIDS);
Arrays.sort(info, Comparator.naturalOrder());
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, CommandsForKey.NO_PENDING_UNMANAGED);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
@ -479,10 +484,10 @@ public class CommandsForKeySerializerTest
long tokenValue = -2311778975040348869L;
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(Murmur3Partitioner.LongToken.keyForToken(tokenValue));
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,
TxnId.fromValues(0,0,0,0),
new TxnId[] {TxnId.fromValues(11,34052499,2,1)},
new CommandsForKey.Info[] { InternalStatus.PREACCEPTED.asNoInfo});
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED, txnId, CommandsForKey.NO_TXNIDS) },
CommandsForKey.NO_PENDING_UNMANAGED);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);

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

@ -23,6 +23,8 @@ import org.junit.Test;
import accord.local.Command;
import accord.primitives.Deps;
import accord.primitives.Routable;
import accord.primitives.TxnId;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.SimpleBitSet;
@ -51,19 +53,20 @@ public class WaitingOnSerializerTest
{
DataOutputBuffer buffer = new DataOutputBuffer();
qt().forAll(waitingOnGen()).check(waitingOn -> {
TxnId txnId = TxnId.NONE;
if (waitingOn.appliedOrInvalidated != null) txnId = new TxnId(txnId.epoch(), txnId.hlc(), txnId.kind(), Routable.Domain.Range, txnId.node);
buffer.clear();
long expectedSize = WaitingOnSerializer.serializedSize(waitingOn);
WaitingOnSerializer.serialize(waitingOn, buffer);
WaitingOnSerializer.serialize(txnId, waitingOn, buffer);
Assertions.assertThat(buffer.getLength()).isEqualTo(expectedSize);
Command.WaitingOn read = WaitingOnSerializer.deserialize(waitingOn.deps, new DataInputBuffer(buffer.unsafeGetBufferAndFlip(), false));
Command.WaitingOn read = WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.txnIds, new DataInputBuffer(buffer.unsafeGetBufferAndFlip(), false));
Assertions.assertThat(read)
.isEqualTo(waitingOn)
.isEqualTo(WaitingOnSerializer.deserialize(waitingOn.deps, WaitingOnSerializer.serialize(waitingOn)));
.isEqualTo(WaitingOnSerializer.deserialize(txnId, waitingOn.keys, waitingOn.txnIds, WaitingOnSerializer.serialize(txnId, waitingOn)));
});
}
private enum WaitingOnSets
{COMMIT, APPLY, APPLYED_OR_INVALIDATED}
private enum WaitingOnSets { APPLY, APPLIED_OR_INVALIDATED }
private static Gen<Command.WaitingOn> waitingOnGen()
{
@ -73,22 +76,20 @@ public class WaitingOnSerializerTest
return rs -> {
Deps deps = depsGen.next(rs);
if (deps.isEmpty()) return Command.WaitingOn.EMPTY;
int[] selected = Gens.arrays(Gens.ints().between(0, deps.txnIdCount() - 1)).unique().ofSizeBetween(0, deps.txnIdCount() - 1).next(rs);
SimpleBitSet waitingOnCommit = new SimpleBitSet(deps.txnIdCount(), false);
SimpleBitSet waitingOnApply = new SimpleBitSet(deps.txnIdCount(), false);
SimpleBitSet appliedOrInvalidated = new SimpleBitSet(deps.txnIdCount(), false);
int txnIdCount = deps.rangeDeps.txnIdCount();
int keyCount = deps.keyDeps.keys().size();
int[] selected = Gens.arrays(Gens.ints().between(0, txnIdCount + keyCount - 1)).unique().ofSizeBetween(0, txnIdCount + keyCount).next(rs);
SimpleBitSet waitingOn = new SimpleBitSet(txnIdCount + keyCount, false);
SimpleBitSet appliedOrInvalidated = rs.nextBoolean() ? null : new SimpleBitSet(txnIdCount, false);
for (int i : selected)
{
WaitingOnSets set = sets.next(rs);
WaitingOnSets set = appliedOrInvalidated == null || i >= txnIdCount ? WaitingOnSets.APPLY : sets.next(rs);
switch (set)
{
case COMMIT:
waitingOnCommit.set(i);
break;
case APPLY:
waitingOnApply.set(i);
waitingOn.set(i);
break;
case APPLYED_OR_INVALIDATED:
case APPLIED_OR_INVALIDATED:
appliedOrInvalidated.set(i);
break;
default:
@ -96,7 +97,7 @@ public class WaitingOnSerializerTest
}
}
return new Command.WaitingOn(deps, Utils.ensureImmutable(waitingOnCommit), Utils.ensureImmutable(waitingOnApply), Utils.ensureImmutable(appliedOrInvalidated));
return new Command.WaitingOn(deps.keyDeps.keys(), deps.rangeDeps.txnIds(), Utils.ensureImmutable(waitingOn), Utils.ensureImmutable(appliedOrInvalidated));
};
}
}