You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/09/16 10:32:53 UTC
(cassandra) branch cep-15-accord updated: Redesign progress mechanisms to be memory efficient, use fewer messages and to resolve dependency chains promptly.
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new f270effe34 Redesign progress mechanisms to be memory efficient, use fewer messages and to resolve dependency chains promptly.
f270effe34 is described below
commit f270effe34ea7c0f30ad1d9c864eed6fc7aa43b3
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Aug 22 21:12:48 2024 +0100
Redesign progress mechanisms to be memory efficient, use fewer messages and to resolve dependency chains promptly.
The SimpleProgressLog had a number of problems:
1. It polled for progress with no attempt to determine whether progress could realistically be made, so:
- as the number of pending transactions grew, the proportion of useful work dropped (as many would be unable to make progress without earlier transactions completing)
- each transaction in the chain could recover only on average 1/2 poll interval behind the last transaction to complete
2. It requested full transaction state from every replica on each attempt
3. It maintained a lot of in-memory state
4. Polling happened en-masse, allowing for little per-transaction control
We also separately maintained fairly expensive per-command listener state that negatively affected our command loading and caching.
The new DefaultProgressLog makes use of several new features: LocalListeners, RemoteListeners, Timers and Await messages.
- LocalListeners provide a memory-efficient collection for managing each CommandStore<E2><80><99>s transaction listeners, with dedicated record keeping for inter-transaction relationships.
- RemoteListeners provide a mechanism for request/response pairs that may be separated by longer than the normal Cassandra message timeout, and require minimal state on sender and recipient. This permits replicas to cheaply update their local state machine as soon as distributed information becomes available.
The DefaultProgressLog tracks each transaction with separate timers to handle per-transaction scheduling, backoff etc, and a succinct state machine. To reduce overhead correspondence is preferentially limited to a handful of replicas, and limited to the home shard where appropriate.
patch by Benedict; reviewed by Ariel Weisberg for CASSANDRA-19870
---
conf/cassandra.yaml | 4 +-
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 2 +-
.../db/compaction/CompactionIterator.java | 4 +
.../apache/cassandra/metrics/AccordMetrics.java | 8 +-
src/java/org/apache/cassandra/net/Verb.java | 21 ++--
.../service/accord/AccordCachingState.java | 34 ------
.../service/accord/AccordCommandStore.java | 11 +-
.../service/accord/AccordCommandStores.java | 11 +-
.../service/accord/AccordConfiguration.java | 10 +-
.../cassandra/service/accord/AccordJournal.java | 37 ------
.../cassandra/service/accord/AccordKeyspace.java | 43 ++++---
.../service/accord/AccordMessageSink.java | 11 +-
.../service/accord/AccordObjectSizes.java | 14 +--
.../service/accord/AccordSafeCommand.java | 23 ----
.../cassandra/service/accord/AccordService.java | 34 ++----
.../cassandra/service/accord/AccordStateCache.java | 17 +--
.../service/accord/AccordSyncPropagator.java | 8 +-
.../service/accord/CommandsForRangesLoader.java | 2 +-
.../cassandra/service/accord/SavedCommand.java | 32 +----
.../cassandra/service/accord/api/AccordAgent.java | 89 ++++++++++++++
.../service/accord/api/AccordScheduler.java | 6 +
.../service/accord/interop/AccordInteropApply.java | 35 +++---
.../accord/interop/AccordInteropCommit.java | 6 +-
.../accord/interop/AccordInteropExecution.java | 4 +-
.../accord/interop/AccordInteropPersist.java | 4 +-
.../accord/interop/AccordInteropReadRepair.java | 2 +-
.../accord/serializers/AcceptSerializers.java | 6 +-
.../accord/serializers/ApplySerializers.java | 9 +-
.../accord/serializers/AwaitSerializer.java | 105 +++++++++++++++++
.../serializers/BeginInvalidationSerializers.java | 13 ++-
...ializers.java => CalculateDepsSerializers.java} | 28 ++---
.../serializers/CommandsForKeySerializer.java | 33 ++++--
.../accord/serializers/CommitSerializers.java | 8 +-
.../accord/serializers/FetchSerializers.java | 102 ----------------
.../serializers/GetEphmrlReadDepsSerializers.java | 4 +-
.../serializers/GetMaxConflictSerializers.java | 4 +-
.../serializers/InformDurableSerializers.java | 10 +-
.../serializers/InformHomeDurableSerializers.java | 60 ----------
.../serializers/InformOfTxnIdSerializers.java | 53 ---------
.../accord/serializers/ListenerSerializers.java | 113 ------------------
.../accord/serializers/PreacceptSerializers.java | 10 +-
.../accord/serializers/RecoverySerializers.java | 4 +-
.../accord/serializers/TxnRequestSerializer.java | 23 ++--
.../accord/serializers/WaitOnCommitSerializer.java | 77 ------------
.../apache/cassandra/triggers/TriggerExecutor.java | 5 -
test/conf/cassandra.yaml | 1 -
.../cassandra/distributed/impl/Instance.java | 2 +-
.../cassandra/distributed/impl/InstanceConfig.java | 2 +-
.../distributed/test/QueriesTableTest.java | 4 +-
.../test/accord/AccordIncrementalRepairTest.java | 6 +-
.../test/accord/AccordIntegrationTest.java | 4 +-
.../test/accord/AccordMigrationRaceTestBase.java | 4 +-
.../test/accord/AccordProgressLogTest.java | 130 +++++++++++++++++++++
.../distributed/test/accord/AccordTestBase.java | 8 +-
.../compaction/CompactionAccordIteratorsTest.java | 23 ++--
.../apache/cassandra/dht/AccordSplitterTest.java | 4 +-
.../cassandra/service/accord/AccordAgentTest.java | 70 +++++++++++
.../service/accord/AccordCommandStoreTest.java | 1 -
.../service/accord/AccordCommandTest.java | 13 +--
.../service/accord/AccordMessageSinkTest.java | 13 ---
.../service/accord/AccordServiceTest.java | 6 +-
.../cassandra/service/accord/AccordTestUtils.java | 38 ++----
.../cassandra/service/accord/EpochSyncTest.java | 6 +-
.../cassandra/service/accord/MockJournal.java | 14 +--
.../accord/SimulatedAccordCommandStore.java | 12 +-
.../service/accord/async/AsyncOperationTest.java | 16 +--
.../serializers/CommandsForKeySerializerTest.java | 16 +--
.../apache/cassandra/utils/AccordGenerators.java | 3 +-
69 files changed, 676 insertions(+), 861 deletions(-)
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index c9d44b5a6e..307c9ebeab 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -2249,8 +2249,8 @@ storage_compatibility_mode: NONE
# # The number of Accord shards on this node; -1 means use the number of cores
# shard_count: -1
#
-# # Progress log scheduling delay
-# progress_log_schedule_delay: 1s
+# # Recover delay: the time between a transaction being initiated and a remote replica being willing to interrupt it to complete it
+# recover_delay: 1s
#
# # how quickly the fast path is reconfigured when nodes go up/down
# fast_path_update_delay: 5s
diff --git a/modules/accord b/modules/accord
index a171322f41..fb3efe9b8a 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit a171322f417c117733ca5b514d03a5202b1ac202
+Subproject commit fb3efe9b8a87f0a182545791a2b0563690d52d00
diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java
index 2e0d614957..b4d25d6689 100644
--- a/src/java/org/apache/cassandra/config/AccordSpec.java
+++ b/src/java/org/apache/cassandra/config/AccordSpec.java
@@ -30,7 +30,7 @@ public class AccordSpec
public volatile OptionaldPositiveInt shard_count = OptionaldPositiveInt.UNDEFINED;
- public volatile DurationSpec.IntMillisecondsBound progress_log_schedule_delay = new DurationSpec.IntMillisecondsBound(100);
+ public volatile DurationSpec.IntMillisecondsBound recover_delay = new DurationSpec.IntMillisecondsBound(1000);
/**
* When a barrier transaction is requested how many times to repeat attempting the barrier before giving up
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 4dbeae7896..3600868b6c 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -96,6 +96,7 @@ import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
+import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.invalidated;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.maybeDropTruncatedCommandColumns;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.truncatedApply;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeysAccessor;
@@ -849,6 +850,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
// We can still encounter sliced command state just because compaction inputs are random
return BTreeRow.emptyDeletedRow(row.clustering(), new Row.Deletion(DeletionTime.build(row.primaryKeyLivenessInfo().timestamp(), nowInSec), false));
+ case INVALIDATE:
+ return invalidated(cleanup.appliesIfNot, row, nowInSec);
+
case TRUNCATE_WITH_OUTCOME:
case TRUNCATE:
if (saveStatus.compareTo(cleanup.appliesIfNot) >= 0)
diff --git a/src/java/org/apache/cassandra/metrics/AccordMetrics.java b/src/java/org/apache/cassandra/metrics/AccordMetrics.java
index 765fe154ee..c9d8e29687 100644
--- a/src/java/org/apache/cassandra/metrics/AccordMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/AccordMetrics.java
@@ -206,7 +206,7 @@ public class AccordMetrics
@Override
public void onStable(Command cmd)
{
- long now = AccordService.uniqueNow();
+ long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@@ -218,7 +218,7 @@ public class AccordMetrics
@Override
public void onExecuted(Command cmd)
{
- long now = AccordService.uniqueNow();
+ long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@@ -232,7 +232,7 @@ public class AccordMetrics
@Override
public void onApplied(Command cmd, long applyStartTimestamp)
{
- long now = AccordService.uniqueNow();
+ long now = AccordService.now();
AccordMetrics metrics = forTransaction(cmd.txnId());
if (metrics != null)
{
@@ -270,7 +270,7 @@ public class AccordMetrics
AccordMetrics metrics = forTransaction(txnId);
if (metrics != null)
{
- long now = AccordService.uniqueNow();
+ long now = AccordService.now();
metrics.recoveryDuration.update(now - recoveryTimestamp.hlc(), MICROSECONDS);
metrics.recoveryDelay.update(recoveryTimestamp.hlc() - txnId.hlc(), MICROSECONDS);
diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java
index c4d87139ea..b278d78e08 100644
--- a/src/java/org/apache/cassandra/net/Verb.java
+++ b/src/java/org/apache/cassandra/net/Verb.java
@@ -92,18 +92,16 @@ import org.apache.cassandra.service.accord.serializers.CheckStatusSerializers;
import org.apache.cassandra.service.accord.serializers.CommitSerializers;
import org.apache.cassandra.service.accord.serializers.EnumSerializer;
import org.apache.cassandra.service.accord.serializers.FetchSerializers;
-import org.apache.cassandra.service.accord.serializers.GetDepsSerializers;
+import org.apache.cassandra.service.accord.serializers.CalculateDepsSerializers;
import org.apache.cassandra.service.accord.serializers.GetEphmrlReadDepsSerializers;
import org.apache.cassandra.service.accord.serializers.GetMaxConflictSerializers;
import org.apache.cassandra.service.accord.serializers.InformDurableSerializers;
-import org.apache.cassandra.service.accord.serializers.InformHomeDurableSerializers;
-import org.apache.cassandra.service.accord.serializers.InformOfTxnIdSerializers;
import org.apache.cassandra.service.accord.serializers.PreacceptSerializers;
import org.apache.cassandra.service.accord.serializers.QueryDurableBeforeSerializers;
import org.apache.cassandra.service.accord.serializers.ReadDataSerializers;
import org.apache.cassandra.service.accord.serializers.RecoverySerializers;
import org.apache.cassandra.service.accord.serializers.SetDurableSerializers;
-import org.apache.cassandra.service.accord.serializers.WaitOnCommitSerializer;
+import org.apache.cassandra.service.accord.serializers.AwaitSerializer;
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState;
import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.ConsensusKeyMigrationFinished;
import org.apache.cassandra.service.paxos.Commit;
@@ -319,21 +317,20 @@ public enum Verb
ACCORD_COMMIT_REQ (127, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_COMMIT_INVALIDATE_REQ (128, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.invalidate, AccordService::verbHandlerOrNoop ),
ACCORD_APPLY_RSP (129, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.reply, RESPONSE_HANDLER ),
- ACCORD_APPLY_REQ (130, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_APPLY_RSP ),
+ ACCORD_APPLY_REQ (130, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_APPLY_RSP ),
ACCORD_BEGIN_RECOVER_RSP (131, P2, writeTimeout, IMMEDIATE, () -> RecoverySerializers.reply, RESPONSE_HANDLER ),
ACCORD_BEGIN_RECOVER_REQ (132, P2, writeTimeout, IMMEDIATE, () -> RecoverySerializers.request, AccordService::verbHandlerOrNoop, ACCORD_BEGIN_RECOVER_RSP ),
ACCORD_BEGIN_INVALIDATE_RSP (133, P2, writeTimeout, IMMEDIATE, () -> BeginInvalidationSerializers.reply, RESPONSE_HANDLER ),
ACCORD_BEGIN_INVALIDATE_REQ (134, P2, writeTimeout, IMMEDIATE, () -> BeginInvalidationSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_BEGIN_INVALIDATE_RSP ),
- ACCORD_WAIT_ON_COMMIT_RSP (136, P2, writeTimeout, IMMEDIATE, () -> WaitOnCommitSerializer.reply, RESPONSE_HANDLER ),
- ACCORD_WAIT_ON_COMMIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> WaitOnCommitSerializer.request, AccordService::verbHandlerOrNoop, ACCORD_WAIT_ON_COMMIT_RSP ),
- ACCORD_WAIT_UNTIL_APPLIED_REQ (137, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
- ACCORD_INFORM_OF_TXN_REQ (138, P2, writeTimeout, IMMEDIATE, () -> InformOfTxnIdSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
- ACCORD_INFORM_HOME_DURABLE_REQ (139, P2, writeTimeout, IMMEDIATE, () -> InformHomeDurableSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
+ ACCORD_AWAIT_RSP (136, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.syncReply, RESPONSE_HANDLER ),
+ ACCORD_AWAIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.request, AccordService::verbHandlerOrNoop, ACCORD_AWAIT_RSP ),
+ ACCORD_AWAIT_ASYNC_RSP_REQ (137, P2, writeTimeout, IMMEDIATE, () -> AwaitSerializer.asyncReply, AccordService::verbHandlerOrNoop ),
+ ACCORD_WAIT_UNTIL_APPLIED_REQ (138, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.waitUntilApplied, AccordService::verbHandlerOrNoop, ACCORD_READ_RSP ),
ACCORD_INFORM_DURABLE_REQ (140, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.reply, RESPONSE_HANDLER ),
ACCORD_CHECK_STATUS_REQ (142, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ),
- ACCORD_GET_DEPS_RSP (143, P2, writeTimeout, IMMEDIATE, () -> GetDepsSerializers.reply, RESPONSE_HANDLER ),
- ACCORD_GET_DEPS_REQ (144, P2, writeTimeout, IMMEDIATE, () -> GetDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_GET_DEPS_RSP ),
+ ACCORD_CALCULATE_DEPS_RSP (143, P2, writeTimeout, IMMEDIATE, () -> CalculateDepsSerializers.reply, RESPONSE_HANDLER ),
+ ACCORD_CALCULATE_DEPS_REQ (144, P2, writeTimeout, IMMEDIATE, () -> CalculateDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_CALCULATE_DEPS_RSP),
ACCORD_GET_EPHMRL_READ_DEPS_RSP (161, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_EPHMRL_READ_DEPS_REQ (162, P2, writeTimeout, IMMEDIATE, () -> GetEphmrlReadDepsSerializers.request, AccordService::verbHandlerOrNoop, ACCORD_GET_EPHMRL_READ_DEPS_RSP),
ACCORD_GET_MAX_CONFLICT_RSP (163, P2, writeTimeout, IMMEDIATE, () -> GetMaxConflictSerializers.reply, RESPONSE_HANDLER ),
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCachingState.java b/src/java/org/apache/cassandra/service/accord/AccordCachingState.java
index 304befe83a..35ceec450d 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCachingState.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCachingState.java
@@ -25,8 +25,6 @@ import java.util.function.ToLongFunction;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
-import accord.local.Command.TransientListener;
-import accord.local.Listeners;
import accord.utils.IntrusiveLinkedListNode;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@@ -69,11 +67,6 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
final byte index;
private boolean shouldUpdateSize;
- /**
- * Transient listeners aren't meant to survive process restart, but must survive cache eviction.
- */
- private Listeners<TransientListener> transientListeners;
-
AccordCachingState(K key, int index)
{
this.key = key;
@@ -157,33 +150,6 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
return complete().status();
}
- public void addListener(TransientListener listener)
- {
- if (transientListeners == null)
- transientListeners = new Listeners<>();
- transientListeners.add(listener);
- }
-
- public boolean removeListener(TransientListener listener)
- {
- return transientListeners != null && transientListeners.remove(listener);
- }
-
- public void listeners(Listeners<TransientListener> listeners)
- {
- transientListeners = listeners;
- }
-
- public Listeners<TransientListener> listeners()
- {
- return transientListeners == null ? Listeners.EMPTY : transientListeners;
- }
-
- public boolean hasListeners()
- {
- return !listeners().isEmpty();
- }
-
State<K, V> complete()
{
return state.isCompleteable() ? state(state.complete()) : state;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 39370a1cfc..c1ce231da9 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.DataStore;
+import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
@@ -114,11 +115,12 @@ public class AccordCommandStore extends CommandStore implements CacheSize
Agent agent,
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
+ LocalListeners.Factory listenerFactory,
EpochUpdateHolder epochUpdateHolder,
IJournal journal,
AccordStateCacheMetrics cacheMetrics)
{
- this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics);
+ this(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics);
}
private static <K, V> void registerJfrListener(int id, AccordStateCache.Instance<K, V, ?> instance, String name)
@@ -194,13 +196,14 @@ public class AccordCommandStore extends CommandStore implements CacheSize
Agent agent,
DataStore dataStore,
ProgressLog.Factory progressLogFactory,
+ LocalListeners.Factory listenerFactory,
EpochUpdateHolder epochUpdateHolder,
IJournal journal,
ExecutorPlus loadExecutor,
ExecutorPlus saveExecutor,
AccordStateCacheMetrics cacheMetrics)
{
- super(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder);
+ super(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
this.journal = journal;
loggingId = String.format("[%s]", id);
executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']');
@@ -257,8 +260,8 @@ public class AccordCommandStore extends CommandStore implements CacheSize
static Factory factory(AccordJournal journal, AccordStateCacheMetrics cacheMetrics)
{
- return (id, time, agent, dataStore, progressLogFactory, rangesForEpoch) ->
- new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal, cacheMetrics);
+ return (id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch) ->
+ new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, cacheMetrics);
}
public CommandsForRangesLoader diskCommandsForRanges()
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
index 7328a31de6..60cbcf84ff 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
@@ -22,6 +22,7 @@ import java.util.function.Supplier;
import accord.api.Agent;
import accord.api.ConfigurationService.EpochReady;
import accord.api.DataStore;
+import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.local.CommandStores;
import accord.local.Node;
@@ -45,17 +46,19 @@ public class AccordCommandStores extends CommandStores implements CacheSize
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
- ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, AccordJournal journal)
+ ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenerFactory,
+ AccordJournal journal)
{
- super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
+ super(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory,
+ AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
setCapacity(DatabaseDescriptor.getAccordCacheSizeInMiB() << 20);
this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this);
}
static Factory factory(AccordJournal journal)
{
- return (time, agent, store, random, shardDistributor, progressLogFactory) ->
- new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, journal);
+ return (time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory) ->
+ new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory, journal);
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfiguration.java b/src/java/org/apache/cassandra/service/accord/AccordConfiguration.java
index d87e6c96cd..a0da2d99af 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfiguration.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordConfiguration.java
@@ -18,9 +18,7 @@
package org.apache.cassandra.service.accord;
-import java.time.Duration;
-
-import accord.config.LocalConfig;
+import accord.api.LocalConfig;
import org.apache.cassandra.config.Config;
// TODO (expected): should this be merged with AccordSpec?
@@ -32,10 +30,4 @@ public class AccordConfiguration implements LocalConfig
{
this.config = config;
}
-
- @Override
- public Duration getProgressLogScheduleDelay()
- {
- return config.accord.progress_log_schedule_delay.toDuration();
- }
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index eda7155a38..932d31356a 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -35,7 +35,6 @@ import org.slf4j.LoggerFactory;
import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.Node;
-import accord.messages.LocalRequest;
import accord.messages.ReplyContext;
import accord.messages.Request;
import accord.primitives.TxnId;
@@ -175,19 +174,6 @@ public class AccordJournal implements IJournal, Shutdownable
delayedRequestProcessor.delay(requestContext);
}
- /**
- * Accord protocol messages originating from local node, e.g. Propagate.
- */
- @SuppressWarnings("rawtypes, unchecked")
- public <R> void processLocalRequest(LocalRequest request, BiConsumer<? super R, Throwable> callback)
- {
- LocalRequestContext requestContext = LocalRequestContext.create(request, callback);
- if (node.topology().hasEpoch(request.waitForEpoch()))
- request.process(node, requestContext.callback);
- else
- delayedRequestProcessor.delay(requestContext);
- }
-
@Override
public Command loadCommand(int commandStoreId, TxnId txnId)
{
@@ -281,29 +267,6 @@ public class AccordJournal implements IJournal, Shutdownable
public abstract void process(Node node, AccordEndpointMapper endpointMapper);
}
- private static class LocalRequestContext<T> extends RequestContext
- {
- private final BiConsumer<T, Throwable> callback;
- private final LocalRequest<T> request;
-
- LocalRequestContext(long waitForEpoch, LocalRequest<T> request, BiConsumer<T, Throwable> callback)
- {
- super(waitForEpoch);
- this.callback = callback;
- this.request = request;
- }
-
- public void process(Node node, AccordEndpointMapper endpointMapper)
- {
- request.process(node, callback);
- }
-
- static <R> LocalRequestContext<R> create(LocalRequest<R> request, BiConsumer<R, Throwable> callback)
- {
- return new LocalRequestContext<>(request.waitForEpoch(), request, callback);
- }
- }
-
/**
* Barebones response context not holding a reference to the entire message
*/
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index bf5d8179fa..45a85998a6 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -53,7 +53,6 @@ import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.DurableBefore;
-import accord.local.Listeners;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.SaveStatus;
@@ -146,7 +145,6 @@ import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers;
import org.apache.cassandra.service.accord.serializers.CommandsForKeySerializer;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
-import org.apache.cassandra.service.accord.serializers.ListenerSerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.utils.Clock.Global;
import org.apache.cassandra.utils.btree.BTree;
@@ -249,7 +247,6 @@ public class AccordKeyspace
public static class LocalVersionedSerializers
{
static final LocalVersionedSerializer<Route<?>> route = localSerializer(KeySerializers.route);
- static final LocalVersionedSerializer<Command.DurableAndIdempotentListener> listeners = localSerializer(ListenerSerializers.listener);
static final LocalVersionedSerializer<Topology> topology = localSerializer(TopologySerializers.topology);
static final LocalVersionedSerializer<ReducingRangeMap<Timestamp>> rejectBefore = localSerializer(CommandStoreSerializers.rejectBefore);
static final LocalVersionedSerializer<DurableBefore> durableBefore = localSerializer(CommandStoreSerializers.durableBefore);
@@ -283,6 +280,7 @@ public class AccordKeyspace
public static final ColumnMetadata execute_at = getColumn(Commands, "execute_at");
public static final ColumnMetadata[] TRUNCATE_FIELDS = new ColumnMetadata[] { durability, execute_at, route, status };
+ public static final ColumnMetadata[] INVALIDATE_FIELDS = new ColumnMetadata[] { status };
static
{
@@ -367,6 +365,31 @@ public class AccordKeyspace
return newLeaf;
}
+ public static Row invalidated(SaveStatus newSaveStatus, Row row, long nowInSec)
+ {
+ long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
+ long newTimestamp = oldTimestamp + 1;
+
+ Object[] newLeaf = invalidatedLeaf(newTimestamp, newSaveStatus);
+
+ // Including a deletion allows future compactions to drop data before it gets to the purger
+ // but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns
+ // regardless
+ Row.Deletion deletion = new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false);
+ return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), deletion, newLeaf);
+ }
+
+ private static Object[] invalidatedLeaf(long newTimestamp, SaveStatus newSaveStatus)
+ {
+ Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(INVALIDATE_FIELDS.length);
+ int colIndex = 0;
+ // Status always needs to use the new timestamp since we are replacing the existing value
+ // All the other columns are being retained unmodified with at most updated timestamps to accomdate deletion
+ //noinspection UnusedAssignment
+ newLeaf[colIndex++] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
+ return newLeaf;
+ }
+
public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, Cell<?> durabilityCell, Cell<?> executeAtCell, Cell<?> routeCell, boolean withOutcome)
{
checkArgument(durabilityCell.column() == CommandsColumns.durability);
@@ -623,7 +646,7 @@ public class AccordKeyspace
if (current == null)
return null;
- CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore);
+ CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore.shardRedundantBefore());
if (current == updated)
return row;
@@ -1137,18 +1160,6 @@ public class AccordKeyspace
}
}
- private static Listeners.Immutable deserializeListeners(UntypedResultSet.Row row) throws IOException
- {
- Set<ByteBuffer> serialized = row.getSet("listeners", BytesType.instance);
- if (serialized == null || serialized.isEmpty())
- return Listeners.Immutable.EMPTY;
-
- Listeners<Command.DurableAndIdempotentListener> result = new Listeners<>();
- for (ByteBuffer bytes : serialized)
- result.add(deserialize(bytes, LocalVersionedSerializers.listeners));
- return new Listeners.Immutable(result);
- }
-
public static PartitionKey deserializeKey(ByteBuffer buffer)
{
List<ByteBuffer> split = KEY_TYPE.unpack(buffer, ByteBufferAccessor.instance);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
index 5a514219e3..bf6c5ae06f 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
@@ -122,8 +122,8 @@ public class AccordMessageSink implements MessageSink
builder.put(MessageType.ACCEPT_REQ, Verb.ACCORD_ACCEPT_REQ);
builder.put(MessageType.ACCEPT_RSP, Verb.ACCORD_ACCEPT_RSP);
builder.put(MessageType.ACCEPT_INVALIDATE_REQ, Verb.ACCORD_ACCEPT_INVALIDATE_REQ);
- builder.put(MessageType.GET_DEPS_REQ, Verb.ACCORD_GET_DEPS_REQ);
- builder.put(MessageType.GET_DEPS_RSP, Verb.ACCORD_GET_DEPS_RSP);
+ builder.put(MessageType.CALCULATE_DEPS_REQ, Verb.ACCORD_CALCULATE_DEPS_REQ);
+ builder.put(MessageType.CALCULATE_DEPS_RSP, Verb.ACCORD_CALCULATE_DEPS_RSP);
builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_REQ, Verb.ACCORD_GET_EPHMRL_READ_DEPS_REQ);
builder.put(MessageType.GET_EPHEMERAL_READ_DEPS_RSP, Verb.ACCORD_GET_EPHMRL_READ_DEPS_RSP);
builder.put(MessageType.GET_MAX_CONFLICT_REQ, Verb.ACCORD_GET_MAX_CONFLICT_REQ);
@@ -144,13 +144,12 @@ public class AccordMessageSink implements MessageSink
builder.put(MessageType.BEGIN_RECOVER_RSP, Verb.ACCORD_BEGIN_RECOVER_RSP);
builder.put(MessageType.BEGIN_INVALIDATE_REQ, Verb.ACCORD_BEGIN_INVALIDATE_REQ);
builder.put(MessageType.BEGIN_INVALIDATE_RSP, Verb.ACCORD_BEGIN_INVALIDATE_RSP);
- builder.put(MessageType.WAIT_ON_COMMIT_REQ, Verb.ACCORD_WAIT_ON_COMMIT_REQ);
- builder.put(MessageType.WAIT_ON_COMMIT_RSP, Verb.ACCORD_WAIT_ON_COMMIT_RSP);
+ builder.put(MessageType.AWAIT_REQ, Verb.ACCORD_AWAIT_REQ);
+ builder.put(MessageType.AWAIT_RSP, Verb.ACCORD_AWAIT_RSP);
+ builder.put(MessageType.ASYNC_AWAIT_COMPLETE_REQ, Verb.ACCORD_AWAIT_ASYNC_RSP_REQ);
builder.put(MessageType.WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_WAIT_UNTIL_APPLIED_REQ);
builder.put(MessageType.APPLY_THEN_WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_APPLY_AND_WAIT_REQ);
- builder.put(MessageType.INFORM_OF_TXN_REQ, Verb.ACCORD_INFORM_OF_TXN_REQ);
builder.put(MessageType.INFORM_DURABLE_REQ, Verb.ACCORD_INFORM_DURABLE_REQ);
- builder.put(MessageType.INFORM_HOME_DURABLE_REQ, Verb.ACCORD_INFORM_HOME_DURABLE_REQ);
builder.put(MessageType.CHECK_STATUS_REQ, Verb.ACCORD_CHECK_STATUS_REQ);
builder.put(MessageType.CHECK_STATUS_RSP, Verb.ACCORD_CHECK_STATUS_RSP);
builder.put(MessageType.FETCH_DATA_REQ, Verb.ACCORD_FETCH_DATA_REQ);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
index 4262749991..53fd581551 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
@@ -69,7 +69,7 @@ import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.apache.cassandra.utils.ObjectSizes;
-import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
+import static accord.primitives.TxnId.NO_TXNIDS;
import static org.apache.cassandra.utils.ObjectSizes.measure;
public class AccordObjectSizes
@@ -266,14 +266,6 @@ public class AccordObjectSizes
return ((TxnResult) result).estimatedSizeOnHeap();
}
- private static final long EMPTY_COMMAND_LISTENER = measure(new Command.ProxyListener(null));
- public static long listener(Command.DurableAndIdempotentListener listener)
- {
- if (listener instanceof Command.ProxyListener)
- return EMPTY_COMMAND_LISTENER + timestamp(((Command.ProxyListener) listener).txnId());
- throw new IllegalArgumentException("Unhandled listener type: " + listener.getClass());
- }
-
private static class CommandEmptySizes
{
private final static TokenKey EMPTY_KEY = new TokenKey(EMPTY_ID, null);
@@ -301,7 +293,7 @@ public class AccordObjectSizes
final static long COMMITTED = measure(Command.SerializerSupport.committed(attrs(true, true), SaveStatus.Committed, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, null));
final static long EXECUTED = measure(Command.SerializerSupport.executed(attrs(true, true), SaveStatus.Applied, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO, WaitingOn.empty(EMPTY_TXNID.domain()), EMPTY_WRITES, EMPTY_RESULT));
final static long TRUNCATED = measure(Command.SerializerSupport.truncatedApply(attrs(false, false), SaveStatus.TruncatedApply, EMPTY_TXNID, null, null));
- final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID, null));
+ final static long INVALIDATED = measure(Command.SerializerSupport.invalidated(EMPTY_TXNID));
private static long emptySize(Command command)
{
@@ -343,8 +335,6 @@ public class AccordObjectSizes
long size = CommandEmptySizes.emptySize(command);
size += sizeNullable(command.route(), AccordObjectSizes::route);
size += sizeNullable(command.promised(), AccordObjectSizes::timestamp);
- for (Command.DurableAndIdempotentListener listener : command.durableListeners())
- size += listener(listener);
size += sizeNullable(command.executeAt(), AccordObjectSizes::timestamp);
size += sizeNullable(command.partialTxn(), AccordObjectSizes::txn);
size += sizeNullable(command.partialDeps(), AccordObjectSizes::dependencies);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
index 28220cbf47..43bc8d54b9 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
@@ -24,8 +24,6 @@ import java.util.function.Function;
import com.google.common.annotations.VisibleForTesting;
import accord.local.Command;
-import accord.local.Command.TransientListener;
-import accord.local.Listeners;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@@ -152,27 +150,6 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
return invalidated;
}
- @Override
- public void addListener(TransientListener listener)
- {
- checkNotInvalidated();
- global.addListener(listener);
- }
-
- @Override
- public boolean removeListener(TransientListener listener)
- {
- checkNotInvalidated();
- return global.removeListener(listener);
- }
-
- @Override
- public Listeners<TransientListener> transientListeners()
- {
- checkNotInvalidated();
- return global.listeners();
- }
-
public static Function<AccordCachingState<TxnId, Command>, AccordSafeCommand> safeRefFactory()
{
return Invariants.testParanoia(LINEAR, LINEAR, HIGH) ? DebugAccordSafeCommand::new : AccordSafeCommand::new;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index 6a46c24371..e59ae24a50 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -27,7 +27,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -37,7 +36,6 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.primitives.Ints;
@@ -45,8 +43,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.BarrierType;
+import accord.api.LocalConfig;
import accord.api.Result;
-import accord.config.LocalConfig;
import accord.coordinate.Barrier;
import accord.coordinate.CoordinateSyncPoint;
import accord.coordinate.CoordinationFailed;
@@ -57,10 +55,13 @@ import accord.coordinate.Timeout;
import accord.coordinate.TopologyMismatch;
import accord.impl.AbstractConfigurationService;
import accord.impl.CoordinateDurabilityScheduling;
-import accord.impl.SimpleProgressLog;
+import accord.impl.DefaultLocalListeners;
+import accord.impl.DefaultRemoteListeners;
+import accord.impl.DefaultRequestTimeouts;
import accord.impl.SizeOfIntersectionSorter;
import accord.local.Command;
import accord.local.CommandStore;
+import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.CommandStores;
import accord.local.DurableBefore;
import accord.local.KeyHistory;
@@ -73,7 +74,6 @@ import accord.local.SaveStatus;
import accord.local.ShardDistributor.EvenSplit;
import accord.local.Status;
import accord.local.cfk.CommandsForKey;
-import accord.messages.LocalRequest;
import accord.messages.Request;
import accord.primitives.Keys;
import accord.primitives.Ranges;
@@ -339,18 +339,11 @@ public class AccordService implements IAccordService, Shutdownable
return i;
}
- public static long uniqueNow()
+ public static long now()
{
- // TODO (now, correctness): This is not unique it's just currentTimeMillis as microseconds
return TimeUnit.MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
}
- public static long unix(TimeUnit timeUnit)
- {
- Preconditions.checkArgument(timeUnit != TimeUnit.NANOSECONDS, "Nanoseconds since the epoch doesn't fit in a long");
- return timeUnit.convert(Clock.Global.currentTimeMillis(), TimeUnit.MILLISECONDS);
- }
-
private AccordService(Id localId)
{
Invariants.checkState(localId != null, "static localId must be set before instantiating AccordService");
@@ -365,9 +358,8 @@ public class AccordService implements IAccordService, Shutdownable
this.journal = new AccordJournal(configService, DatabaseDescriptor.getAccord().journal);
this.node = new Node(localId,
messageSink,
- this::handleLocalRequest,
configService,
- AccordService::uniqueNow,
+ AccordService::now,
NodeTimeService.elapsedWrapperFromMonotonicSource(NANOSECONDS, Clock.Global::nanoTime),
() -> dataStore,
new KeyspaceSplitter(new EvenSplit<>(DatabaseDescriptor.getAccordShardCount(), getPartitioner().accordSplitter())),
@@ -376,7 +368,10 @@ public class AccordService implements IAccordService, Shutdownable
scheduler,
CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER,
new AccordTopologySorter.Supplier(configService, DatabaseDescriptor.getEndpointSnitch())),
- SimpleProgressLog::new,
+ DefaultRemoteListeners::new,
+ DefaultRequestTimeouts::new,
+ DefaultProgressLogs::new,
+ DefaultLocalListeners.Factory::new,
AccordCommandStores.factory(journal),
new AccordInteropFactory(agent, configService),
configuration);
@@ -754,13 +749,6 @@ public class AccordService implements IAccordService, Shutdownable
}
}
- private <R> void handleLocalRequest(LocalRequest<R> request, BiConsumer<? super R, Throwable> callback, Node node)
- {
- // currently, we only create LocalRequests that have side effects and need to be persisted
- Invariants.checkState(request.type().hasSideEffects());
- journal.processLocalRequest(request, callback);
- }
-
private static RequestTimeoutException newTimeout(TxnId txnId, boolean isWrite, ConsistencyLevel consistencyLevel)
{
// Client protocol doesn't handle null consistency level so use ANY
diff --git a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
index 5fa5c6a941..42d4a4d243 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
@@ -216,18 +216,11 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
instance.validateLoadEvicted(node);
- if (!node.hasListeners())
- {
- AccordCachingState<?, ?> self = instances.get(node.index).cache.remove(node.key());
- Invariants.checkState(self.references == 0);
- checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
- if (instance.listeners != null)
- instance.listeners.forEach(l -> l.onEvict((AccordCachingState) node));
- }
- else
- {
- node.markEvicted(); // keep the node in the cache to prevent transient listeners from being GCd
- }
+ AccordCachingState<?, ?> self = instances.get(node.index).cache.remove(node.key());
+ Invariants.checkState(self.references == 0);
+ checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
+ if (instance.listeners != null)
+ instance.listeners.forEach(l -> l.onEvict((AccordCachingState) node));
}
public ImmutableStats stats()
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
index 2c9626718d..6f1e479f91 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
@@ -113,7 +113,7 @@ public class AccordSyncPropagator
if (closed.containsAll(addClosed))
return null;
- addClosed = addClosed.subtract(closed);
+ addClosed = addClosed.without(closed);
closed = closed.with(addClosed);
return new Notification(epoch, Collections.emptySet(), addClosed, Ranges.EMPTY);
}
@@ -123,7 +123,7 @@ public class AccordSyncPropagator
if (redundant.containsAll(addRedundant))
return null;
- addRedundant = addRedundant.subtract(redundant);
+ addRedundant = addRedundant.without(redundant);
redundant = redundant.with(addRedundant);
return new Notification(epoch, Collections.emptySet(), Ranges.EMPTY, addRedundant);
}
@@ -140,8 +140,8 @@ public class AccordSyncPropagator
if (notification.syncComplete.containsAll(syncComplete)) syncComplete = ImmutableSet.of();
else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.syncComplete.contains(v)));
}
- closed = closed.subtract(notification.closed);
- redundant = redundant.subtract(notification.redundant);
+ closed = closed.without(notification.closed);
+ redundant = redundant.without(notification.redundant);
return syncComplete.isEmpty() && closed.isEmpty() && redundant.isEmpty();
}
diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
index e660b686ce..b0fcd8e37f 100644
--- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
+++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
@@ -231,7 +231,7 @@ public class CommandsForRangesLoader
accum.add(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end));
return accum;
}, new ArrayList<Range>(), ignore -> false).toArray(Range[]::new));
- Ranges newRanges = ranges.subtract(durableAlready);
+ Ranges newRanges = ranges.without(durableAlready);
if (newRanges.isEmpty())
return null;
diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
index bb1fe6da73..090e4da755 100644
--- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
@@ -29,7 +29,6 @@ import com.google.common.annotations.VisibleForTesting;
import accord.api.Result;
import accord.local.Command;
import accord.local.CommonAttributes;
-import accord.local.Listeners;
import accord.local.SaveStatus;
import accord.local.Status;
import accord.primitives.Ballot;
@@ -69,7 +68,6 @@ public class SavedCommand
ADDITIONAL_KEYS,
WAITING_ON,
WRITES,
- LISTENERS
}
public interface Writer<K> extends Journal.Writer
@@ -179,13 +177,6 @@ public class SavedCommand
if (getFieldChanged(Fields.WRITES, flags) && after.writes() != null)
CommandSerializers.writes.serialize(after.writes(), out, userVersion);
-
- if (getFieldChanged(Fields.LISTENERS, flags) && after.durableListeners() != null)
- {
- out.writeByte(after.durableListeners().size());
- for (Command.DurableAndIdempotentListener listener : after.durableListeners())
- AccordKeyspace.LocalVersionedSerializers.listeners.serialize(listener, out);
- }
}
@VisibleForTesting
@@ -210,7 +201,6 @@ public class SavedCommand
flags = collectFlags(before, after, SavedCommand::getWaitingOn, false, Fields.WAITING_ON, flags);
flags = collectFlags(before, after, Command::writes, false, Fields.WRITES, flags);
- flags = collectFlags(before, after, c -> c.durableListeners().isEmpty() ? null : c.durableListeners(), true, Fields.LISTENERS, flags);
return flags;
}
@@ -289,7 +279,6 @@ public class SavedCommand
SavedCommand.WaitingOnProvider waitingOn = (txn, deps) -> null;
Writes writes = null;
- Listeners.Immutable<?> listeners = null;
Result result = CommandSerializers.APPLIED;
boolean nextCalled = false;
@@ -427,22 +416,6 @@ public class SavedCommand
else
writes = CommandSerializers.writes.deserialize(in, userVersion);
}
-
- if (getFieldChanged(Fields.LISTENERS, flags))
- {
- if (getFieldIsNull(Fields.LISTENERS, flags))
- {
- listeners = null;
- }
- else
- {
- Listeners builder = Listeners.Immutable.EMPTY.mutable();
- int cnt = in.readByte();
- for (int i = 0; i < cnt; i++)
- builder.add(AccordKeyspace.LocalVersionedSerializers.listeners.deserialize(in));
- listeners = new Listeners.Immutable(builder);
- }
- }
}
public void forceResult(Result newValue)
@@ -469,8 +442,6 @@ public class SavedCommand
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
- if (listeners != null && !listeners.isEmpty())
- attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (this.waitingOn != null)
@@ -518,7 +489,7 @@ public class SavedCommand
case Erased:
return Command.Truncated.erased(attrs.txnId(), attrs.durability(), attrs.route());
case Invalidated:
- return Command.Truncated.invalidated(attrs.txnId(), attrs.durableListeners());
+ return Command.Truncated.invalidated(attrs.txnId());
}
}
@@ -537,7 +508,6 @@ public class SavedCommand
", additionalKeysOrRanges=" + additionalKeysOrRanges +
", waitingOn=" + waitingOn +
", writes=" + writes +
- ", listeners=" + listeners +
'}';
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
index 7caeead9b1..a386bd0476 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
@@ -21,29 +21,45 @@ package org.apache.cassandra.service.accord.api;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
+import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.EventsListener;
+import accord.api.ProgressLog.BlockedUntil;
import accord.api.Result;
+import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.Node;
+import accord.local.SafeCommand;
+import accord.local.SafeCommandStore;
+import accord.messages.ReplyContext;
import accord.primitives.Ranges;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind;
import accord.primitives.TxnId;
+import accord.topology.Shard;
+import accord.utils.DefaultRandom;
+import accord.utils.Invariants;
+import accord.utils.RandomSource;
+import accord.utils.SortedList;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.metrics.AccordMetrics;
+import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static accord.primitives.Routable.Domain.Key;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getReadRpcTimeout;
import static org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.maybeSaveAccordKeyMigrationLocally;
@@ -55,6 +71,7 @@ public class AccordAgent implements Agent
// TODO (required): this should be configurable and have exponential back-off, escaping to operator input past a certain number of retries
private long retryBootstrapDelayMicros = SECONDS.toMicros(1L);
+ private final RandomSource random = new DefaultRandom();
public void setRetryBootstrapDelay(long delay, TimeUnit units)
{
@@ -147,4 +164,76 @@ public class AccordAgent implements Agent
{
return AccordMetrics.Listener.instance;
}
+
+ @Override
+ public long replyTimeout(ReplyContext replyContext, TimeUnit units)
+ {
+ return Math.max(1, units.convert(((ResponseContext)replyContext).expiresAtNanos() - Clock.Global.nanoTime(), NANOSECONDS));
+ }
+
+ @Override
+ public long attemptCoordinationDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int retryCount)
+ {
+ SafeCommand safeCommand = safeStore.ifInitialised(txnId);
+ Invariants.nonNull(safeCommand);
+
+ Command command = safeCommand.current();
+ Invariants.nonNull(command);
+
+ Timestamp mostRecentAttempt = Timestamp.max(command.txnId(), command.promised());
+ RoutingKey homeKey = command.route().homeKey();
+ Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch());
+
+ // TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
+ long oneSecond = SECONDS.toMicros(1L);
+ long startTime = mostRecentAttempt.hlc() + DatabaseDescriptor.getAccord().recover_delay.to(MICROSECONDS)
+ + (retryCount == 0 ? 0 : random.nextLong(oneSecond << Math.min(retryCount, 4)));
+
+ startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), oneSecond, random);
+ long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
+ return units.convert(Math.max(1, startTime - nowMicros), MICROSECONDS);
+ }
+
+ @VisibleForTesting
+ public static long nonClashingStartTime(long startTime, SortedList<Node.Id> nodes, Node.Id id, long granularity, RandomSource random)
+ {
+ long perSecondStartTime;
+ if (nodes != null)
+ {
+ int position = nodes.indexOf(id);
+ perSecondStartTime = position * (SECONDS.toMicros(1) / nodes.size());
+ }
+ else
+ {
+ // we've raced with topology update, this should be rare so just pick a random start time
+ perSecondStartTime = random.nextLong(granularity);
+ }
+
+ // TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
+ long subSecondRemainder = startTime % granularity;
+ long newStartTime = startTime - subSecondRemainder + perSecondStartTime;
+ if (newStartTime < startTime)
+ newStartTime += granularity;
+ return newStartTime;
+ }
+
+ @Override
+ public long seekProgressDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil blockedUntil, TimeUnit units)
+ {
+ // TODO (required): make this configurable and dependent upon normal request latencies, and perhaps offset from txnId.hlc()
+ if (retryCount == 0)
+ return units.convert(100, MILLISECONDS);
+
+ return units.convert((1L << Math.min(retryCount, 4)), SECONDS);
+ }
+
+ @Override
+ public long retryAwaitTimeout(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, BlockedUntil retrying, TimeUnit units)
+ {
+ // TODO (expected): integrate with contention backoff
+ if (retryCount == 0)
+ return units.convert(100, MILLISECONDS);
+
+ return units.convert((1L << Math.min(retryCount, 4)), SECONDS);
+ }
}
diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java b/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java
index df83a92932..a616cf05d4 100644
--- a/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java
+++ b/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java
@@ -46,6 +46,12 @@ public class AccordScheduler implements Scheduler, Shutdownable
{
future.cancel(false);
}
+
+ @Override
+ public boolean isDone()
+ {
+ return future.isDone();
+ }
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
index 49c06e811c..8e3758a381 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java
@@ -21,10 +21,10 @@ package org.apache.cassandra.service.accord.interop;
import java.util.BitSet;
import javax.annotation.Nullable;
+import accord.api.LocalListeners;
import accord.api.Result;
import accord.local.Command;
import accord.local.Node.Id;
-import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.Status;
@@ -34,8 +34,8 @@ import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -47,9 +47,9 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType;
import org.apache.cassandra.service.accord.serializers.ApplySerializers.ApplySerializer;
import org.apache.cassandra.service.accord.txn.AccordUpdate;
+import org.jctools.queues.MpscChunkedArrayQueue;
import static accord.utils.Invariants.checkState;
-import static accord.utils.MapReduceConsume.forEach;
import static com.google.common.base.Preconditions.checkArgument;
/**
@@ -58,7 +58,7 @@ import static com.google.common.base.Preconditions.checkArgument;
* // and these all are a bit copy pasta in terms of managing things like waiting on, obsoletion, cancellation/listeners, insufficient etc. and it would be less fragile
* // in the long run to not duplicate these kind of difficult to get right mechanism and have a single pluggable framework to request each specific behavior
*/
-public class AccordInteropApply extends Apply implements Command.TransientListener
+public class AccordInteropApply extends Apply implements LocalListeners.ComplexListener
{
public static final Apply.Factory FACTORY = new Apply.Factory()
{
@@ -77,7 +77,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
public static final IVersionedSerializer<AccordInteropApply> serializer = new ApplySerializer<AccordInteropApply>()
{
@Override
- protected AccordInteropApply deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
+ protected AccordInteropApply deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
return new AccordInteropApply(kind, txnId, scope, waitForEpoch, keys, executeAt, deps, txn, fullRoute, writes, result);
}
@@ -85,8 +85,9 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
transient BitSet waitingOn;
transient int waitingOnCount;
+ final MpscChunkedArrayQueue<LocalListeners.Registered> listeners = new MpscChunkedArrayQueue<>(4, 1 << 30);
- private AccordInteropApply(Kind kind, TxnId txnId, PartialRoute<?> route, long waitForEpoch, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
+ private AccordInteropApply(Kind kind, TxnId txnId, Route<?> route, long waitForEpoch, Seekables<?, ?> keys, Timestamp executeAt, PartialDeps deps, @Nullable PartialTxn txn, @Nullable FullRoute<?> fullRoute, Writes writes, Result result)
{
super(kind, txnId, route, waitForEpoch, keys, executeAt, deps, txn, fullRoute, writes, result);
}
@@ -137,7 +138,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
waitingOn.set(safeStore.commandStore().id());
++waitingOnCount;
}
- safeCommand.addListener(this);
+ listeners.add(safeStore.register(txnId, this));
break;
case Applied:
@@ -190,11 +191,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
private void cancel()
{
- node.commandStores().mapReduceConsume(this, waitingOn.stream(), forEach(safeStore -> {
- SafeCommand safeCommand = safeStore.ifInitialised(txnId);
- if (safeCommand != null)
- safeCommand.removeListener(this);
- }, node.agent()));
+ listeners.drain(LocalListeners.Registered::cancel);
}
@Override
@@ -234,7 +231,7 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
}
@Override
- public void onChange(SafeCommandStore safeStore, SafeCommand safeCommand)
+ public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand)
{
Command command = safeCommand.current();
@@ -248,20 +245,14 @@ public class AccordInteropApply extends Apply implements Command.TransientListen
case PreCommitted:
case Committed:
case PreApplied:
- return;
+ return true;
case Applied:
case Invalidated:
case Truncated:
}
- if (safeCommand.removeListener(this))
- ack();
- }
-
- @Override
- public PreLoadContext listenerPreLoadContext(TxnId caller)
- {
- return PreLoadContext.contextFor(txnId);
+ ack();
+ return false;
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java
index cd13b99f93..d1a26d22b4 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropCommit.java
@@ -29,8 +29,8 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -46,13 +46,13 @@ public class AccordInteropCommit extends Commit
public static final IVersionedSerializer<AccordInteropCommit> serializer = new CommitSerializer<AccordInteropCommit, AccordInteropRead>(AccordInteropRead.class, AccordInteropRead.requestSerializer)
{
@Override
- protected AccordInteropCommit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
+ protected AccordInteropCommit deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return new AccordInteropCommit(kind, txnId, scope, waitForEpoch, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, read);
}
};
- public AccordInteropCommit(Kind kind, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
+ public AccordInteropCommit(Kind kind, TxnId txnId, Route<?> scope, long waitForEpoch, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nonnull ReadData readData)
{
super(kind, txnId, scope, waitForEpoch, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, readData);
}
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
index a277f40d8d..eb8e76b40f 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
@@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.Data;
import accord.api.Result;
+import accord.coordinate.CoordinationAdapter;
import accord.local.AgentExecutor;
import accord.local.CommandStore;
import accord.local.Node;
@@ -93,7 +94,6 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.Clock;
import static accord.coordinate.CoordinationAdapter.Factory.Step.Continue;
-import static accord.coordinate.CoordinationAdapter.Invoke.persist;
import static accord.utils.Invariants.checkArgument;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteMetrics;
@@ -345,7 +345,7 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
CommandStore cs = node.commandStores().select(route.homeKey());
result.beginAsResult().withExecutor(cs).begin((data, failure) -> {
if (failure == null)
- persist(node.coordinationAdapter(txnId, Continue), node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
+ ((CoordinationAdapter)node.coordinationAdapter(txnId, Continue)).persist(node, executes, route, txnId, txn, executeAt, deps, txn.execute(txnId, executeAt, data), txn.result(txnId, executeAt, data), callback);
else
callback.accept(null, failure);
});
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
index 857d082a2c..a0fd42229f 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropPersist.java
@@ -22,7 +22,7 @@ import java.util.function.BiConsumer;
import accord.api.Result;
import accord.coordinate.Persist;
-import accord.coordinate.tracking.AppliedTracker;
+import accord.coordinate.tracking.AllTracker;
import accord.coordinate.tracking.QuorumTracker;
import accord.coordinate.tracking.RequestStatus;
import accord.coordinate.tracking.ResponseTracker;
@@ -110,7 +110,7 @@ public class AccordInteropPersist extends Persist
callback = new CallbackHolder(new QuorumTracker(topologies), result, clientCallback);
break;
case ALL:
- callback = new CallbackHolder(new AppliedTracker(topologies), result, clientCallback);
+ callback = new CallbackHolder(new AllTracker(topologies), result, clientCallback);
break;
default:
throw new IllegalArgumentException("Unhandled consistency level: " + consistencyLevel);
diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
index 708443f025..8e4ec2da26 100644
--- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
+++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
@@ -108,7 +108,7 @@ public class AccordInteropReadRepair extends ReadData
private final Mutation mutation;
- private static final IVersionedSerializer<Data> noop_data_serializer = new IVersionedSerializer<Data>()
+ private static final IVersionedSerializer<Data> noop_data_serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(Data t, DataOutputPlus out, int version) throws IOException {}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java
index 8377991354..99c7f820cc 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/AcceptSerializers.java
@@ -22,7 +22,7 @@ import java.io.IOException;
import accord.messages.Accept;
import accord.messages.Accept.AcceptReply;
-import accord.primitives.PartialRoute;
+import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -47,9 +47,9 @@ public class AcceptSerializers
}
@Override
- public Accept deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
+ public Accept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
- return create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey,
+ return create(txnId, scope, waitForEpoch, minEpoch,
CommandSerializers.ballot.deserialize(in, version),
CommandSerializers.timestamp.deserialize(in, version),
KeySerializers.seekables.deserialize(in, version),
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
index 102ffb57b9..8370d59b80 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
@@ -24,8 +24,8 @@ import accord.api.Result;
import accord.messages.Apply;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@@ -58,7 +58,6 @@ public class ApplySerializers
}
};
-// public static final IVersionedSerializer<Apply> request = new TxnRequestSerializer<Apply>()
public abstract static class ApplySerializer<A extends Apply> extends TxnRequestSerializer<A>
{
@Override
@@ -73,11 +72,11 @@ public class ApplySerializers
CommandSerializers.writes.serialize(apply.writes, out, version);
}
- protected abstract A deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
+ protected abstract A deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result);
@Override
- public A deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
+ public A deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
return deserializeApply(txnId, scope, waitForEpoch,
kind.deserialize(in, version),
@@ -106,7 +105,7 @@ public class ApplySerializers
public static final IVersionedSerializer<Apply> request = new ApplySerializer<Apply>()
{
@Override
- protected Apply deserializeApply(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
+ protected Apply deserializeApply(TxnId txnId, Route<?> scope, long waitForEpoch, Apply.Kind kind, Seekables<?, ?> keys,
Timestamp executeAt, PartialDeps deps, PartialTxn txn, FullRoute<?> fullRoute, Writes writes, Result result)
{
return Apply.SerializationSupport.create(txnId, scope, waitForEpoch, kind, keys, executeAt, deps, txn, fullRoute, writes, result);
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java
new file mode 100644
index 0000000000..f18b40f473
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/accord/serializers/AwaitSerializer.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.serializers;
+
+import java.io.IOException;
+
+import accord.api.ProgressLog.BlockedUntil;
+import accord.local.SaveStatus;
+import accord.messages.Await;
+import accord.messages.Await.AsyncAwaitComplete;
+import accord.messages.Await.AwaitOk;
+import accord.primitives.Participants;
+import accord.primitives.Route;
+import accord.primitives.TxnId;
+import accord.utils.Invariants;
+import org.apache.cassandra.db.TypeSizes;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.utils.vint.VIntCoding;
+
+public class AwaitSerializer
+{
+ public static final IVersionedSerializer<Await> request = new IVersionedSerializer<>()
+ {
+ @Override
+ public void serialize(Await await, DataOutputPlus out, int version) throws IOException
+ {
+ CommandSerializers.txnId.serialize(await.txnId, out, version);
+ KeySerializers.participants.serialize(await.scope, out, version);
+ out.writeByte(await.blockedUntil.ordinal());
+ out.writeUnsignedVInt32(await.callbackId + 1);
+ Invariants.checkState(await.callbackId >= -1);
+ }
+
+ @Override
+ public Await deserialize(DataInputPlus in, int version) throws IOException
+ {
+ TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
+ Participants<?> scope = KeySerializers.participants.deserialize(in, version);
+ BlockedUntil blockedUntil = BlockedUntil.forOrdinal(in.readByte());
+ int callbackId = in.readUnsignedVInt32() - 1;
+ Invariants.checkState(callbackId >= -1);
+ return Await.SerializerSupport.create(txnId, scope, blockedUntil, callbackId);
+ }
+
+ @Override
+ public long serializedSize(Await await, int version)
+ {
+ return CommandSerializers.txnId.serializedSize(await.txnId, version)
+ + KeySerializers.participants.serializedSize(await.scope, version)
+ + TypeSizes.BYTE_SIZE
+ + VIntCoding.computeUnsignedVIntSize(await.callbackId + 1);
+ }
+ };
+
+ public static final IVersionedSerializer<AwaitOk> syncReply = new EnumSerializer<>(AwaitOk.class);
+
+ public static final IVersionedSerializer<AsyncAwaitComplete> asyncReply = new IVersionedSerializer<>()
+ {
+ @Override
+ public void serialize(AsyncAwaitComplete ok, DataOutputPlus out, int version) throws IOException
+ {
+ CommandSerializers.txnId.serialize(ok.txnId, out, version);
+ KeySerializers.route.serialize(ok.route, out, version);
+ out.writeByte(ok.newStatus.ordinal());
+ out.writeUnsignedVInt32(ok.callbackId);
+ }
+
+ @Override
+ public AsyncAwaitComplete deserialize(DataInputPlus in, int version) throws IOException
+ {
+ TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
+ Route<?> scope = KeySerializers.route.deserialize(in, version);
+ SaveStatus newStatus = SaveStatus.forOrdinal(in.readByte());
+ int callbackId = in.readUnsignedVInt32();
+ return new AsyncAwaitComplete(txnId, scope, newStatus, callbackId);
+ }
+
+ @Override
+ public long serializedSize(AsyncAwaitComplete ok, int version)
+ {
+ return CommandSerializers.txnId.serializedSize(ok.txnId, version)
+ + KeySerializers.route.serializedSize(ok.route, version)
+ + TypeSizes.BYTE_SIZE
+ + VIntCoding.computeVIntSize(ok.callbackId);
+ }
+ };
+}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
index 54c130ac2b..94390bd905 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
@@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.RoutingKey;
-import accord.local.Status;
+import accord.local.SaveStatus;
import accord.messages.BeginInvalidation;
import accord.messages.BeginInvalidation.InvalidateReply;
import accord.primitives.Ballot;
@@ -67,7 +67,8 @@ public class BeginInvalidationSerializers
{
CommandSerializers.nullableBallot.serialize(reply.supersededBy, out, version);
CommandSerializers.ballot.serialize(reply.accepted, out, version);
- CommandSerializers.status.serialize(reply.status, out, version);
+ CommandSerializers.saveStatus.serialize(reply.maxStatus, out, version);
+ CommandSerializers.saveStatus.serialize(reply.maxKnowledgeStatus, out, version);
out.writeBoolean(reply.acceptedFastPath);
KeySerializers.nullableRoute.serialize(reply.route, out, version);
KeySerializers.nullableRoutingKey.serialize(reply.homeKey, out, version);
@@ -78,11 +79,12 @@ public class BeginInvalidationSerializers
{
Ballot supersededBy = CommandSerializers.nullableBallot.deserialize(in, version);
Ballot accepted = CommandSerializers.ballot.deserialize(in, version);
- Status status = CommandSerializers.status.deserialize(in, version);
+ SaveStatus maxStatus = CommandSerializers.saveStatus.deserialize(in, version);
+ SaveStatus maxKnowledgeStatus = CommandSerializers.saveStatus.deserialize(in, version);
boolean acceptedFastPath = in.readBoolean();
Route<?> route = KeySerializers.nullableRoute.deserialize(in, version);
RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
- return new InvalidateReply(supersededBy, accepted, status, acceptedFastPath, route, homeKey);
+ return new InvalidateReply(supersededBy, accepted, maxStatus, maxKnowledgeStatus, acceptedFastPath, route, homeKey);
}
@Override
@@ -90,7 +92,8 @@ public class BeginInvalidationSerializers
{
return CommandSerializers.nullableBallot.serializedSize(reply.supersededBy, version)
+ CommandSerializers.ballot.serializedSize(reply.accepted, version)
- + CommandSerializers.status.serializedSize(reply.status, version)
+ + CommandSerializers.saveStatus.serializedSize(reply.maxStatus, version)
+ + CommandSerializers.saveStatus.serializedSize(reply.maxKnowledgeStatus, version)
+ TypeSizes.BOOL_SIZE
+ KeySerializers.nullableRoute.serializedSize(reply.route, version)
+ KeySerializers.nullableRoutingKey.serializedSize(reply.homeKey, version);
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CalculateDepsSerializers.java
similarity index 62%
rename from src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java
rename to src/java/org/apache/cassandra/service/accord/serializers/CalculateDepsSerializers.java
index 37f9302ae3..e10425a321 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/GetDepsSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CalculateDepsSerializers.java
@@ -20,9 +20,9 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
-import accord.messages.GetDeps;
-import accord.messages.GetDeps.GetDepsOk;
-import accord.primitives.PartialRoute;
+import accord.messages.CalculateDeps;
+import accord.messages.CalculateDeps.CalculateDepsOk;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@@ -30,49 +30,49 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
-public class GetDepsSerializers
+public class CalculateDepsSerializers
{
- public static final IVersionedSerializer<GetDeps> request = new TxnRequestSerializer.WithUnsyncedSerializer<GetDeps>()
+ public static final IVersionedSerializer<CalculateDeps> request = new TxnRequestSerializer.WithUnsyncedSerializer<CalculateDeps>()
{
@Override
- public void serializeBody(GetDeps msg, DataOutputPlus out, int version) throws IOException
+ public void serializeBody(CalculateDeps msg, DataOutputPlus out, int version) throws IOException
{
KeySerializers.seekables.serialize(msg.keys, out, version);
CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
}
@Override
- public GetDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
+ public CalculateDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version);
- return GetDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, keys, executeAt);
+ return CalculateDeps.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, keys, executeAt);
}
@Override
- public long serializedBodySize(GetDeps msg, int version)
+ public long serializedBodySize(CalculateDeps msg, int version)
{
return KeySerializers.seekables.serializedSize(msg.keys, version)
+ CommandSerializers.timestamp.serializedSize(msg.executeAt, version);
}
};
- public static final IVersionedSerializer<GetDepsOk> reply = new IVersionedSerializer<GetDepsOk>()
+ public static final IVersionedSerializer<CalculateDepsOk> reply = new IVersionedSerializer<CalculateDepsOk>()
{
@Override
- public void serialize(GetDepsOk reply, DataOutputPlus out, int version) throws IOException
+ public void serialize(CalculateDepsOk reply, DataOutputPlus out, int version) throws IOException
{
DepsSerializer.partialDeps.serialize(reply.deps, out, version);
}
@Override
- public GetDepsOk deserialize(DataInputPlus in, int version) throws IOException
+ public CalculateDepsOk deserialize(DataInputPlus in, int version) throws IOException
{
- return new GetDepsOk(DepsSerializer.partialDeps.deserialize(in, version));
+ return new CalculateDepsOk(DepsSerializer.partialDeps.deserialize(in, version));
}
@Override
- public long serializedSize(GetDepsOk reply, int version)
+ public long serializedSize(CalculateDepsOk reply, int version)
{
return DepsSerializer.partialDeps.serializedSize(reply.deps, version);
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
index 369fe57d8e..4dc9ed1805 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializer.java
@@ -41,7 +41,7 @@ import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.utils.vint.VIntCoding;
import static accord.local.cfk.CommandsForKey.NO_PENDING_UNMANAGED;
-import static accord.local.cfk.CommandsForKey.NO_TXNIDS;
+import static accord.primitives.TxnId.NO_TXNIDS;
import static accord.primitives.Txn.Kind.Read;
import static accord.primitives.Txn.Kind.SyncPoint;
import static accord.primitives.Txn.Kind.Write;
@@ -117,19 +117,21 @@ public class CommandsForKeySerializer
if (commandCount == 0)
return ByteBuffer.allocate(1);
- int[] nodeIds = cachedInts().getInts(Math.min(64, commandCount));
+ int[] nodeIds = cachedInts().getInts(Math.min(64, Math.max(4, commandCount)));
try
{
// first compute the unique Node Ids and some basic characteristics of the data, such as
// whether we have any missing transactions to encode, any executeAt that are not equal to their TxnId
// and whether there are any non-standard flag bits to encode
boolean hasNonStandardFlags = false;
- int nodeIdCount = 0, missingIdCount = 0, executeAtCount = 0, ballotCount = 0;
+ int nodeIdCount, missingIdCount = 0, executeAtCount = 0, ballotCount = 0;
int bitsPerExecuteAtEpoch = 0, bitsPerExecuteAtFlags = 0, bitsPerExecuteAtHlc = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
{
+ nodeIds[0] = cfk.redundantBefore().node.id;
+ nodeIdCount = 1;
for (int i = 0 ; i < commandCount ; ++i)
{
- if (nodeIdCount + 2 >= nodeIds.length)
+ if (nodeIdCount + 3 >= nodeIds.length)
{
nodeIdCount = compact(nodeIds);
if (nodeIdCount > nodeIds.length/2 || nodeIdCount + 2 >= nodeIds.length)
@@ -176,10 +178,10 @@ public class CommandsForKeySerializer
int maxHeaderBits = minHeaderBits;
int totalBytes = 0;
- long prevEpoch = cfk.get(0).epoch();
- long prevHlc = cfk.get(0).hlc();
int prunedBeforeIndex = cfk.prunedBefore().equals(TxnId.NONE) ? -1 : cfk.indexOf(cfk.prunedBefore());
+ long prevEpoch = cfk.redundantBefore().epoch();
+ long prevHlc = cfk.redundantBefore().hlc();
int[] bytesHistogram = cachedInts().getInts(12);
Arrays.fill(bytesHistogram, 0);
for (int i = 0 ; i < commandCount ; ++i)
@@ -277,10 +279,12 @@ public class CommandsForKeySerializer
cachedInts().forceDiscard(bytesHistogram);
- prevEpoch = cfk.get(0).epoch();
- prevHlc = cfk.get(0).hlc();
+ prevEpoch = cfk.redundantBefore().epoch();
+ prevHlc = cfk.redundantBefore().hlc();
totalBytes += TypeSizes.sizeofUnsignedVInt(prevEpoch);
totalBytes += TypeSizes.sizeofUnsignedVInt(prevHlc);
+ totalBytes += TypeSizes.sizeofUnsignedVInt(cfk.redundantBefore().flags());
+ totalBytes += TypeSizes.sizeofUnsignedVInt(Arrays.binarySearch(nodeIds, 0, nodeIdCount, cfk.redundantBefore().node.id));
totalBytes += TypeSizes.sizeofUnsignedVInt(prunedBeforeIndex + 1);
int bitsPerBallotEpoch = 0, bitsPerBallotHlc = 1, bitsPerBallotFlags = 0;
@@ -348,8 +352,11 @@ public class CommandsForKeySerializer
VIntCoding.writeUnsignedVInt32(nodeIds[i] - nodeIds[i-1], out);
out.putShort((short)flags);
+
VIntCoding.writeUnsignedVInt(prevEpoch, out);
VIntCoding.writeUnsignedVInt(prevHlc, out);
+ VIntCoding.writeUnsignedVInt32(cfk.redundantBefore().flags(), out);
+ VIntCoding.writeUnsignedVInt32(Arrays.binarySearch(nodeIds, 0, nodeIdCount, cfk.redundantBefore().node.id), out);
VIntCoding.writeUnsignedVInt32(prunedBeforeIndex + 1, out);
int executeAtMask = executeAtCount > 0 ? 1 : 0;
@@ -372,7 +379,7 @@ public class CommandsForKeySerializer
bits |= hasExecuteAt << bitIndex;
bitIndex += statusHasInfo & executeAtMask;
- long hasMissingIds = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).missing != CommandsForKey.NO_TXNIDS ? 1 : 0;
+ long hasMissingIds = info.getClass() == TxnInfoExtra.class && ((TxnInfoExtra)info).missing != NO_TXNIDS ? 1 : 0;
bits |= hasMissingIds << bitIndex;
bitIndex += statusHasInfo & missingDepsMask;
@@ -639,6 +646,12 @@ public class CommandsForKeySerializer
long prevEpoch = VIntCoding.readUnsignedVInt(in);
long prevHlc = VIntCoding.readUnsignedVInt(in);
+ TxnId redundantBefore;
+ {
+ int flags = VIntCoding.readUnsignedVInt32(in);
+ Node.Id node = nodeIds[VIntCoding.readUnsignedVInt32(in)];
+ redundantBefore = TxnId.fromValues(prevEpoch, prevHlc, flags, node);
+ }
int prunedBeforeIndex = VIntCoding.readUnsignedVInt32(in) - 1;
for (int i = 0 ; i < commandCount ; ++i)
@@ -879,7 +892,7 @@ public class CommandsForKeySerializer
}
cachedTxnIds().forceDiscard(txnIds, commandCount);
- return CommandsForKey.SerializerSupport.create(key, txns, unmanageds, prunedBeforeIndex == -1 ? TxnId.NONE : txns[prunedBeforeIndex]);
+ return CommandsForKey.SerializerSupport.create(key, txns, unmanageds, redundantBefore, prunedBeforeIndex == -1 ? TxnId.NONE : txns[prunedBeforeIndex]);
}
private static int getHlcBytes(int lookup, int index)
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
index 629da34f77..9e439233ae 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
@@ -26,8 +26,8 @@ import accord.messages.ReadData;
import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@@ -68,13 +68,13 @@ public class CommitSerializers
serializeNullable(msg.readData, out, version, read);
}
- protected abstract C deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind,
+ protected abstract C deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Commit.Kind kind,
Ballot ballot, Timestamp executeAt,
Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps,
@Nullable FullRoute<?> fullRoute, @Nullable ReadData read);
@Override
- public C deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
+ public C deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
Commit.Kind kind = CommitSerializers.kind.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
@@ -104,7 +104,7 @@ public class CommitSerializers
public static final IVersionedSerializer<Commit> request = new CommitSerializer<Commit, ReadData>(ReadData.class, ReadDataSerializers.readData)
{
@Override
- protected Commit deserializeCommit(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
+ protected Commit deserializeCommit(TxnId txnId, Route<?> scope, long waitForEpoch, Commit.Kind kind, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, @Nullable PartialTxn partialTxn, PartialDeps partialDeps, @Nullable FullRoute<?> fullRoute, @Nullable ReadData read)
{
return Commit.SerializerSupport.create(txnId, scope, waitForEpoch, kind, ballot, executeAt, keys, partialTxn, partialDeps, fullRoute, read);
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
index 4512776154..9389bcc51b 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
@@ -21,25 +21,11 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.Data;
-import accord.api.Result;
-import accord.api.RoutingKey;
import accord.impl.AbstractFetchCoordinator.FetchRequest;
import accord.impl.AbstractFetchCoordinator.FetchResponse;
-import accord.local.SaveStatus;
-import accord.local.Status.Durability;
-import accord.local.Status.Known;
-import accord.messages.CheckStatus;
-import accord.messages.Propagate;
import accord.messages.ReadData.CommitOrReadNack;
import accord.messages.ReadData.ReadReply;
-import accord.primitives.Ballot;
-import accord.primitives.PartialDeps;
-import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
-import accord.primitives.Route;
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import accord.primitives.Writes;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
@@ -134,92 +120,4 @@ public class FetchSerializers
}
};
- public static final IVersionedSerializer<Propagate> propagate = new IVersionedSerializer<Propagate>()
- {
- @Override
- public void serialize(Propagate p, DataOutputPlus out, int version) throws IOException
- {
- CommandSerializers.txnId.serialize(p.txnId, out, version);
- KeySerializers.route.serialize(p.route, out, version);
- CommandSerializers.saveStatus.serialize(p.maxKnowledgeSaveStatus, out, version);
- CommandSerializers.saveStatus.serialize(p.maxSaveStatus, out, version);
- CommandSerializers.ballot.serialize(p.ballot, out, version);
- CommandSerializers.durability.serialize(p.durability, out, version);
- KeySerializers.nullableRoutingKey.serialize(p.homeKey, out, version);
- KeySerializers.nullableRoutingKey.serialize(p.progressKey, out, version);
- CommandSerializers.known.serialize(p.achieved, out, version);
- CheckStatusSerializers.foundKnownMap.serialize(p.known, out, version);
- out.writeBoolean(p.isShardTruncated);
- CommandSerializers.nullablePartialTxn.serialize(p.partialTxn, out, version);
- DepsSerializer.nullablePartialDeps.serialize(p.stableDeps, out, version);
- out.writeLong(p.toEpoch);
- CommandSerializers.nullableTimestamp.serialize(p.committedExecuteAt, out, version);
- CommandSerializers.nullableWrites.serialize(p.writes, out, version);
- }
-
- @Override
- public Propagate deserialize(DataInputPlus in, int version) throws IOException
- {
- TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
- Route<?> route = KeySerializers.route.deserialize(in, version);
- SaveStatus maxKnowledgeSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
- SaveStatus maxSaveStatus = CommandSerializers.saveStatus.deserialize(in, version);
- Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
- Durability durability = CommandSerializers.durability.deserialize(in, version);
- RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
- RoutingKey progressKey = KeySerializers.nullableRoutingKey.deserialize(in, version);
- Known achieved = CommandSerializers.known.deserialize(in, version);
- CheckStatus.FoundKnownMap known = CheckStatusSerializers.foundKnownMap.deserialize(in, version);
- boolean isTruncated = in.readBoolean();
- PartialTxn partialTxn = CommandSerializers.nullablePartialTxn.deserialize(in, version);
- PartialDeps committedDeps = DepsSerializer.nullablePartialDeps.deserialize(in, version);
- long toEpoch = in.readLong();
- Timestamp committedExecuteAt = CommandSerializers.nullableTimestamp.deserialize(in, version);
- Writes writes = CommandSerializers.nullableWrites.deserialize(in, version);
-
- Result result = null;
- if (achieved.outcome.isOrWasApply())
- result = CommandSerializers.APPLIED;
-
- return Propagate.SerializerSupport.create(txnId,
- route,
- maxKnowledgeSaveStatus,
- maxSaveStatus,
- ballot,
- durability,
- homeKey,
- progressKey,
- achieved,
- known,
- isTruncated,
- partialTxn,
- committedDeps,
- toEpoch,
- committedExecuteAt,
- writes,
- result);
- }
-
- @Override
- public long serializedSize(Propagate p, int version)
- {
- return CommandSerializers.txnId.serializedSize(p.txnId, version)
- + KeySerializers.route.serializedSize(p.route, version)
- + CommandSerializers.saveStatus.serializedSize(p.maxKnowledgeSaveStatus, version)
- + CommandSerializers.saveStatus.serializedSize(p.maxSaveStatus, version)
- + CommandSerializers.ballot.serializedSize(p.ballot, version)
- + CommandSerializers.durability.serializedSize(p.durability, version)
- + KeySerializers.nullableRoutingKey.serializedSize(p.homeKey, version)
- + KeySerializers.nullableRoutingKey.serializedSize(p.progressKey, version)
- + CommandSerializers.known.serializedSize(p.achieved, version)
- + CheckStatusSerializers.foundKnownMap.serializedSize(p.known, version)
- + TypeSizes.BOOL_SIZE
- + CommandSerializers.nullablePartialTxn.serializedSize(p.partialTxn, version)
- + DepsSerializer.nullablePartialDeps.serializedSize(p.stableDeps, version)
- + TypeSizes.sizeof(p.toEpoch)
- + CommandSerializers.nullableTimestamp.serializedSize(p.committedExecuteAt, version)
- + CommandSerializers.nullableWrites.serializedSize(p.writes, version)
- ;
- }
- };
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/GetEphmrlReadDepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/GetEphmrlReadDepsSerializers.java
index 44d202d716..7fe67842de 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/GetEphmrlReadDepsSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/GetEphmrlReadDepsSerializers.java
@@ -23,7 +23,7 @@ import java.io.IOException;
import accord.messages.GetEphemeralReadDeps;
import accord.messages.GetEphemeralReadDeps.GetEphemeralReadDepsOk;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
@@ -43,7 +43,7 @@ public class GetEphmrlReadDepsSerializers
}
@Override
- public GetEphemeralReadDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
+ public GetEphemeralReadDeps deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
long executionEpoch = in.readUnsignedVInt();
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java
index d50e6993b9..ad8af3ba88 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/GetMaxConflictSerializers.java
@@ -22,7 +22,7 @@ import java.io.IOException;
import accord.messages.GetMaxConflict;
import accord.messages.GetMaxConflict.GetMaxConflictOk;
-import accord.primitives.PartialRoute;
+import accord.primitives.Route;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@@ -43,7 +43,7 @@ public class GetMaxConflictSerializers
}
@Override
- public GetMaxConflict deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
+ public GetMaxConflict deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
Seekables<?, ?> keys = KeySerializers.seekables.deserialize(in, version);
long executionEpoch = in.readUnsignedVInt();
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java
index 66c649eb92..d23e6c99b9 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java
@@ -22,7 +22,7 @@ import java.io.IOException;
import accord.local.Status;
import accord.messages.InformDurable;
-import accord.primitives.PartialRoute;
+import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -36,14 +36,14 @@ public class InformDurableSerializers
@Override
public void serializeBody(InformDurable msg, DataOutputPlus out, int version) throws IOException
{
- CommandSerializers.timestamp.serialize(msg.executeAt, out, version);
+ CommandSerializers.nullableTimestamp.serialize(msg.executeAt, out, version);
CommandSerializers.durability.serialize(msg.durability, out, version);
}
@Override
- public InformDurable deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute scope, long waitForEpoch) throws IOException
+ public InformDurable deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
- Timestamp executeAt = CommandSerializers.timestamp.deserialize(in, version);
+ Timestamp executeAt = CommandSerializers.nullableTimestamp.deserialize(in, version);
Status.Durability durability = CommandSerializers.durability.deserialize(in, version);
return InformDurable.SerializationSupport.create(txnId, scope, waitForEpoch, executeAt, durability);
}
@@ -51,7 +51,7 @@ public class InformDurableSerializers
@Override
public long serializedBodySize(InformDurable msg, int version)
{
- return CommandSerializers.timestamp.serializedSize(msg.executeAt, version)
+ return CommandSerializers.nullableTimestamp.serializedSize(msg.executeAt, version)
+ CommandSerializers.durability.serializedSize(msg.durability, version);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/InformHomeDurableSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/InformHomeDurableSerializers.java
deleted file mode 100644
index c4f2155577..0000000000
--- a/src/java/org/apache/cassandra/service/accord/serializers/InformHomeDurableSerializers.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord.serializers;
-
-import java.io.IOException;
-
-import accord.messages.InformHomeDurable;
-import org.apache.cassandra.io.IVersionedSerializer;
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataOutputPlus;
-
-public class InformHomeDurableSerializers
-{
- public static final IVersionedSerializer<InformHomeDurable> request = new IVersionedSerializer<>()
- {
- @Override
- public void serialize(InformHomeDurable inform, DataOutputPlus out, int version) throws IOException
- {
- CommandSerializers.txnId.serialize(inform.txnId, out, version);
- KeySerializers.route.serialize(inform.route, out, version);
- CommandSerializers.timestamp.serialize(inform.executeAt, out, version);
- CommandSerializers.durability.serialize(inform.durability, out, version);
- }
-
- @Override
- public InformHomeDurable deserialize(DataInputPlus in, int version) throws IOException
- {
- return new InformHomeDurable(CommandSerializers.txnId.deserialize(in, version),
- KeySerializers.route.deserialize(in, version),
- CommandSerializers.timestamp.deserialize(in, version),
- CommandSerializers.durability.deserialize(in, version));
- }
-
- @Override
- public long serializedSize(InformHomeDurable inform, int version)
- {
- return CommandSerializers.txnId.serializedSize(inform.txnId, version)
- + KeySerializers.route.serializedSize(inform.route, version)
- + CommandSerializers.timestamp.serializedSize(inform.executeAt, version)
- + CommandSerializers.durability.serializedSize(inform.durability, version);
- }
-
- };
-}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/InformOfTxnIdSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/InformOfTxnIdSerializers.java
deleted file mode 100644
index c6f2098a16..0000000000
--- a/src/java/org/apache/cassandra/service/accord/serializers/InformOfTxnIdSerializers.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord.serializers;
-
-import java.io.IOException;
-
-import accord.messages.InformOfTxnId;
-import org.apache.cassandra.io.IVersionedSerializer;
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataOutputPlus;
-
-public class InformOfTxnIdSerializers
-{
- public static final IVersionedSerializer<InformOfTxnId> request = new IVersionedSerializer<InformOfTxnId>()
- {
- @Override
- public void serialize(InformOfTxnId inform, DataOutputPlus out, int version) throws IOException
- {
- CommandSerializers.txnId.serialize(inform.txnId, out, version);
- KeySerializers.route.serialize(inform.someRoute, out, version);
- }
-
- @Override
- public InformOfTxnId deserialize(DataInputPlus in, int version) throws IOException
- {
- return new InformOfTxnId(CommandSerializers.txnId.deserialize(in, version),
- KeySerializers.route.deserialize(in, version));
- }
-
- @Override
- public long serializedSize(InformOfTxnId inform, int version)
- {
- return CommandSerializers.txnId.serializedSize(inform.txnId, version)
- + KeySerializers.route.serializedSize(inform.someRoute, version);
- }
- };
-}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java
deleted file mode 100644
index 150c5af394..0000000000
--- a/src/java/org/apache/cassandra/service/accord/serializers/ListenerSerializers.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord.serializers;
-
-import java.io.IOException;
-
-import accord.local.Command;
-import org.apache.cassandra.db.TypeSizes;
-import org.apache.cassandra.io.IVersionedSerializer;
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataOutputPlus;
-
-public class ListenerSerializers
-{
- public enum Kind
- {
- COMMAND;
-
- private static Kind of(Command.DurableAndIdempotentListener listener)
- {
- if (listener instanceof Command.ProxyListener)
- return COMMAND;
-
- throw new IllegalArgumentException("Unsupported listener type: " + listener.getClass().getName());
- }
- }
-
-
- private static final IVersionedSerializer<Command.ProxyListener> commandListener = new IVersionedSerializer<Command.ProxyListener>()
- {
- @Override
- public void serialize(Command.ProxyListener listener, DataOutputPlus out, int version) throws IOException
- {
- CommandSerializers.txnId.serialize(listener.txnId(), out, version);
- }
-
- @Override
- public Command.ProxyListener deserialize(DataInputPlus in, int version) throws IOException
- {
- return new Command.ProxyListener(CommandSerializers.txnId.deserialize(in, version));
- }
-
- @Override
- public long serializedSize(Command.ProxyListener listener, int version)
- {
- return CommandSerializers.txnId.serializedSize(listener.txnId(), version);
- }
- };
-
- public static final IVersionedSerializer<Command.DurableAndIdempotentListener> listener = new IVersionedSerializer<Command.DurableAndIdempotentListener>()
- {
- @Override
- public void serialize(Command.DurableAndIdempotentListener listener, DataOutputPlus out, int version) throws IOException
- {
- Kind kind = Kind.of(listener);
- out.write(kind.ordinal());
- switch (kind)
- {
- case COMMAND:
- commandListener.serialize((Command.ProxyListener) listener, out, version);
- break;
- default:
- throw new IllegalArgumentException();
- }
- }
-
- @Override
- public Command.DurableAndIdempotentListener deserialize(DataInputPlus in, int version) throws IOException
- {
- Kind kind = Kind.values()[in.readByte()];
- switch (kind)
- {
- case COMMAND:
- return commandListener.deserialize(in, version);
- default:
- throw new IllegalArgumentException();
- }
- }
-
- @Override
- public long serializedSize(Command.DurableAndIdempotentListener listener, int version)
- {
- Kind kind = Kind.of(listener);
- long size = TypeSizes.BYTE_SIZE;
- switch (kind)
- {
- case COMMAND:
- size += commandListener.serializedSize((Command.ProxyListener) listener, version);
- break;
- default:
- throw new IllegalArgumentException();
- }
-
- return size;
- }
- };
-}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
index c12e92e18b..89f4abdec1 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
@@ -25,8 +25,8 @@ import accord.messages.PreAccept;
import accord.messages.PreAccept.PreAcceptOk;
import accord.messages.PreAccept.PreAcceptReply;
import accord.primitives.FullRoute;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -49,16 +49,16 @@ public class PreacceptSerializers
{
CommandSerializers.partialTxn.serialize(msg.partialTxn, out, version);
serializeNullable(msg.route, out, version, KeySerializers.fullRoute);
- out.writeUnsignedVInt(msg.maxEpoch - msg.minUnsyncedEpoch);
+ out.writeUnsignedVInt(msg.maxEpoch - msg.minEpoch);
}
@Override
- public PreAccept deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException
+ public PreAccept deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException
{
PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version);
@Nullable FullRoute<?> fullRoute = deserializeNullable(in, version, KeySerializers.fullRoute);
long maxEpoch = in.readUnsignedVInt() + minEpoch;
- return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey,
+ return PreAccept.SerializerSupport.create(txnId, scope, waitForEpoch, minEpoch,
maxEpoch, partialTxn, fullRoute);
}
@@ -67,7 +67,7 @@ public class PreacceptSerializers
{
return CommandSerializers.partialTxn.serializedSize(msg.partialTxn, version)
+ serializedNullableSize(msg.route, version, KeySerializers.fullRoute)
- + TypeSizes.sizeofUnsignedVInt(msg.maxEpoch - msg.minUnsyncedEpoch);
+ + TypeSizes.sizeofUnsignedVInt(msg.maxEpoch - msg.minEpoch);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
index 346d1c8bdf..5caab8fb2b 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
@@ -33,8 +33,8 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.LatestDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
+import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
@@ -60,7 +60,7 @@ public class RecoverySerializers
}
@Override
- public BeginRecovery deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
+ public BeginRecovery deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
index c9da12801e..6dc5f9e564 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/TxnRequestSerializer.java
@@ -21,7 +21,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.messages.TxnRequest;
-import accord.primitives.PartialRoute;
+import accord.primitives.Route;
import accord.primitives.TxnId;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -33,7 +33,7 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
void serializeHeader(T msg, DataOutputPlus out, int version) throws IOException
{
CommandSerializers.txnId.serialize(msg.txnId, out, version);
- KeySerializers.partialRoute.serialize(msg.scope, out, version);
+ KeySerializers.route.serialize(msg.scope, out, version);
out.writeUnsignedVInt(msg.waitForEpoch);
}
@@ -46,13 +46,13 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
serializeBody(msg, out, version);
}
- public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException;
+ public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException;
@Override
public final T deserialize(DataInputPlus in, int version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
- PartialRoute<?> scope = KeySerializers.partialRoute.deserialize(in, version);
+ Route<?> scope = KeySerializers.route.deserialize(in, version);
// TODO: there should be a base epoch
long waitForEpoch = in.readUnsignedVInt();
return deserializeBody(in, version, txnId, scope, waitForEpoch);
@@ -61,7 +61,7 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
long serializedHeaderSize(T msg, int version)
{
return CommandSerializers.txnId.serializedSize(msg.txnId, version)
- + KeySerializers.partialRoute.serializedSize(msg.scope(), version) +
+ + KeySerializers.route.serializedSize(msg.scope(), version) +
TypeSizes.sizeofUnsignedVInt(msg.waitForEpoch);
}
@@ -79,26 +79,23 @@ public abstract class TxnRequestSerializer<T extends TxnRequest<?>> implements I
void serializeHeader(T msg, DataOutputPlus out, int version) throws IOException
{
super.serializeHeader(msg, out, version);
- out.writeUnsignedVInt(msg.minUnsyncedEpoch);
- out.writeBoolean(msg.doNotComputeProgressKey);
+ out.writeUnsignedVInt(msg.minEpoch);
}
- public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey) throws IOException;
+ public abstract T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch, long minEpoch) throws IOException;
@Override
- public final T deserializeBody(DataInputPlus in, int version, TxnId txnId, PartialRoute<?> scope, long waitForEpoch) throws IOException
+ public final T deserializeBody(DataInputPlus in, int version, TxnId txnId, Route<?> scope, long waitForEpoch) throws IOException
{
long minEpoch = in.readUnsignedVInt();
- boolean doNotComputeProgressKey = in.readBoolean();
- return deserializeBody(in, version, txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey);
+ return deserializeBody(in, version, txnId, scope, waitForEpoch, minEpoch);
}
@Override
long serializedHeaderSize(T msg, int version)
{
long size = super.serializedHeaderSize(msg, version);
- size += TypeSizes.sizeofUnsignedVInt(msg.minUnsyncedEpoch);
- size += TypeSizes.BOOL_SIZE;
+ size += TypeSizes.sizeofUnsignedVInt(msg.minEpoch);
return size;
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/WaitOnCommitSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/WaitOnCommitSerializer.java
deleted file mode 100644
index 821b8d9a30..0000000000
--- a/src/java/org/apache/cassandra/service/accord/serializers/WaitOnCommitSerializer.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord.serializers;
-
-import java.io.IOException;
-
-import accord.messages.WaitOnCommit;
-import accord.messages.WaitOnCommit.WaitOnCommitOk;
-import accord.primitives.Participants;
-import accord.primitives.TxnId;
-import org.apache.cassandra.io.IVersionedSerializer;
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataOutputPlus;
-
-public class WaitOnCommitSerializer
-{
- public static final IVersionedSerializer<WaitOnCommit> request = new IVersionedSerializer<WaitOnCommit>()
- {
- @Override
- public void serialize(WaitOnCommit wait, DataOutputPlus out, int version) throws IOException
- {
- CommandSerializers.txnId.serialize(wait.txnId, out, version);
- KeySerializers.participants.serialize(wait.scope, out, version);
- }
-
- @Override
- public WaitOnCommit deserialize(DataInputPlus in, int version) throws IOException
- {
- TxnId txnId = CommandSerializers.txnId.deserialize(in, version);
- Participants<?> scope = KeySerializers.participants.deserialize(in, version);
- return WaitOnCommit.SerializerSupport.create(txnId, scope);
- }
-
- @Override
- public long serializedSize(WaitOnCommit wait, int version)
- {
- return CommandSerializers.txnId.serializedSize(wait.txnId, version)
- + KeySerializers.participants.serializedSize(wait.scope, version);
- }
- };
-
- public static final IVersionedSerializer<WaitOnCommitOk> reply = new IVersionedSerializer<WaitOnCommitOk>()
- {
- @Override
- public void serialize(WaitOnCommitOk ok, DataOutputPlus out, int version) throws IOException
- {
- }
-
- @Override
- public WaitOnCommitOk deserialize(DataInputPlus in, int version) throws IOException
- {
- return WaitOnCommitOk.INSTANCE;
- }
-
- @Override
- public long serializedSize(WaitOnCommitOk ok, int version)
- {
- return 0;
- }
- };
-}
diff --git a/src/java/org/apache/cassandra/triggers/TriggerExecutor.java b/src/java/org/apache/cassandra/triggers/TriggerExecutor.java
index 335131d805..c3fe269559 100644
--- a/src/java/org/apache/cassandra/triggers/TriggerExecutor.java
+++ b/src/java/org/apache/cassandra/triggers/TriggerExecutor.java
@@ -19,11 +19,6 @@
package org.apache.cassandra.triggers;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
import java.util.*;
import java.util.concurrent.TimeUnit;
diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml
index 91fd89b6ab..597d209b83 100644
--- a/test/conf/cassandra.yaml
+++ b/test/conf/cassandra.yaml
@@ -118,4 +118,3 @@ accord:
enabled: true
journal_directory: build/test/cassandra/accord_journal
shard_count: 4
- progress_log_schedule_delay: 1s
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 8a508b03a5..64f83dd2e4 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -446,7 +446,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
byte[] bytes = out.toByteArray();
if (messageOut.serializedSize(toVersion) != bytes.length)
throw new AssertionError(String.format("Message serializedSize(%s) does not match what was written with serialize(out, %s) for verb %s and serializer %s; " +
- "expected %s, actual %s ", toVersion, toVersion, messageOut.verb(), Message.serializer.getClass(),
+ "expected %s, actual %s ", toVersion, toVersion, messageOut.verb(), messageOut.verb().serializer().getClass(),
messageOut.serializedSize(toVersion), bytes.length));
return new MessageImpl(messageOut.verb().id, bytes, messageOut.id(), toVersion, messageOut.expiresAtNanos(), fromCassandraInetAddressAndPort(from));
}
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
index dd791aa8ac..d8267e8d10 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
@@ -99,7 +99,7 @@ public class InstanceConfig implements IInstanceConfig
.set("accord.enabled", accord.enabled)
.set("accord.journal_directory", accord.journal_directory)
.set("accord.shard_count", accord.shard_count.toString())
- .set("accord.progress_log_schedule_delay", accord.progress_log_schedule_delay.toString())
+ .set("accord.recover_delay", accord.recover_delay.toString())
.set("partitioner", "org.apache.cassandra.dht.Murmur3Partitioner")
.set("start_native_transport", true)
.set("concurrent_writes", 2)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java
index 528e5ad00c..3ccdad7997 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/QueriesTableTest.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
+import accord.impl.progresslog.DefaultProgressLogs;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
@@ -31,7 +32,6 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import accord.impl.SimpleProgressLog;
import com.datastax.driver.core.Session;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
@@ -177,7 +177,7 @@ public class QueriesTableTest extends TestBaseImpl
// Disable recovery to make sure only one local read occurs:
for (IInvokableInstance instance : SHARED_CLUSTER)
- instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = true);
+ instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(true));
String update = "BEGIN TRANSACTION\n" +
" LET row1 = (SELECT * FROM " + KEYSPACE + ".accord_tbl WHERE k = 0);\n" +
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
index 69d2622830..f295ce6e1d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
@@ -36,7 +36,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import accord.impl.SimpleProgressLog;
+import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@@ -156,7 +156,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
public void tearDown()
{
for (IInvokableInstance instance : SHARED_CLUSTER)
- instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = false);
+ instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(false));
SHARED_CLUSTER.filters().reset();
}
@@ -292,7 +292,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
// heal partition and wait for node 1 to see node 3 again
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> {
- SimpleProgressLog.PAUSE_FOR_TEST = true;
+ DefaultProgressLogs.unsafePauseForTesting(true);
Assert.assertTrue(agent().executedBarriers().isEmpty());
});
SHARED_CLUSTER.filters().reset();
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java
index 0ba67894d2..572ffaae49 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java
@@ -26,7 +26,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import accord.impl.SimpleProgressLog;
+import accord.impl.progresslog.DefaultProgressLogs;
import accord.messages.Commit;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IMessageFilters;
@@ -131,6 +131,6 @@ public class AccordIntegrationTest extends AccordTestBase
private void pauseSimpleProgressLog()
{
for (IInvokableInstance instance : SHARED_CLUSTER)
- instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = true);
+ instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(true));
}
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java
index e5865ef9d3..1be3fd686d 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java
@@ -42,8 +42,8 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.messages.PreAccept;
import accord.primitives.PartialKeyRoute;
-import accord.primitives.PartialRoute;
import accord.primitives.Routable.Domain;
+import accord.primitives.Route;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.Util;
import org.apache.cassandra.batchlog.BatchlogManager;
@@ -458,7 +458,7 @@ public abstract class AccordMigrationRaceTestBase extends AccordTestBase
{
boolean drop = cluster.get(to).callsOnInstance(() -> {
PreAccept preAccept = (PreAccept)Instance.deserializeMessage(message).payload;
- PartialRoute<?> route = preAccept.scope;
+ Route<?> route = preAccept.scope;
if (route.domain() == Domain.Key)
for (RoutingKey key : (PartialKeyRoute)route)
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
new file mode 100644
index 0000000000..86785256c3
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.accord;
+
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+
+public class AccordProgressLogTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(AccordProgressLogTest.class);
+
+ @Test
+ public void testRecoveryTimeWindow() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(3)
+ .withoutVNodes()
+ .withConfig(c -> c.with(Feature.NETWORK).set("accord.enabled", "true"))
+ .start()))
+ {
+ cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}");
+ cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key (k, c)) WITH " + TransactionalMode.full.asCqlParam());
+ String query = "BEGIN TRANSACTION\n" +
+ " SELECT * FROM ks.tbl WHERE k=0 AND c=0;\n" +
+ "COMMIT TRANSACTION";
+
+ IMessageFilters.Filter dropCommit = cluster.filters().outbound().from(1).verbs(Verb.ACCORD_COMMIT_REQ.id).drop();
+ AtomicLong recoveryStartedAt = new AtomicLong();
+ Semaphore waitForRecovery = new Semaphore(0);
+ IMessageFilters.Filter recovery = cluster.filters().outbound().messagesMatching((from, to, message) -> {
+ if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_RSP.id)
+ {
+ recoveryStartedAt.compareAndSet(0, System.nanoTime());
+ waitForRecovery.release();
+ }
+ return false;
+ }).drop();
+
+ long coordinationStartedAt = System.nanoTime();
+ boolean failed = false;
+ try
+ {
+ cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.ANY);
+ }
+ catch (Throwable e)
+ {
+ failed = true;
+ }
+ Assertions.assertTrue(failed);
+
+ waitForRecovery.acquire();
+ long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(recoveryStartedAt.get() - coordinationStartedAt);
+ Assertions.assertTrue(timeDeltaMillis >= 1000, "Recovery started in " + timeDeltaMillis + "ms");
+ Assertions.assertTrue(timeDeltaMillis <= 3000, "Recovery started in " + timeDeltaMillis + "ms");
+ }
+ }
+
+ @Test
+ public void testFetchTimeWindow() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(3)
+ .withoutVNodes()
+ .withConfig(c -> c.with(Feature.NETWORK).set("accord.enabled", "true"))
+ .start()))
+ {
+ cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}");
+ cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key (k, c)) WITH " + TransactionalMode.full.asCqlParam());
+ String query = "BEGIN TRANSACTION\n" +
+ " SELECT * FROM ks.tbl WHERE k=0 AND c=0;\n" +
+ "COMMIT TRANSACTION";
+
+ IMessageFilters.Filter dropApply = cluster.filters().outbound().from(1).verbs(Verb.ACCORD_APPLY_REQ.id).drop();
+ AtomicLong fetchStartedAt = new AtomicLong();
+ Semaphore waitForFetch = new Semaphore(0);
+ IMessageFilters.Filter fetch = cluster.filters().outbound().messagesMatching((from, to, message) -> {
+ if (message.verb() == Verb.ACCORD_AWAIT_REQ.id)
+ {
+ fetchStartedAt.compareAndSet(0, System.nanoTime());
+ waitForFetch.release();
+ }
+ return false;
+ }).drop();
+
+ long coordinationStartedAt = System.nanoTime();
+ try
+ {
+ cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.ANY);
+ }
+ catch (Throwable e)
+ {
+ }
+
+ waitForFetch.acquire();
+ logger.info("Coordinated at {}", coordinationStartedAt);
+ logger.info("Awaited at {}", fetchStartedAt.get());
+ long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(fetchStartedAt.get() - coordinationStartedAt);
+ Assertions.assertTrue(timeDeltaMillis >= 100, "Fetch started in " + timeDeltaMillis + "ms");
+ Assertions.assertTrue(timeDeltaMillis <= 2000, "Fetch started in " + timeDeltaMillis + "ms");
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
index f30ef762cc..c60c064e68 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
@@ -44,11 +44,11 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.coordinate.Invalidated;
-import accord.impl.SimpleProgressLog;
+import accord.impl.progresslog.DefaultProgressLogs;
import accord.messages.PreAccept;
import accord.primitives.PartialKeyRoute;
-import accord.primitives.PartialRoute;
import accord.primitives.Routable.Domain;
+import accord.primitives.Route;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
@@ -142,7 +142,7 @@ public abstract class AccordTestBase extends TestBaseImpl
public void tearDown() throws Exception
{
for (IInvokableInstance instance : SHARED_CLUSTER)
- instance.runOnInstance(() -> SimpleProgressLog.PAUSE_FOR_TEST = false);
+ instance.runOnInstance(() -> DefaultProgressLogs.unsafePauseForTesting(false));
}
protected static void assertRowSerial(Cluster cluster, String query, int k, int c, int v, int s)
@@ -558,7 +558,7 @@ public abstract class AccordTestBase extends TestBaseImpl
{
boolean drop = cluster.get(to).callsOnInstance(() -> {
PreAccept preAccept = (PreAccept)Instance.deserializeMessage(message).payload;
- PartialRoute<?> route = preAccept.scope;
+ Route<?> route = preAccept.scope;
if (route.domain() == Domain.Key)
for (RoutingKey key : (PartialKeyRoute)route)
{
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
index a955a9c315..e69bffb8e1 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
@@ -32,8 +32,8 @@ import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterators;
-import accord.primitives.Routable;
import accord.local.CommandStores;
+import accord.primitives.Route;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.distributed.shared.WithProperties;
import org.apache.cassandra.service.accord.*;
@@ -58,7 +58,6 @@ import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Seekable;
@@ -95,6 +94,7 @@ import org.assertj.core.api.Assertions;
import static accord.impl.TimestampsForKey.NO_LAST_EXECUTED_HLC;
import static accord.local.KeyHistory.COMMANDS;
import static accord.local.PreLoadContext.contextFor;
+import static accord.primitives.Routable.Domain.Range;
import static accord.utils.async.AsyncChains.getUninterruptibly;
import static org.apache.cassandra.Util.spinAssertEquals;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
@@ -118,7 +118,7 @@ public class CompactionAccordIteratorsTest
private static final TxnId LT_TXN_ID = AccordTestUtils.txnId(EPOCH, HLC_START, NODE);
private static final TxnId TXN_ID = AccordTestUtils.txnId(EPOCH, LT_TXN_ID.hlc() + 1, NODE);
private static final TxnId SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 1, NODE, Kind.Read);
- private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 2, NODE, Kind.Read, Routable.Domain.Range);
+ private static final TxnId RANGE_TXN_ID = AccordTestUtils.txnId(EPOCH, TXN_ID.hlc() + 2, NODE, Kind.Read, Range);
private static final TxnId GT_TXN_ID = SECOND_TXN_ID;
// For CommandsForKey where we test with two commands
private static final TxnId[] TXN_IDS = new TxnId[]{ TXN_ID, SECOND_TXN_ID };
@@ -383,8 +383,8 @@ public class CompactionAccordIteratorsTest
private static RedundantBefore redundantBefore(TxnId txnId)
{
Ranges ranges = AccordTestUtils.fullRange(AccordTestUtils.keys(table, 42));
- txnId = txnId.as(Kind.Read, Routable.Domain.Range);
- return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, LT_TXN_ID);
+ txnId = txnId.as(Kind.Read, Range);
+ return RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, txnId, txnId, LT_TXN_ID.as(Range));
}
enum DurableBeforeType
@@ -468,25 +468,22 @@ public class CompactionAccordIteratorsTest
Txn txn = txnId.kind().isWrite() ? writeTxn : readTxn;
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true);
- PartialRoute<?> partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
+ Route<?> partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
- CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToKeyspace(commandStore));
+ CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
- CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, txnId, partialDeps, appendDiffToKeyspace(commandStore));
+ CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), txnId, partialDeps, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
- CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, txnId, partialDeps, appendDiffToKeyspace(commandStore));
+ CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
Pair<Writes, Result> result = AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
- CheckedCommands.apply(safe, txnId, route, null, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
- }).beginAsResult());
- getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys(), COMMANDS), safe -> {
- safe.get(txnId, txnId, route).addListener(new Command.ProxyListener(RANGE_TXN_ID)); // add a junk listener just to test it in compaction
+ CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
}).beginAsResult());
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied
diff --git a/test/unit/org/apache/cassandra/dht/AccordSplitterTest.java b/test/unit/org/apache/cassandra/dht/AccordSplitterTest.java
index 2ddec3906f..532e0efe4d 100644
--- a/test/unit/org/apache/cassandra/dht/AccordSplitterTest.java
+++ b/test/unit/org/apache/cassandra/dht/AccordSplitterTest.java
@@ -98,7 +98,7 @@ public class AccordSplitterTest
Assertions.assertThat(ranges).describedAs("num splits not as expected for partitioner %s", partitioner).hasSizeBetween(numSplits, numSplits + 1);
Ranges split = Ranges.of(ranges.toArray(new Range[0])).mergeTouching();
- Ranges missing = Ranges.of(range).subtract(split);
+ Ranges missing = Ranges.of(range).without(split);
Assertions.assertThat(missing).isEmpty();
testEventSplit(partitioner, range, rs, numSplits);
@@ -115,7 +115,7 @@ public class AccordSplitterTest
Assertions.assertThat(ranges).describedAs("num splits not as expected for partitioner %s", partitioner).hasSize(numSplits);
Ranges split = ranges.stream().reduce(Ranges.EMPTY, Ranges::with).mergeTouching();
- Ranges missing = topLevel.subtract(split);
+ Ranges missing = topLevel.without(split);
Assertions.assertThat(missing).isEmpty();
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordAgentTest.java b/test/unit/org/apache/cassandra/service/accord/AccordAgentTest.java
new file mode 100644
index 0000000000..6699720920
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/accord/AccordAgentTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import accord.local.Node;
+import accord.utils.RandomTestRunner;
+import accord.utils.SortedArrays.SortedArrayList;
+import org.apache.cassandra.service.accord.api.AccordAgent;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertTrue;
+
+public class AccordAgentTest
+{
+ @Test
+ public void testNonClashingStartTimes()
+ {
+ RandomTestRunner.test().check(rnd -> {
+ SortedArrayList<Node.Id> nodes; {
+ Node.Id[] ids = new Node.Id[rnd.nextInt(4, 16)];
+ for (int i = 0 ; i < ids.length ; ++i)
+ ids[i] = new Node.Id(i);
+ nodes = new SortedArrayList<>(ids);
+ }
+
+ long[] startTimes = new long[nodes.size()];
+ long oneSecond = SECONDS.toMicros(1);
+ long targetDelta = oneSecond / nodes.size();
+ for (int i = 0 ; i < 10000 ; ++i)
+ {
+ long startTime = rnd.nextLong(1, TimeUnit.DAYS.toMicros(100L));
+ for (int j = 0 ; j < startTimes.length ; ++j)
+ {
+ long nonClashingStartTime = AccordAgent.nonClashingStartTime(startTime, nodes, nodes.get(j), oneSecond, rnd);
+ assertTrue(nonClashingStartTime >= startTime);
+ startTimes[j] = nonClashingStartTime;
+ }
+
+ Arrays.sort(startTimes);
+ for (int j = 1 ; j < startTimes.length ; ++j)
+ {
+ long actualDelta = startTimes[j] - startTimes[j - 1];
+ assertTrue(Math.abs(targetDelta - actualDelta) <= 1);
+ }
+ }
+ });
+ }
+
+}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
index dde88ca468..036e513ae3 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
@@ -135,7 +135,6 @@ public class AccordCommandStoreTest
SimpleBitSet waitingOnApply = new SimpleBitSet(3);
waitingOnApply.set(1);
Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps, dependencies.directKeyDeps, new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
- attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt);
Command expected = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted,
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
index 70a66f2be4..8f01b8f685 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
@@ -41,7 +41,6 @@ import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
import accord.primitives.Timestamp;
@@ -100,9 +99,9 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
- PartialRoute<?> route = fullRoute.slice(fullRange(txn));
+ Route<?> route = fullRoute.slice(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
- PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, false, 1, partialTxn, fullRoute);
+ PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, fullRoute);
// Check preaccept
getUninterruptibly(commandStore.execute(preAccept, safeStore -> {
@@ -141,7 +140,7 @@ public class AccordCommandTest
builder.add(key, txnId2);
deps = builder.build();
}
- Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, false, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
+ Accept accept = Accept.SerializerSupport.create(txnId, route, 1, 1, Ballot.ZERO, executeAt, partialTxn.keys(), deps);
getUninterruptibly(commandStore.execute(accept, safeStore -> {
Command before = safeStore.ifInitialised(txnId).current();
@@ -192,9 +191,9 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
- PartialRoute<?> route = fullRoute.slice(fullRange(txn));
+ Route<?> route = fullRoute.slice(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
- PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, false, 1, partialTxn, fullRoute);
+ PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, fullRoute);
getUninterruptibly(commandStore.execute(preAccept1, safeStore -> {
persistDiff(commandStore, safeStore, txnId1, route, () -> {
@@ -204,7 +203,7 @@ public class AccordCommandTest
// second preaccept should identify txnId1 as a dependency
TxnId txnId2 = txnId(1, clock.incrementAndGet(), 1);
- PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, false, 1, partialTxn, fullRoute);
+ PreAccept preAccept2 = PreAccept.SerializerSupport.create(txnId2, route, 1, 1, 1, partialTxn, fullRoute);
getUninterruptibly(commandStore.execute(preAccept2, safeStore -> {
persistDiff(commandStore, safeStore, txnId2, route, () -> {
PreAccept.PreAcceptReply reply = preAccept2.apply(safeStore);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
index 1ece95f164..7cc2d081dc 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
@@ -32,12 +32,9 @@ import accord.api.Agent;
import accord.impl.AbstractFetchCoordinator;
import accord.impl.IntKey;
import accord.local.Node;
-import accord.messages.InformOfTxnId;
-import accord.messages.MessageType;
import accord.messages.ReadTxnData;
import accord.messages.Reply;
import accord.messages.Request;
-import accord.messages.SimpleReply;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
@@ -72,16 +69,6 @@ public class AccordMessageSinkTest
ClusterMetadataService.initializeForClients();
}
- @Test
- public void informOfTxn()
- {
- // There was an issue where the reply was the wrong verb
- // see CASSANDRA-18375
- InformOfTxnId request = Mockito.mock(InformOfTxnId.class);
- Mockito.when(request.type()).thenReturn(MessageType.INFORM_OF_TXN_REQ);
- checkRequestReplies(request, SimpleReply.Ok);
- }
-
@Test
public void bootstrapRead()
{
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordServiceTest.java
index f6918422e9..afdea31883 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordServiceTest.java
@@ -74,7 +74,7 @@ public class AccordServiceTest
throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);
case 4:
attempts++;
- throw new Exhausted(null, null);
+ throw new Exhausted(null, null, null);
case 5:
attempts++;
throw AccordService.newBarrierExhausted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);
@@ -124,7 +124,7 @@ public class AccordServiceTest
timeoutFailures.add(() -> {throw AccordService.newBarrierTimeout(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
timeoutFailures.add(() -> {throw new Preempted(null, null);});
timeoutFailures.add(() -> {throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
- timeoutFailures.add(() -> {throw new Exhausted(null, null);});
+ timeoutFailures.add(() -> {throw new Exhausted(null, null, null);});
Collections.shuffle(timeoutFailures, rs.asJdkRandom());
Iterator<Runnable> it = timeoutFailures.iterator();
Supplier<Seekables> failing = () -> {
@@ -162,7 +162,7 @@ public class AccordServiceTest
failures.add(() -> {throw AccordService.newBarrierTimeout(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
failures.add(() -> {throw new Preempted(null, null);});
failures.add(() -> {throw AccordService.newBarrierPreempted(TxnId.NONE, BarrierType.local, true, Ranges.EMPTY);});
- failures.add(() -> {throw new Exhausted(null, null);});
+ failures.add(() -> {throw new Exhausted(null, null, null);});
boolean isError = rs.nextBoolean();
failures.add(new Unexpected(isError));
Collections.shuffle(failures, rs.asJdkRandom());
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
index 54a311a67b..a2fd50d5f9 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
@@ -30,10 +30,13 @@ import java.util.function.LongSupplier;
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import javax.annotation.Nullable;
import com.google.common.collect.Sets;
+import accord.api.LocalListeners;
+import accord.api.ProgressLog.NoOpProgressLog;
+import accord.api.RemoteListeners;
+import accord.impl.DefaultLocalListeners;
import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -41,7 +44,6 @@ import org.apache.cassandra.io.util.File;
import org.junit.Assert;
import accord.api.Data;
-import accord.api.ProgressLog;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.impl.InMemoryCommandStore;
@@ -56,17 +58,14 @@ import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.SaveStatus;
-import accord.local.SaveStatus.LocalExecution;
import accord.primitives.Ballot;
import accord.primitives.FullKeyRoute;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
-import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Routable;
-import accord.primitives.Route;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
@@ -202,21 +201,6 @@ public class AccordTestUtils
Assert.assertEquals(val, safeState.current());
}
- public static final ProgressLog NOOP_PROGRESS_LOG = new ProgressLog()
- {
- @Override public void unwitnessed(TxnId txnId, ProgressShard progressShard) {}
- @Override public void preaccepted(Command command, ProgressShard progressShard) {}
- @Override public void accepted(Command command, ProgressShard progressShard) {}
- @Override public void precommitted(Command command) {}
- @Override public void stable(Command command, ProgressShard progressShard) {}
- @Override public void readyToExecute(Command command) {}
- @Override public void executed(Command command, ProgressShard progressShard) {}
- @Override public void clear(TxnId txnId) {}
- @Override public void durable(Command command) {}
- @Override public void waiting(SafeCommand blockedBy, LocalExecution blockedUntil, Route<?> blockedOnRoute, Participants<?> blockedOnParticipants) {}
- @Override public void waiting(TxnId blockedBy, LocalExecution blockedUntil, @Nullable Route<?> blockedOnRoute, @Nullable Participants<?> blockedOnParticipants) {}
- };
-
public static TxnId txnId(long epoch, long hlc, int node)
{
return txnId(epoch, hlc, node, Txn.Kind.Write);
@@ -390,11 +374,8 @@ public class AccordTestUtils
};
SingleEpochRanges holder = new SingleEpochRanges(Ranges.of(range));
- InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0,
- time,
- new AccordAgent(),
- null,
- cs -> null, holder);
+ InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0, time, new AccordAgent(),
+ null, null, cs -> null, holder);
holder.set(result);
return result;
}
@@ -425,7 +406,12 @@ public class AccordTestUtils
time,
new AccordAgent(),
null,
- cs -> NOOP_PROGRESS_LOG,
+ cs -> new NoOpProgressLog(),
+ cs -> new DefaultLocalListeners(new RemoteListeners.NoOpRemoteListeners(), new DefaultLocalListeners.NotifySink()
+ {
+ @Override public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener) {}
+ @Override public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, LocalListeners.ComplexListener listener) { return false; }
+ }),
holder,
journal,
loadExecutor,
diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
index 216889dd72..761ecddcad 100644
--- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
@@ -50,7 +50,7 @@ import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.ConfigurationService.EpochReady;
import accord.api.Scheduler;
-import accord.config.LocalConfig;
+import accord.api.LocalConfig;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Node;
@@ -352,7 +352,7 @@ public class EpochSyncTest
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Ranges ranges = tm.globalForEpoch(epoch).ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
- Assertions.assertThat(actual).describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.subtract(actual)).isEqualTo(ranges);
+ Assertions.assertThat(actual).describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual)).isEqualTo(ranges);
}
else
{
@@ -369,7 +369,7 @@ public class EpochSyncTest
if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.syncComplete(epoch - 1).mergeTouching()))
continue;
Assertions.assertThat(actual)
- .describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", id, epoch, ranges.subtract(actual), topology.nodes(),
+ .describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", id, epoch, ranges.without(actual), topology.nodes(),
LongStream.range(inst.epoch.getEpoch(), epoch + 1).mapToObj(e -> e + " -> " + conf.getEpochSnapshot(e).syncStatus + "(synced=" + globalSynced(e) + "): " + tm.syncComplete(e)).collect(Collectors.joining("\n")))
.isEqualTo(ranges);
}
diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
index 8f3192afa5..a20ee61a08 100644
--- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java
+++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
@@ -26,7 +26,6 @@ import com.google.common.annotations.VisibleForTesting;
import accord.api.Result;
import accord.local.Command;
import accord.local.CommonAttributes;
-import accord.local.Listeners;
import accord.local.SaveStatus;
import accord.local.Status;
import accord.primitives.Ballot;
@@ -97,8 +96,7 @@ public class MockJournal implements IJournal
ifNotEqual(before, after, Command::additionalKeysOrRanges, false),
new NewValue<>((k, deps) -> waitingOn),
- ifNotEqual(before, after, Command::writes, false),
- ifNotEqual(before, after, Command::durableListeners, true));
+ ifNotEqual(before, after, Command::writes, false));
}
static Command reconstructFromDiff(List<LoadedDiff> diffs)
@@ -131,7 +129,6 @@ public class MockJournal implements IJournal
SavedCommand.WaitingOnProvider waitingOnProvider = null;
Writes writes = null;
- Listeners.Immutable listeners = null;
for (LoadedDiff diff : diffs)
{
@@ -162,8 +159,6 @@ public class MockJournal implements IJournal
waitingOnProvider = diff.waitingOn.get();
if (diff.writes != null)
writes = diff.writes.get();
- if (diff.listeners != null)
- listeners = diff.listeners.get();
}
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
@@ -180,8 +175,6 @@ public class MockJournal implements IJournal
attrs.partialDeps(partialDeps);
if (additionalKeysOrRanges != null)
attrs.additionalKeysOrRanges(additionalKeysOrRanges);
- if (listeners != null && !listeners.isEmpty())
- attrs.setListeners(listeners);
Command.WaitingOn waitingOn = null;
if (waitingOnProvider != null)
@@ -292,7 +285,6 @@ public class MockJournal implements IJournal
public final NewValue<Seekables<?, ?>> additionalKeysOrRanges;
public final NewValue<Writes> writes;
- public final NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners;
public final NewValue<WaitingOnProvider> waitingOn;
public LoadedDiff(TxnId txnId,
@@ -309,8 +301,7 @@ public class MockJournal implements IJournal
NewValue<Seekables<?, ?>> additionalKeysOrRanges,
NewValue<SavedCommand.WaitingOnProvider> waitingOn,
- NewValue<Writes> writes,
- NewValue<Listeners.Immutable<Command.DurableAndIdempotentListener>> listeners)
+ NewValue<Writes> writes)
{
this.txnId = txnId;
this.executeAt = executeAt;
@@ -326,7 +317,6 @@ public class MockJournal implements IJournal
this.additionalKeysOrRanges = additionalKeysOrRanges;
this.writes = writes;
- this.listeners = listeners;
this.waitingOn = waitingOn;
}
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
index 96d8a84876..68eadd0d1b 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
@@ -28,6 +28,10 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.ToLongFunction;
+import accord.api.LocalListeners;
+import accord.api.ProgressLog;
+import accord.api.RemoteListeners;
+import accord.impl.DefaultLocalListeners;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Command;
@@ -36,6 +40,7 @@ import accord.local.CommandStores;
import accord.local.Node;
import accord.local.NodeTimeService;
import accord.local.PreLoadContext;
+import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.messages.BeginRecovery;
import accord.messages.PreAccept;
@@ -164,7 +169,12 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
},
null,
- ignore -> AccordTestUtils.NOOP_PROGRESS_LOG,
+ ignore -> new ProgressLog.NoOpProgressLog(),
+ cs -> new DefaultLocalListeners(new RemoteListeners.NoOpRemoteListeners(), new DefaultLocalListeners.NotifySink()
+ {
+ @Override public void notify(SafeCommandStore safeStore, SafeCommand safeCommand, TxnId listener) {}
+ @Override public boolean notify(SafeCommandStore safeStore, SafeCommand safeCommand, LocalListeners.ComplexListener listener) { return false; }
+ }),
updateHolder,
journal,
new AccordStateCacheMetrics("test"));
diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
index fc6921f8d4..aca8702454 100644
--- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
+import accord.primitives.Route;
import accord.utils.DefaultRandom;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
@@ -52,7 +53,6 @@ import accord.primitives.Ballot;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.PartialDeps;
-import accord.primitives.PartialRoute;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
@@ -212,8 +212,8 @@ public class AsyncOperationTest
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys(), COMMANDS), safe -> {
- CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToLog(commandStore));
- CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
+ CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
+ CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
}).beginAsResult());
@@ -253,16 +253,16 @@ public class AsyncOperationTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
- PartialRoute<?> partialRoute = route.slice(ranges);
+ Route<?> partialRoute = route.slice(ranges);
PartialDeps deps = PartialDeps.builder(ranges).build();
try
{
Command command = getUninterruptibly(commandStore.submit(contextFor(txnId, partialTxn.keys(), COMMANDS), safe -> {
- CheckedCommands.preaccept(safe, txnId, partialTxn, route, null, appendDiffToLog(commandStore));
- CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, executeAt, deps, appendDiffToLog(commandStore));
- CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
- CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, null, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
+ CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
+ CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), executeAt, deps, appendDiffToLog(commandStore));
+ CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
+ CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
}).beginAsResult());
diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
index c2a2b32af1..93d66eb1ec 100644
--- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
@@ -46,7 +46,6 @@ import accord.local.cfk.CommandsForKey.TxnInfo;
import accord.local.cfk.CommandsForKey.Unmanaged;
import accord.local.CommonAttributes;
import accord.local.CommonAttributes.Mutable;
-import accord.local.Listeners;
import accord.local.Node;
import accord.local.SaveStatus;
import accord.local.Status;
@@ -151,8 +150,9 @@ public class CommandsForKeySerializerTest
return Command.SerializerSupport.notDefined(attributes(), Ballot.ZERO);
case PreAccepted:
return Command.SerializerSupport.preaccepted(attributes(), executeAt, Ballot.ZERO);
- case Accepted:
case AcceptedInvalidate:
+ return Command.SerializerSupport.acceptedInvalidateWithoutDefinition(attributes(), ballot, ballot);
+ case Accepted:
case AcceptedWithDefinition:
case AcceptedInvalidateWithDefinition:
case PreCommittedWithDefinition:
@@ -185,7 +185,7 @@ public class CommandsForKeySerializerTest
case Erased:
case ErasedOrInvalidOrVestigial:
case Invalidated:
- return Command.SerializerSupport.invalidated(txnId, Listeners.Immutable.EMPTY);
+ return Command.SerializerSupport.invalidated(txnId);
}
}
@@ -356,7 +356,7 @@ public class CommandsForKeySerializerTest
@Test
public void serde()
{
- testOne(-669467611022826851L);
+ testOne(-8928257345122888710L);
Random random = new Random();
for (int i = 0 ; i < 10000 ; ++i)
{
@@ -508,7 +508,7 @@ public class CommandsForKeySerializerTest
for (int i = 0; i < info.length; i++)
{
InternalStatus status = rs.pick(InternalStatus.values());
- info[i] = TxnInfo.create(ids[i], status, ids[i], CommandsForKey.NO_TXNIDS, Ballot.ZERO);
+ info[i] = TxnInfo.create(ids[i], status, ids[i], TxnId.NO_TXNIDS, Ballot.ZERO);
}
Gen<Unmanaged.Pending> pendingGen = Gens.enums().allMixedDistribution(Unmanaged.Pending.class).next(rs);
@@ -537,7 +537,7 @@ public class CommandsForKeySerializerTest
}
else unmanaged = CommandsForKey.NO_PENDING_UNMANAGED;
- CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE);
+ CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
@@ -553,8 +553,8 @@ public class CommandsForKeySerializerTest
PartitionKey pk = new PartitionKey(TableId.fromString("1b255f4d-ef25-40a6-0000-000000000009"), key);
TxnId txnId = TxnId.fromValues(11,34052499,2,1);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk,
- new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, txnId, CommandsForKey.NO_TXNIDS, Ballot.ZERO) },
- CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE);
+ new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) },
+ CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, TxnId.NONE);
ByteBuffer buffer = CommandsForKeySerializer.toBytesWithoutKey(expected);
CommandsForKey roundTrip = CommandsForKeySerializer.fromBytes(pk, buffer);
diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java
index 492b45a79e..58042e2d5f 100644
--- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java
@@ -29,7 +29,6 @@ import java.util.stream.Stream;
import accord.local.Command;
import accord.local.CommonAttributes;
-import accord.local.Listeners;
import accord.local.RedundantBefore;
import accord.local.SaveStatus;
import accord.primitives.Ballot;
@@ -262,7 +261,7 @@ public class AccordGenerators
case Erased:
case ErasedOrInvalidOrVestigial:
case Invalidated:
- return Command.SerializerSupport.invalidated(txnId, Listeners.Immutable.EMPTY);
+ return Command.SerializerSupport.invalidated(txnId);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org