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/10/02 09:51:09 UTC
(cassandra) branch cep-15-accord updated (a9e54351e7 -> 55a2272f52)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a change to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
discard a9e54351e7 Follow-up to CASSANDRA-19967 and CASSANDRA-19869
new 55a2272f52 Follow-up to CASSANDRA-19967 and CASSANDRA-19869
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (a9e54351e7)
\
N -- N -- N refs/heads/cep-15-accord (55a2272f52)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 5 ++--
.../db/compaction/CompactionIterator.java | 4 ++-
.../distributed/test/accord/AccordLoadTest.java | 33 +++++++++++++++++-----
4 files changed, 33 insertions(+), 11 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/01: Follow-up to CASSANDRA-19967 and CASSANDRA-19869
Posted by be...@apache.org.
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
commit 55a2272f5288da44ff58d717ea5a17b69b44a0f3
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Tue Oct 1 10:29:45 2024 +0200
Follow-up to CASSANDRA-19967 and CASSANDRA-19869
patch by Alex Petrov, Ariel Weisberg, Benedict Elliott Smith, Blake Eggleston and David Capwell
---
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 17 +-
.../db/compaction/CompactionIterator.java | 79 ++++---
.../cassandra/db/partitions/PartitionUpdate.java | 24 ++
.../org/apache/cassandra/journal/OnDiskIndex.java | 2 +-
src/java/org/apache/cassandra/net/Verb.java | 2 +-
.../service/accord/AccordCachingState.java | 9 +-
.../service/accord/AccordCommandStore.java | 241 +++++++++++----------
.../service/accord/AccordCommandStores.java | 52 ++++-
.../service/accord/AccordConfigurationService.java | 16 +-
.../cassandra/service/accord/AccordJournal.java | 63 ++++--
.../accord/AccordJournalValueSerializers.java | 2 +-
.../cassandra/service/accord/AccordKeyspace.java | 2 -
.../service/accord/AccordObjectSizes.java | 2 +-
.../cassandra/service/accord/AccordService.java | 9 +-
.../cassandra/service/accord/AccordStateCache.java | 19 +-
.../cassandra/service/accord/AccordTopology.java | 2 +
.../service/accord/AccordVerbHandler.java | 2 -
.../apache/cassandra/service/accord/IJournal.java | 2 +-
.../cassandra/service/accord/JournalKey.java | 134 +++++-------
.../cassandra/service/accord/SavedCommand.java | 113 ++++++----
.../service/accord/async/AsyncOperation.java | 1 -
.../cassandra/tcm/AtomicLongBackedProcessor.java | 59 ++++-
.../cassandra/tcm/ClusterMetadataService.java | 2 +-
src/java/org/apache/cassandra/tcm/Startup.java | 6 +-
.../org/apache/cassandra/tcm/log/LocalLog.java | 5 +
.../org/apache/cassandra/tcm/log/LogReader.java | 6 +
.../distributed/test/accord/AccordLoadTest.java | 40 +++-
.../test/accord/AccordMigrationRaceTestBase.java | 5 +-
.../unit/org/apache/cassandra/ServerTestUtils.java | 6 +-
.../config/YamlConfigurationLoaderTest.java | 17 ++
.../compaction/CompactionAccordIteratorsTest.java | 6 +-
.../service/accord/AccordCommandTest.java | 4 +-
.../accord/AccordConfigurationServiceTest.java | 136 ++++++++----
.../service/accord/AccordJournalOrderTest.java | 2 +-
.../cassandra/service/accord/AccordTestUtils.java | 12 +-
.../cassandra/service/accord/MockJournal.java | 2 +-
.../cassandra/service/accord/SavedCommandTest.java | 4 +-
.../accord/SimulatedAccordCommandStore.java | 12 +-
.../service/accord/SimulatedDepsTest.java | 62 ++++--
.../SimulatedRandomKeysWithRangeConflictTest.java | 104 +++++----
.../service/accord/async/AsyncLoaderTest.java | 4 +-
.../service/accord/async/AsyncOperationTest.java | 14 +-
.../tcm/ValidatingClusterMetadataService.java | 63 ++++++
44 files changed, 889 insertions(+), 477 deletions(-)
diff --git a/modules/accord b/modules/accord
index e2e72287a7..4cf0070d60 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit e2e72287a70bfff4cf6a5639d4a60a0c9e8086b4
+Subproject commit 4cf0070d604abd2db460a5f1c3f8cd8dc7d26696
diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java
index 0350e587c6..b99b06ddab 100644
--- a/src/java/org/apache/cassandra/config/AccordSpec.java
+++ b/src/java/org/apache/cassandra/config/AccordSpec.java
@@ -63,12 +63,12 @@ public class AccordSpec
public DurationSpec.IntMillisecondsBound range_barrier_timeout = new DurationSpec.IntMillisecondsBound("2m");
- public volatile DurationSpec fast_path_update_delay = new DurationSpec.IntSecondsBound(5);
+ public volatile DurationSpec.IntSecondsBound fast_path_update_delay = new DurationSpec.IntSecondsBound(5);
- public volatile DurationSpec schedule_durability_frequency = new DurationSpec.IntSecondsBound(5);
- public volatile DurationSpec durability_txnid_lag = new DurationSpec.IntSecondsBound(5);
- public volatile DurationSpec shard_durability_cycle = new DurationSpec.IntMinutesBound(1);
- public volatile DurationSpec global_durability_cycle = new DurationSpec.IntMinutesBound(10);
+ public volatile DurationSpec.IntSecondsBound schedule_durability_frequency = new DurationSpec.IntSecondsBound(5);
+ public volatile DurationSpec.IntSecondsBound durability_txnid_lag = new DurationSpec.IntSecondsBound(5);
+ public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(1, TimeUnit.MINUTES);
+ public volatile DurationSpec.IntSecondsBound global_durability_cycle = new DurationSpec.IntSecondsBound(10, TimeUnit.MINUTES);
public enum TransactionalRangeMigration
{
@@ -87,7 +87,7 @@ public class AccordSpec
* default transactional mode for tables created by this node when no transactional mode has been specified in the DDL
*/
public TransactionalMode default_transactional_mode = TransactionalMode.off;
- public boolean ephemeralReadEnabled = false;
+ public boolean ephemeralReadEnabled = true;
public boolean state_cache_listener_jfr_enabled = true;
public final JournalSpec journal = new JournalSpec();
public final MinEpochRetrySpec minEpochSyncRetry = new MinEpochRetrySpec();
@@ -102,11 +102,12 @@ public class AccordSpec
public static class JournalSpec implements Params
{
- public int segmentSize = 32 << 20;
+ public int segmentSize = 32 << 10;
public FailurePolicy failurePolicy = FailurePolicy.STOP;
public FlushMode flushMode = FlushMode.PERIODIC;
public DurationSpec.IntMillisecondsBound flushPeriod; // pulls default from 'commitlog_sync_period'
public DurationSpec.IntMillisecondsBound periodicFlushLagBlock = new DurationSpec.IntMillisecondsBound("1500ms");
+ public DurationSpec.IntMillisecondsBound compactionPeriod = new DurationSpec.IntMillisecondsBound("60000ms");
@Override
public int segmentSize()
@@ -135,7 +136,7 @@ public class AccordSpec
@Override
public int compactionPeriodMillis()
{
- return 60_000;
+ return compactionPeriod.toMilliseconds();
}
@JsonIgnore
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 17160c4fa8..3e9a1de462 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -32,6 +32,9 @@ import javax.annotation.Nonnull;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Ordering;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import accord.local.Cleanup;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
@@ -43,6 +46,7 @@ import accord.primitives.Status;
import accord.primitives.Status.Durability;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
+import accord.utils.Invariants;
import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.ColumnIdentifier;
@@ -104,8 +108,8 @@ import org.apache.cassandra.service.paxos.PaxosRepairHistory;
import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
import org.apache.cassandra.utils.TimeUUID;
+import static accord.local.Cleanup.ERASE;
import static accord.local.Cleanup.TRUNCATE_WITH_OUTCOME;
-import static accord.local.Cleanup.shouldCleanup;
import static accord.local.Cleanup.shouldCleanupPartial;
import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
@@ -142,6 +146,7 @@ import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeTime
*/
public class CompactionIterator extends CompactionInfo.Holder implements UnfilteredPartitionIterator
{
+ private static final Logger logger = LoggerFactory.getLogger(CompactionIterator.class);
private static final long UNFILTERED_TO_UPDATE_PROGRESS = 100;
private final OperationType type;
@@ -1057,13 +1062,17 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
try
{
- PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey());
-
+ List<Row> rows = new ArrayList<>();
while (partition.hasNext())
- applyToRow((Row) partition.next());
+ {
+ Row row = (Row) partition.next();
+ rows.add(row);
+ collect(row);
+ }
if (key.type != JournalKey.Type.COMMAND_DIFF)
{
+ PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey());
try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
{
serializer.reserialize(key, builder, out, userVersion);
@@ -1080,49 +1089,32 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
SavedCommand.Builder commandBuilder = (SavedCommand.Builder) builder;
-
- // Do not have txnId in selected SSTables; remove
- if (commandBuilder.txnId() == null)
- return newVersion.build().unfilteredIterator();
+ if (commandBuilder.isEmpty())
+ {
+ Invariants.checkState(rows.isEmpty());
+ return partition;
+ }
RedundantBefore redundantBefore = redundantBefores.get(key.commandStoreId);
+ Cleanup cleanup = commandBuilder.shouldCleanup(redundantBefore, durableBefore);
+ if (cleanup == ERASE)
+ return PartitionUpdate.fullPartitionDelete(metadata(), partition.partitionKey(), maxSeenTimestamp, nowInSec).unfilteredIterator();
- Cleanup cleanup = shouldCleanup(commandBuilder.txnId(), commandBuilder.saveStatus(),
- commandBuilder.durability(), commandBuilder.participants(),
- redundantBefore, durableBefore);
- switch (cleanup)
+ commandBuilder = commandBuilder.maybeCleanup(cleanup);
+ if (commandBuilder != builder)
{
- case EXPUNGE:
+ if (commandBuilder == null)
return null;
- case EXPUNGE_PARTIAL:
- newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey());
- commandBuilder = commandBuilder.expungePartial();
-
- newVersion.row(lastClustering)
- .add(recordColumn.name.toString(), commandBuilder.asByteBuffer(userVersion));
-
- return newVersion.build().unfilteredIterator();
-
- case ERASE:
- return PartitionUpdate.fullPartitionDelete(metadata(), partition.partitionKey(), maxSeenTimestamp, nowInSec).unfilteredIterator();
-
- case VESTIGIAL:
- case INVALIDATE:
- case TRUNCATE_WITH_OUTCOME:
- case TRUNCATE:
- newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey());
- commandBuilder = commandBuilder.saveStatusOnly();
-
- newVersion.row(lastClustering)
- .add(recordColumn.name.toString(), commandBuilder.asByteBuffer(userVersion));
-
- return newVersion.build().unfilteredIterator();
+ PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey());
+ newVersion.row(lastClustering)
+ .add("record", commandBuilder.asByteBuffer(userVersion))
+ .add("user_version", userVersion);
+ return newVersion.build().unfilteredIterator();
+ }
- case NO:
- return newVersion.build().unfilteredIterator();
- default:
- throw new IllegalStateException("Unknown cleanup: " + cleanup);}
+ return PartitionUpdate.multiRowUpdate(AccordKeyspace.Journal, partition.partitionKey(), rows)
+ .unfilteredIterator();
}
catch (IOException e)
{
@@ -1132,6 +1124,11 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
@Override
protected Row applyToRow(Row row)
+ {
+ return row;
+ }
+
+ protected void collect(Row row)
{
updateProgress();
maxSeenTimestamp = row.primaryKeyLivenessInfo().timestamp();
@@ -1146,7 +1143,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
{
throw new RuntimeException(e);
}
- return null;
}
@Override
@@ -1157,7 +1153,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
}
-
private static class AbortableUnfilteredPartitionTransformation extends Transformation<UnfilteredRowIterator>
{
private final AbortableUnfilteredRowTransformation abortableIter;
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index c365b46e45..12b22882cf 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
@@ -198,6 +198,30 @@ public class PartitionUpdate extends AbstractBTreePartition
return singleRowUpdate(metadata, key, row.isStatic() ? null : row, row.isStatic() ? row : null);
}
+ /**
+ * Creates an immutable partition update that contains a single row update.
+ *
+ * @param metadata the metadata for the created update.
+ * @param key the partition key for the partition to update.
+ * @param rows the rows for the update (may not be static).
+ *
+ * @return the newly created partition update containing only {@code row}.
+ */
+ public static PartitionUpdate multiRowUpdate(TableMetadata metadata, DecoratedKey key, List<Row> rows)
+ {
+ if (rows.isEmpty())
+ return emptyUpdate(metadata, key);
+ MutableDeletionInfo deletionInfo = MutableDeletionInfo.live();
+ Columns columns = Columns.NONE;
+ for (Row row : rows)
+ columns = columns.mergeTo(Columns.from(row));
+
+ BTreePartitionData holder = new BTreePartitionData(new RegularAndStaticColumns(Columns.NONE, columns),
+ BTree.build(rows), deletionInfo, Rows.EMPTY_STATIC_ROW,
+ EncodingStats.NO_STATS);
+ return new PartitionUpdate(metadata, metadata.epoch, key, holder, deletionInfo, false);
+ }
+
/**
* Creates an immutable partition update that contains a single row update.
*
diff --git a/src/java/org/apache/cassandra/journal/OnDiskIndex.java b/src/java/org/apache/cassandra/journal/OnDiskIndex.java
index fe2c2713b9..b2f1487ad0 100644
--- a/src/java/org/apache/cassandra/journal/OnDiskIndex.java
+++ b/src/java/org/apache/cassandra/journal/OnDiskIndex.java
@@ -356,7 +356,7 @@ final class OnDiskIndex<K> extends Index<K>
int low = 0, mid = entryCount, high = mid - 1, result = -1;
while (low <= high)
{
- mid = (low + high) >> 1;
+ mid = (low + high) >>> 1;
result = compareWithKeyAt(key, mid);
if (result > 0)
{
diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java
index c656ccc527..87b366300f 100644
--- a/src/java/org/apache/cassandra/net/Verb.java
+++ b/src/java/org/apache/cassandra/net/Verb.java
@@ -331,7 +331,7 @@ public enum Verb
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_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_CALCULATE_DEPS_REQ (144, P2, longTimeout, 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 35ceec450d..50a48be1fb 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCachingState.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCachingState.java
@@ -64,14 +64,14 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
int references = 0;
int lastQueriedEstimatedSizeOnHeap = 0;
- final byte index;
+ final int index;
private boolean shouldUpdateSize;
AccordCachingState(K key, int index)
{
this.key = key;
- Invariants.checkArgument(index >= 0 && index <= Byte.MAX_VALUE);
- this.index = (byte) index;
+ Invariants.checkArgument(index >= 0);
+ this.index = index;
//noinspection unchecked
this.state = (State<K, V>) Uninitialized.instance;
}
@@ -79,8 +79,7 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
private AccordCachingState(K key, int index, State<K, V> state)
{
this.key = key;
- Invariants.checkArgument(index >= 0 && index <= Byte.MAX_VALUE);
- this.index = (byte) index;
+ this.index = index;
this.state = state;
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index e5a888954f..a2367e5768 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -26,9 +26,11 @@ import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
+import java.util.function.IntFunction;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
@@ -40,7 +42,6 @@ import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.api.RoutingKey;
-import accord.local.cfk.CommandsForKey;
import accord.impl.TimestampsForKey;
import accord.local.Cleanup;
import accord.local.Command;
@@ -54,6 +55,7 @@ import accord.local.PreLoadContext;
import accord.local.RedundantBefore;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
+import accord.local.cfk.CommandsForKey;
import accord.primitives.Deps;
import accord.primitives.Participants;
import accord.primitives.Ranges;
@@ -66,12 +68,10 @@ import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import org.apache.cassandra.cache.CacheSize;
-import org.apache.cassandra.concurrent.ExecutorPlus;
-import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.SequentialExecutorPlus;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Mutation;
-import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.async.AsyncOperation;
import org.apache.cassandra.service.accord.events.CacheEvents;
@@ -87,36 +87,15 @@ import static accord.primitives.Status.PreApplied;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
import static accord.utils.Invariants.checkState;
-import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
-public class AccordCommandStore extends CommandStore implements CacheSize
+public class AccordCommandStore extends CommandStore
{
private static final Logger logger = LoggerFactory.getLogger(AccordCommandStore.class);
private static final boolean CHECK_THREADS = CassandraRelevantProperties.TEST_ACCORD_STORE_THREAD_CHECKS_ENABLED.getBoolean();
- private static long getThreadId(ExecutorService executor)
- {
- if (!CHECK_THREADS)
- return 0;
- try
- {
- return executor.submit(() -> Thread.currentThread().getId()).get();
- }
- catch (InterruptedException e)
- {
- throw new AssertionError(e);
- }
- catch (ExecutionException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- private final long threadId;
public final String loggingId;
private final IJournal journal;
- private final ExecutorService executor;
- private final AccordStateCache stateCache;
+ private final CommandStoreExecutor executor;
private final AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache;
private final AccordStateCache.Instance<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKeyCache;
private final AccordStateCache.Instance<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKeyCache;
@@ -125,29 +104,6 @@ public class AccordCommandStore extends CommandStore implements CacheSize
private long lastSystemTimestampMicros = Long.MIN_VALUE;
private final CommandsForRangesLoader commandsForRangesLoader;
- public AccordCommandStore(int id,
- NodeTimeService time,
- Agent agent,
- DataStore dataStore,
- ProgressLog.Factory progressLogFactory,
- LocalListeners.Factory listenerFactory,
- EpochUpdateHolder epochUpdateHolder,
- IJournal journal,
- AccordStateCacheMetrics 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)
{
if (!DatabaseDescriptor.getAccordStateCacheListenerJFREnabled())
@@ -215,7 +171,6 @@ public class AccordCommandStore extends CommandStore implements CacheSize
event.update();
}
- @VisibleForTesting
public AccordCommandStore(int id,
NodeTimeService time,
Agent agent,
@@ -224,16 +179,13 @@ public class AccordCommandStore extends CommandStore implements CacheSize
LocalListeners.Factory listenerFactory,
EpochUpdateHolder epochUpdateHolder,
IJournal journal,
- ExecutorPlus loadExecutor,
- ExecutorPlus saveExecutor,
- AccordStateCacheMetrics cacheMetrics)
+ CommandStoreExecutor commandStoreExecutor)
{
super(id, time, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
this.journal = journal;
loggingId = String.format("[%s]", id);
- executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']');
- threadId = getThreadId(executor);
- stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20, cacheMetrics);
+ executor = commandStoreExecutor;
+ AccordStateCache stateCache = executor.stateCache;
commandCache =
stateCache.instance(TxnId.class,
AccordSafeCommand.class,
@@ -275,10 +227,10 @@ public class AccordCommandStore extends CommandStore implements CacheSize
executor.execute(() -> CommandStore.register(this));
}
- static Factory factory(AccordJournal journal, AccordStateCacheMetrics cacheMetrics)
+ static Factory factory(AccordJournal journal, IntFunction<CommandStoreExecutor> executorFactory)
{
return (id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch) ->
- new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, cacheMetrics);
+ new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, executorFactory.apply(id));
}
public CommandsForRangesLoader diskCommandsForRanges()
@@ -295,34 +247,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
@Override
public boolean inStore()
{
- if (!CHECK_THREADS)
- return true;
- return Thread.currentThread().getId() == threadId;
- }
-
- @Override
- public void setCapacity(long bytes)
- {
- checkInStoreThread();
- stateCache.setCapacity(bytes);
- }
-
- @Override
- public long capacity()
- {
- return stateCache.capacity();
- }
-
- @Override
- public int size()
- {
- return stateCache.size();
- }
-
- @Override
- public long weightedSize()
- {
- return stateCache.weightedSize();
+ return executor.isInThread();
}
public void checkInStoreThread()
@@ -339,7 +264,15 @@ public class AccordCommandStore extends CommandStore implements CacheSize
public ExecutorService executor()
{
- return executor;
+ return executor.delegate();
+ }
+
+ /**
+ * Note that this cache is shared with other commandStores!
+ */
+ public AccordStateCache cache()
+ {
+ return executor.cache();
}
public AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache()
@@ -443,18 +376,6 @@ public class AccordCommandStore extends CommandStore implements CacheSize
return null != mutation ? mutation::applyUnsafe : null;
}
- @VisibleForTesting
- public AccordStateCache cache()
- {
- return stateCache;
- }
-
- @VisibleForTesting
- public void unsafeClearCache()
- {
- stateCache.unsafeClear();
- }
-
public void setCurrentOperation(AsyncOperation<?> operation)
{
checkState(currentOperation == null);
@@ -487,7 +408,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
@Override
public <T> AsyncChain<T> submit(Callable<T> task)
{
- return AsyncChains.ofCallable(executor, task);
+ return AsyncChains.ofCallable(executor.delegate(), task);
}
public DataStore dataStore()
@@ -570,15 +491,6 @@ public class AccordCommandStore extends CommandStore implements CacheSize
@Override
public void shutdown()
{
- executor.shutdown();
- try
- {
- executor.awaitTermination(20, TimeUnit.SECONDS);
- }
- catch (InterruptedException t)
- {
- throw new RuntimeException("Could not shut down command store " + this);
- }
}
public void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore)
@@ -633,7 +545,7 @@ public class AccordCommandStore extends CommandStore implements CacheSize
@VisibleForTesting
public Command loadCommand(TxnId txnId)
{
- return journal.loadCommand(id, txnId);
+ return journal.loadCommand(id, txnId, redundantBefore(), durableBefore());
}
public interface Loader
@@ -777,4 +689,111 @@ public class AccordCommandStore extends CommandStore implements CacheSize
});
}
}
+
+ public static class CommandStoreExecutor implements CacheSize
+ {
+ final AccordStateCache stateCache;
+ final SequentialExecutorPlus delegate;
+ final long threadId;
+
+ CommandStoreExecutor(AccordStateCache stateCache, SequentialExecutorPlus delegate, long threadId)
+ {
+ this.stateCache = stateCache;
+ this.delegate = delegate;
+ this.threadId = threadId;
+ }
+
+ CommandStoreExecutor(AccordStateCache stateCache, SequentialExecutorPlus delegate)
+ {
+ this.stateCache = stateCache;
+ this.delegate = delegate;
+ this.threadId = getThreadId();
+ }
+
+ public boolean isInThread()
+ {
+ if (!CHECK_THREADS)
+ return true;
+
+ return threadId == Thread.currentThread().getId();
+ }
+
+ public void shutdown()
+ {
+ delegate.shutdown();
+ }
+
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
+ {
+ return delegate.awaitTermination(timeout, unit);
+ }
+
+ public Future<?> submit(Runnable task)
+ {
+ return delegate.submit(task);
+ }
+
+ public ExecutorService delegate()
+ {
+ return delegate;
+ }
+
+ public void execute(Runnable command)
+ {
+ delegate.submit(command);
+ }
+
+ private long getThreadId()
+ {
+ try
+ {
+ return delegate.submit(() -> Thread.currentThread().getId()).get();
+ }
+ catch (InterruptedException e)
+ {
+ throw new AssertionError(e);
+ }
+ catch (ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @VisibleForTesting
+ public AccordStateCache cache()
+ {
+ return stateCache;
+ }
+
+ @VisibleForTesting
+ public void unsafeClearCache()
+ {
+ stateCache.unsafeClear();
+ }
+
+ @Override
+ public void setCapacity(long bytes)
+ {
+ Invariants.checkState(isInThread());
+ stateCache.setCapacity(bytes);
+ }
+
+ @Override
+ public long capacity()
+ {
+ return stateCache.capacity();
+ }
+
+ @Override
+ public int size()
+ {
+ return stateCache.size();
+ }
+
+ @Override
+ public long weightedSize()
+ {
+ return stateCache.weightedSize();
+ }
+ }
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
index 60cbcf84ff..bc34a7ad50 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
@@ -17,6 +17,7 @@
*/
package org.apache.cassandra.service.accord;
+import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import accord.api.Agent;
@@ -24,6 +25,7 @@ import accord.api.ConfigurationService.EpochReady;
import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
+import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Node;
import accord.local.NodeTimeService;
@@ -32,33 +34,48 @@ import accord.primitives.Range;
import accord.topology.Topology;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
+import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
import org.apache.cassandra.metrics.CacheSizeMetrics;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordCommandStore.CommandStoreExecutor;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+
public class AccordCommandStores extends CommandStores implements CacheSize
{
public static final String ACCORD_STATE_CACHE = "AccordStateCache";
private final CacheSizeMetrics cacheSizeMetrics;
+ private final CommandStoreExecutor[] executors;
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenerFactory,
- AccordJournal journal)
+ AccordJournal journal, CommandStoreExecutor[] executors)
{
super(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory,
- AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE)));
+ AccordCommandStore.factory(journal, id -> executors[id % executors.length]));
setCapacity(DatabaseDescriptor.getAccordCacheSizeInMiB() << 20);
+ this.executors = executors;
this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this);
}
static Factory factory(AccordJournal journal)
{
- return (time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory) ->
- new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory, journal);
+ return (time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory) -> {
+ CommandStoreExecutor[] executors = new CommandStoreExecutor[DatabaseDescriptor.getAccordShardCount()];
+ for (int id = 0; id < executors.length; id++)
+ {
+ AccordStateCacheMetrics metrics = new AccordStateCacheMetrics(ACCORD_STATE_CACHE);
+ AccordStateCache stateCache = new AccordStateCache(Stage.READ.executor(), Stage.MUTATION.executor(), 8 << 20, metrics);
+ executors[id] = new CommandStoreExecutor(stateCache, executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']'));
+ }
+
+ return new AccordCommandStores(time, agent, store, random, shardDistributor, progressLogFactory, listenerFactory, journal, executors);
+ };
}
@Override
@@ -96,22 +113,29 @@ public class AccordCommandStores extends CommandStores implements CacheSize
@Override
public int size()
{
- return unsafeFoldLeft(0, (size, commandStore) -> size + ((AccordCommandStore) commandStore).size());
+ int size = 0;
+ for (CommandStoreExecutor executor : executors)
+ size += executor.size();
+ return size;
}
@Override
public long weightedSize()
{
- return unsafeFoldLeft(0L, (size, commandStore) -> size + ((AccordCommandStore) commandStore).weightedSize());
+ long size = 0;
+ for (CommandStoreExecutor executor : executors)
+ size += executor.weightedSize();
+ return size;
}
synchronized void refreshCacheSizes()
{
if (count() == 0)
return;
- long perStore = cacheSize / count();
+ long perExecutor = cacheSize / executors.length;
// TODO (low priority, safety): we might transiently breach our limit if we increase one store before decreasing another
- forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCapacity(perStore));
+ for (CommandStoreExecutor executor : executors)
+ executor.execute(() -> executor.setCapacity(perExecutor));
}
@Override
@@ -134,6 +158,18 @@ public class AccordCommandStores extends CommandStores implements CacheSize
public synchronized void shutdown()
{
super.shutdown();
+ for (CommandStoreExecutor executor : executors)
+ {
+ executor.shutdown();
+ try
+ {
+ executor.awaitTermination(1, TimeUnit.MINUTES);
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
//TODO shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface)
}
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
index f52d7a935e..0a2a192aa1 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
@@ -412,15 +412,25 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
protected void fetchTopologyInternal(long epoch)
{
ClusterMetadata metadata = ClusterMetadata.current();
- if (metadata.directory.peerIds().size() < 2)
+ if (metadata.directory.peerIds().isEmpty())
return; // just let CMS handle it when it's ready
// TODO (desired): randomise
NodeId first = metadata.directory.peerIds().first();
InetAddressAndPort peer = metadata.directory.getNodeAddresses(first).broadcastAddress;
if (FBUtilities.getBroadcastAddressAndPort().equals(peer))
- peer = metadata.directory.getNodeAddresses(metadata.directory.peerIds().higher(first)).broadcastAddress;;
- ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(metadata, peer, Epoch.create(epoch));
+ {
+ NodeId second = metadata.directory.peerIds().higher(first);
+ if (second == null)
+ return;
+
+ peer = metadata.directory.getNodeAddresses(second).broadcastAddress;
+ }
+ ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(metadata, peer, Epoch.create(epoch))
+ .addCallback((success, fail) -> {
+ if (fail != null)
+ fetchTopologyInternal(epoch);
+ });
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index 52b65a33a2..b1cd3ee19e 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -32,6 +32,8 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.impl.ErasedSafeCommand;
+import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
@@ -57,8 +59,10 @@ import org.apache.cassandra.journal.ValueSerializer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsAccumulator;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator;
+import org.apache.cassandra.service.accord.JournalKey.JournalKeySupport;
import org.apache.cassandra.utils.ExecutorUtils;
+import static accord.primitives.SaveStatus.ErasedOrVestigial;
import static accord.primitives.Status.Truncated;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.RedundantBeforeAccumulator;
@@ -78,7 +82,7 @@ public class AccordJournal implements IJournal, Shutdownable
private static final Set<Integer> SENTINEL_HOSTS = Collections.singleton(0);
- static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[23]);
+ static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[JournalKeySupport.TOTAL_SIZE]);
private final Journal<JournalKey, Object> journal;
private final AccordJournalTable<JournalKey, Object> journalTable;
@@ -165,50 +169,59 @@ public class AccordJournal implements IJournal, Shutdownable
}
@Override
- public Command loadCommand(int commandStoreId, TxnId txnId)
+ public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
- return loadDiffs(commandStoreId, txnId).construct();
+ SavedCommand.Builder builder = loadDiffs(commandStoreId, txnId);
+ Cleanup cleanup = builder.shouldCleanup(redundantBefore, durableBefore);
+ switch (cleanup)
+ {
+ case EXPUNGE_PARTIAL:
+ case EXPUNGE:
+ case ERASE:
+ return ErasedSafeCommand.erased(txnId, ErasedOrVestigial);
+ }
+ return builder.construct();
}
@VisibleForTesting
public RedundantBefore loadRedundantBefore(int store)
{
- RedundantBeforeAccumulator accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.REDUNDANT_BEFORE, store));
+ RedundantBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store));
return accumulator.get();
}
@Override
public DurableBefore loadDurableBefore(int store)
{
- DurableBeforeAccumulator accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.DURABLE_BEFORE, store));
+ DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, store));
return accumulator.get();
}
@Override
public NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int store)
{
- IdentityAccumulator<NavigableMap<TxnId, Ranges>> accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store));
+ IdentityAccumulator<NavigableMap<TxnId, Ranges>> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store));
return accumulator.get();
}
@Override
public NavigableMap<Timestamp, Ranges> loadSafeToRead(int store)
{
- IdentityAccumulator<NavigableMap<Timestamp, Ranges>> accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.SAFE_TO_READ, store));
+ IdentityAccumulator<NavigableMap<Timestamp, Ranges>> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store));
return accumulator.get();
}
@Override
public CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int store)
{
- IdentityAccumulator<RangesForEpoch.Snapshot> accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store));
+ IdentityAccumulator<RangesForEpoch.Snapshot> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store));
return accumulator.get();
}
@Override
public List<Deps> loadHistoricalTransactions(int store)
{
- HistoricalTransactionsAccumulator accumulator = readAll(new JournalKey(Timestamp.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store));
+ HistoricalTransactionsAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store));
return accumulator.get();
}
@@ -235,17 +248,17 @@ public class AccordJournal implements IJournal, Shutdownable
RecordPointer pointer = null;
// TODO: avoid allocating keys
if (fieldUpdates.redundantBefore != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), fieldUpdates.redundantBefore);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.REDUNDANT_BEFORE, store), fieldUpdates.redundantBefore);
if (fieldUpdates.durableBefore != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.DURABLE_BEFORE, store), fieldUpdates.durableBefore);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, store), fieldUpdates.durableBefore);
if (fieldUpdates.bootstrapBeganAt != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), fieldUpdates.bootstrapBeganAt);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.BOOTSTRAP_BEGAN_AT, store), fieldUpdates.bootstrapBeganAt);
if (fieldUpdates.safeToRead != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.SAFE_TO_READ, store), fieldUpdates.safeToRead);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store), fieldUpdates.safeToRead);
if (fieldUpdates.rangesForEpoch != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), fieldUpdates.rangesForEpoch);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), fieldUpdates.rangesForEpoch);
if (fieldUpdates.historicalTransactions != null)
- pointer = appendInternal(new JournalKey(Timestamp.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store), fieldUpdates.historicalTransactions);
+ pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store), fieldUpdates.historicalTransactions);
if (onFlush == null)
return;
@@ -260,11 +273,23 @@ public class AccordJournal implements IJournal, Shutdownable
public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId)
{
JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
- SavedCommand.Builder builder = new SavedCommand.Builder();
+ SavedCommand.Builder builder = new SavedCommand.Builder(txnId);
journalTable.readAll(key, builder::deserializeNext);
return builder;
}
+ public List<SavedCommand.Builder> loadSeparateDiffs(int commandStoreId, TxnId txnId)
+ {
+ JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
+ List<SavedCommand.Builder> builders = new ArrayList<>();
+ journalTable.readAll(key, (in, version) -> {
+ SavedCommand.Builder builder = new SavedCommand.Builder(txnId);
+ builder.deserializeNext(in, version);
+ builders.add(builder);
+ });
+ return builders;
+ }
+
private <BUILDER> BUILDER readAll(JournalKey key)
{
BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key);
@@ -313,7 +338,7 @@ public class AccordJournal implements IJournal, Shutdownable
public void replay()
{
- // TODO: optimize replay memory footprint
+ // TODO (expected): optimize replay memory footprint
class ToApply
{
final JournalKey key;
@@ -331,11 +356,11 @@ public class AccordJournal implements IJournal, Shutdownable
{
isReplay.set(true);
- JournalKey key = null;
+ JournalKey key;
SavedCommand.Builder builder = new SavedCommand.Builder();
while ((key = iter.key()) != null)
{
- builder.clear();
+ builder.reset(key.id);
if (key.type != JournalKey.Type.COMMAND_DIFF)
{
// TODO (required): add "skip" for the key to avoid getting stuck
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
index f06ccd4c4f..289b5c1b98 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
@@ -63,7 +63,7 @@ public class AccordJournalValueSerializers
@Override
public SavedCommand.Builder mergerFor(JournalKey journalKey)
{
- return new SavedCommand.Builder();
+ return new SavedCommand.Builder(journalKey.id);
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index f39fdbf902..ca1326156e 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -1395,7 +1395,6 @@ public class AccordKeyspace
"SET remote_sync_complete = remote_sync_complete + ? WHERE epoch = ?";
executeInternal(cql,
Collections.singleton(node.id), epoch);
- flush(Topologies);
return diskState;
}
@@ -1406,7 +1405,6 @@ public class AccordKeyspace
"SET closed = closed + ? WHERE epoch = ?";
executeInternal(cql,
KeySerializers.rangesToBlobMap(ranges), epoch);
- flush(Topologies);
return diskState;
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
index ff86a41ed0..d9351f8dae 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
@@ -292,7 +292,7 @@ public class AccordObjectSizes
private static final Result EMPTY_RESULT = new Result() {};
final static long NOT_DEFINED = measure(Command.SerializerSupport.notDefined(attrs(false, false), Ballot.ZERO));
- final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true), EMPTY_TXNID, null));;
+ final static long PREACCEPTED = measure(Command.SerializerSupport.preaccepted(attrs(false, true), EMPTY_TXNID, Ballot.ZERO));;
final static long ACCEPTED = measure(Command.SerializerSupport.accepted(attrs(true, false), SaveStatus.Accepted, EMPTY_TXNID, Ballot.ZERO, Ballot.ZERO));
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(Domain.Key), EMPTY_WRITES, EMPTY_RESULT));
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index 4095c31c67..bce8a198da 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -534,13 +534,12 @@ public class AccordService implements IAccordService, Shutdownable
{
List<ClusterMetadata> afterLoad = ClusterMetadataService.instance().processor().reconstructFull(Epoch.create(min), Epoch.create(max));
if (Invariants.isParanoid())
- assert afterLoad.get(0).epoch.getEpoch() == min : String.format("Unexpected epoch: expected %d but given %d", min, afterLoad.get(0).epoch.getEpoch());
+ Invariants.checkState(afterLoad.get(0).epoch.getEpoch() == min, "Unexpected epoch: expected %d but given %d", min, afterLoad.get(0).epoch.getEpoch());
while (!afterLoad.isEmpty() && afterLoad.get(0).epoch.getEpoch() < min)
afterLoad.remove(0);
- assert !afterLoad.isEmpty() : String.format("TCM was unable to return the needed epochs: %d -> %d", min, max);
- assert afterLoad.get(0).epoch.getEpoch() == min : String.format("Unexpected epoch: expected %d but given %d", min, afterLoad.get(0).epoch.getEpoch());
- if (max != Long.MAX_VALUE)
- assert afterLoad.get(afterLoad.size() - 1).epoch.getEpoch() == max : String.format("Unexpected epoch: expected %d but given %d", max, afterLoad.get(afterLoad.size() - 1).epoch.getEpoch());
+ Invariants.checkState(!afterLoad.isEmpty(), "TCM was unable to return the needed epochs: %d -> %d", min, max);
+ Invariants.checkState(afterLoad.get(0).epoch.getEpoch() == min, "Unexpected epoch: expected %d but given %d", min, afterLoad.get(0).epoch.getEpoch());
+ Invariants.checkState(max == Long.MAX_VALUE || afterLoad.get(afterLoad.size() - 1).epoch.getEpoch() == max, "Unexpected epoch: expected %d but given %d", max, afterLoad.get(afterLoad.size() - 1).epoch.getEpoch());
return afterLoad;
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
index 04f618e338..8439a05f2e 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java
@@ -29,13 +29,13 @@ import java.util.function.ToLongFunction;
import java.util.stream.Stream;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.IntrusiveLinkedList;
import accord.utils.Invariants;
import accord.utils.async.AsyncChains;
+import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.metrics.AccordStateCacheMetrics;
@@ -90,7 +90,9 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
}
}
- private ImmutableList<Instance<?, ?, ?>> instances = ImmutableList.of();
+ // TODO (required): cleanup on drop table, or else share between command stores
+ private Int2ObjectHashMap<Instance<?, ?, ?>> instances = new Int2ObjectHashMap<>();
+ private int nextIndex;
private final ExecutorPlus loadExecutor, saveExecutor;
@@ -243,13 +245,14 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
ToLongFunction<V> heapEstimator,
AccordCachingState.Factory<K, V> nodeFactory)
{
- int index = instances.size();
-
+ int index = ++nextIndex;
Instance<K, V, S> instance =
new Instance<>(index, keyClass, safeRefFactory, loadFunction, saveFunction, validateFunction, heapEstimator, nodeFactory);
- instances = ImmutableList.<Instance<?, ?, ?>>builder().addAll(instances).add(instance).build();
+ Int2ObjectHashMap<Instance<?, ?, ?>> newInstances = new Int2ObjectHashMap<>(instances);
+ newInstances.put(index, instance);
+ instances = newInstances;
return instance;
}
@@ -268,7 +271,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
public Collection<Instance<?, ? ,? >> instances()
{
- return instances;
+ return instances.values();
}
public interface Listener<K, V>
@@ -701,7 +704,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
{
bytesCached = 0;
metrics.reset();;
- instances.forEach(instance -> {
+ instances.values().forEach(instance -> {
instance.cache.forEach((k, v) -> Invariants.checkState(v.references == 0));
instance.cache.clear();
instance.bytesCached = 0;
@@ -739,7 +742,7 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
private int cacheSize()
{
int size = 0;
- for (Instance<?, ?, ?> instance : instances)
+ for (Instance<?, ?, ?> instance : instances.values())
size += instance.cache.size();
return size;
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopology.java b/src/java/org/apache/cassandra/service/accord/AccordTopology.java
index 2489929c16..bced75a401 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordTopology.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordTopology.java
@@ -43,6 +43,7 @@ import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.fastpath.FastPathStrategy;
import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeId;
@@ -333,6 +334,7 @@ public class AccordTopology
try
{
+ ClusterMetadataService.instance().fetchLogFromCMS(epoch);
AccordService.instance().epochReady(epoch).get(DatabaseDescriptor.getTransactionTimeout(MILLISECONDS), MILLISECONDS);
}
catch (InterruptedException e)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
index 9bda1950e9..5d8747d4a5 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
@@ -43,8 +43,6 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
@Override
public void doVerb(Message<T> message) throws IOException
{
- // TODO (desired): need a non-blocking way to inform CMS of an unknown epoch and add callback to it's receipt
-// ClusterMetadataService.instance().maybeCatchup(message.epoch());
logger.trace("Receiving {} from {}", message.payload, message.from());
T request = message.payload;
diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java
index 721d69c5a6..e327c1a5ce 100644
--- a/src/java/org/apache/cassandra/service/accord/IJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/IJournal.java
@@ -32,7 +32,7 @@ import accord.primitives.TxnId;
public interface IJournal
{
- Command loadCommand(int commandStoreId, TxnId txnId);
+ Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore);
RedundantBefore loadRedundantBefore(int commandStoreId);
DurableBefore loadDurableBefore(int commandStoreId);
diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java
index d3775869da..b8b09ef236 100644
--- a/src/java/org/apache/cassandra/service/accord/JournalKey.java
+++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java
@@ -25,6 +25,7 @@ import java.util.zip.Checksum;
import accord.local.Node.Id;
import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@@ -40,22 +41,21 @@ import org.apache.cassandra.utils.ByteArrayUtil;
import static org.apache.cassandra.db.TypeSizes.BYTE_SIZE;
import static org.apache.cassandra.db.TypeSizes.INT_SIZE;
import static org.apache.cassandra.db.TypeSizes.LONG_SIZE;
-import static org.apache.cassandra.db.TypeSizes.SHORT_SIZE;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.RangesForEpochSerializer;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.SafeToReadSerializer;
public final class JournalKey
{
public final Type type;
- public final Timestamp timestamp;
+ public final TxnId id;
public final int commandStoreId;
- public JournalKey(Timestamp timestamp, Type type, int commandStoreId)
+ public JournalKey(TxnId id, Type type, int commandStoreId)
{
Invariants.nonNull(type);
- Invariants.nonNull(timestamp);
+ Invariants.nonNull(id);
this.type = type;
- this.timestamp = timestamp;
+ this.id = id;
this.commandStoreId = commandStoreId;
}
@@ -67,86 +67,85 @@ public final class JournalKey
* when ordering timestamps. This is done for more precise elimination of candidate
* segments by min/max record key in segment.
*/
- public static final KeySupport<JournalKey> SUPPORT = new KeySupport<>()
+ public static final JournalKeySupport SUPPORT = new JournalKeySupport();
+
+ public static final class JournalKeySupport implements KeySupport<JournalKey>
{
- private static final int HLC_OFFSET = 0;
- private static final int EPOCH_AND_FLAGS_OFFSET = HLC_OFFSET + LONG_SIZE;
- private static final int NODE_OFFSET = EPOCH_AND_FLAGS_OFFSET + LONG_SIZE;
+ private static final int MSB_OFFSET = 0;
+ private static final int LSB_OFFSET = MSB_OFFSET + LONG_SIZE;
+ private static final int NODE_OFFSET = LSB_OFFSET + LONG_SIZE;
private static final int TYPE_OFFSET = NODE_OFFSET + INT_SIZE;
private static final int CS_ID_OFFSET = TYPE_OFFSET + BYTE_SIZE;
+ // TODO (required): revisit commandStoreId - this can go arbitrarily high so may want to use vint
+ public static final int TOTAL_SIZE = CS_ID_OFFSET + INT_SIZE;
@Override
public int serializedSize(int userVersion)
{
- return LONG_SIZE // timestamp.hlc()
- + 6 // timestamp.epoch()
- + 2 // timestamp.flags()
- + INT_SIZE // timestamp.node
- + BYTE_SIZE // type
- + SHORT_SIZE; // commandStoreId
+ return TOTAL_SIZE;
}
@Override
public void serialize(JournalKey key, DataOutputPlus out, int userVersion) throws IOException
{
- serializeTimestamp(key.timestamp, out);
+ serializeTxnId(key.id, out);
out.writeByte(key.type.id);
- out.writeShort(key.commandStoreId);
+ out.writeInt(key.commandStoreId);
}
private void serialize(JournalKey key, byte[] out)
{
- serializeTimestamp(key.timestamp, out);
- out[20] = (byte) (key.type.id & 0xFF);
- ByteArrayUtil.putShort(out, 21, (short) key.commandStoreId);
+ serializeTxnId(key.id, out);
+ out[TYPE_OFFSET] = (byte) (key.type.id & 0xFF);
+ ByteArrayUtil.putInt(out, CS_ID_OFFSET, key.commandStoreId);
}
@Override
public JournalKey deserialize(DataInputPlus in, int userVersion) throws IOException
{
- Timestamp timestamp = deserializeTimestamp(in);
+ TxnId txnId = deserializeTxnId(in);
int type = in.readByte();
- int commandStoreId = in.readShort();
- return new JournalKey(timestamp, Type.fromId(type), commandStoreId);
+ int commandStoreId = in.readInt();
+ return new JournalKey(txnId, Type.fromId(type), commandStoreId);
}
@Override
public JournalKey deserialize(ByteBuffer buffer, int position, int userVersion)
{
- Timestamp timestamp = deserializeTimestamp(buffer, position);
+ TxnId txnId = deserializeTxnId(buffer, position);
int type = buffer.get(position + TYPE_OFFSET);
- int commandStoreId = buffer.getShort(position + CS_ID_OFFSET);
- return new JournalKey(timestamp, Type.fromId(type), commandStoreId);
+ int commandStoreId = buffer.getInt(position + CS_ID_OFFSET);
+ return new JournalKey(txnId, Type.fromId(type), commandStoreId);
}
- private void serializeTimestamp(Timestamp timestamp, DataOutputPlus out) throws IOException
+ private void serializeTxnId(TxnId txnId, DataOutputPlus out) throws IOException
{
- out.writeLong(timestamp.hlc());
- out.writeLong(epochAndFlags(timestamp));
- out.writeInt(timestamp.node.id);
+ out.writeLong(txnId.msb);
+ out.writeLong(txnId.lsb);
+ out.writeInt(txnId.node.id);
}
- private Timestamp deserializeTimestamp(DataInputPlus in) throws IOException
+ private TxnId deserializeTxnId(DataInputPlus in) throws IOException
{
- long hlc = in.readLong();
- long epochAndFlags = in.readLong();
+ long msb = in.readLong();
+ long lsb = in.readLong();
int nodeId = in.readInt();
- return Timestamp.fromValues(epoch(epochAndFlags), hlc, flags(epochAndFlags), new Id(nodeId));
+ return TxnId.fromBits(msb, lsb, new Id(nodeId));
}
- private void serializeTimestamp(Timestamp timestamp, byte[] out)
+ private void serializeTxnId(TxnId txnId, byte[] out)
{
- ByteArrayUtil.putLong(out, 0, timestamp.hlc());
- ByteArrayUtil.putLong(out, 8, epochAndFlags(timestamp));
- ByteArrayUtil.putInt(out, 16, timestamp.node.id);
+ ByteArrayUtil.putLong(out, MSB_OFFSET, txnId.msb);
+ ByteArrayUtil.putLong(out, LSB_OFFSET, txnId.lsb);
+ ByteArrayUtil.putInt(out, NODE_OFFSET, txnId.node.id);
}
- private Timestamp deserializeTimestamp(ByteBuffer buffer, int position)
+ private TxnId deserializeTxnId(ByteBuffer buffer, int position)
{
- long hlc = buffer.getLong(position + HLC_OFFSET);
- long epochAndFlags = buffer.getLong(position + EPOCH_AND_FLAGS_OFFSET);
+ long msb = buffer.getLong(position + MSB_OFFSET);
+ long lsb = buffer.getLong(position + LSB_OFFSET);
int nodeId = buffer.getInt(position + NODE_OFFSET);
- return Timestamp.fromValues(epoch(epochAndFlags), hlc, flags(epochAndFlags), new Id(nodeId));
+ return TxnId.fromBits(msb, lsb, new Id(nodeId));
}
@Override
@@ -160,64 +159,41 @@ public final class JournalKey
@Override
public int compareWithKeyAt(JournalKey k, ByteBuffer buffer, int position, int userVersion)
{
- int cmp = compareWithTimestampAt(k.timestamp, buffer, position);
+ int cmp = compareWithTxnIdAt(k.id, buffer, position);
if (cmp != 0) return cmp;
byte type = buffer.get(position + TYPE_OFFSET);
cmp = Byte.compare((byte) k.type.id, type);
if (cmp != 0) return cmp;
- short commandStoreId = buffer.getShort(position + CS_ID_OFFSET);
- cmp = Short.compare((byte) k.commandStoreId, commandStoreId);
+ int commandStoreId = buffer.getInt(position + CS_ID_OFFSET);
+ cmp = Integer.compare(k.commandStoreId, commandStoreId);
return cmp;
}
- private int compareWithTimestampAt(Timestamp timestamp, ByteBuffer buffer, int position)
+ private int compareWithTxnIdAt(TxnId txnId, ByteBuffer buffer, int position)
{
- long hlc = buffer.getLong(position + HLC_OFFSET);
- int cmp = Long.compareUnsigned(timestamp.hlc(), hlc);
+ long msb = buffer.getLong(position + MSB_OFFSET);
+ int cmp = Timestamp.compareMsb(txnId.msb, msb);
if (cmp != 0) return cmp;
- long epochAndFlags = buffer.getLong(position + EPOCH_AND_FLAGS_OFFSET);
- cmp = Long.compareUnsigned(epochAndFlags(timestamp), epochAndFlags);
+ long lsb = buffer.getLong(position + LSB_OFFSET);
+ cmp = Timestamp.compareLsb(txnId.lsb, lsb);
if (cmp != 0) return cmp;
int nodeId = buffer.getInt(position + NODE_OFFSET);
- cmp = Integer.compareUnsigned(timestamp.node.id, nodeId);
+ cmp = Integer.compare(txnId.node.id, nodeId);
return cmp;
}
@Override
public int compare(JournalKey k1, JournalKey k2)
{
- int cmp = compare(k1.timestamp, k2.timestamp);
+ int cmp = k1.id.compareTo(k2.id);
if (cmp == 0) cmp = Byte.compare((byte) k1.type.id, (byte) k2.type.id);
- if (cmp == 0) cmp = Short.compare((short) k1.commandStoreId, (short) k2.commandStoreId);
+ if (cmp == 0) cmp = Integer.compare(k1.commandStoreId, k2.commandStoreId);
return cmp;
}
-
- private int compare(Timestamp timestamp1, Timestamp timestamp2)
- {
- int cmp = Long.compareUnsigned(timestamp1.hlc(), timestamp2.hlc());
- if (cmp == 0) cmp = Long.compareUnsigned(epochAndFlags(timestamp1), epochAndFlags(timestamp2));
- if (cmp == 0) cmp = Integer.compareUnsigned(timestamp1.node.id, timestamp2.node.id);
- return cmp;
- }
-
- private long epochAndFlags(Timestamp timestamp)
- {
- return (timestamp.epoch() << 16) | (long) timestamp.flags();
- }
-
- private long epoch(long epochAndFlags)
- {
- return epochAndFlags >>> 16;
- }
-
- private int flags(long epochAndFlags)
- {
- return (int) (epochAndFlags & ((1 << 16) - 1));
- }
};
@Override
@@ -230,7 +206,7 @@ public final class JournalKey
boolean equals(JournalKey other)
{
- return this.timestamp.equals(other.timestamp) &&
+ return this.id.equals(other.id) &&
this.type == other.type &&
this.commandStoreId == other.commandStoreId;
}
@@ -238,13 +214,13 @@ public final class JournalKey
@Override
public int hashCode()
{
- return Objects.hash(timestamp, type, commandStoreId);
+ return Objects.hash(id, type, commandStoreId);
}
public String toString()
{
return "Key{" +
- "timestamp=" + timestamp +
+ "id=" + id +
"type=" + type +
", commandStoreId=" + commandStoreId +
'}';
diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
index 75348c0e15..e81edde336 100644
--- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
@@ -26,8 +26,11 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Result;
+import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommonAttributes;
+import accord.local.DurableBefore;
+import accord.local.RedundantBefore;
import accord.local.StoreParticipants;
import accord.primitives.Ballot;
import accord.primitives.PartialDeps;
@@ -37,6 +40,7 @@ import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
+import accord.utils.Invariants;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
@@ -46,9 +50,11 @@ import org.apache.cassandra.service.accord.serializers.DepsSerializer;
import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer;
import org.apache.cassandra.utils.Throwables;
+import static accord.local.Cleanup.NO;
import static accord.primitives.Known.KnownDeps.DepsErased;
import static accord.primitives.Known.KnownDeps.DepsUnknown;
import static accord.primitives.Known.KnownDeps.NoDeps;
+import static accord.primitives.Status.Durability.NotDurable;
import static accord.utils.Invariants.illegalState;
public class SavedCommand
@@ -56,7 +62,6 @@ public class SavedCommand
// This enum is order-dependent
public enum Fields
{
- TXN_ID,
EXECUTE_AT,
EXECUTES_AT_LEAST,
SAVE_STATUS,
@@ -68,6 +73,9 @@ public class SavedCommand
PARTIAL_DEPS,
WAITING_ON,
WRITES,
+ ;
+
+ static final Fields[] FIELDS = values();
}
// TODO: maybe rename this and enclosing classes?
@@ -156,8 +164,6 @@ public class SavedCommand
out.writeInt(flags);
// We encode all changed fields unless their value is null
- if (getFieldChanged(Fields.TXN_ID, flags) && after.txnId() != null)
- CommandSerializers.txnId.serialize(after.txnId(), out, userVersion);
if (getFieldChanged(Fields.EXECUTE_AT, flags) && after.executeAt() != null)
CommandSerializers.timestamp.serialize(after.executeAt(), out, userVersion);
// TODO (desired): check if this can fold into executeAt
@@ -198,7 +204,6 @@ public class SavedCommand
{
int flags = 0;
- flags = collectFlags(before, after, Command::txnId, true, Fields.TXN_ID, flags);
flags = collectFlags(before, after, Command::executeAt, true, Fields.EXECUTE_AT, flags);
flags = collectFlags(before, after, Command::executesAtLeast, true, Fields.EXECUTES_AT_LEAST, flags);
flags = collectFlags(before, after, Command::saveStatus, false, Fields.SAVE_STATUS, flags);
@@ -303,9 +308,13 @@ public class SavedCommand
boolean nextCalled;
int count;
+ public Builder(TxnId txnId)
+ {
+ init(txnId);
+ }
+
public Builder()
{
- clear();
}
public TxnId txnId()
@@ -376,28 +385,33 @@ public class SavedCommand
public void clear()
{
flags = 0;
-
- txnId = null;
-
executeAt = null;
saveStatus = null;
durability = null;
-
- acceptedOrCommitted = Ballot.ZERO;
promised = null;
-
participants = null;
partialTxn = null;
partialDeps = null;
-
- waitingOn = (txn, deps) -> null;
writes = null;
- result = CommandSerializers.APPLIED;
-
nextCalled = false;
count = 0;
}
+ public void reset(TxnId txnId)
+ {
+ clear();
+ init(txnId);
+ }
+
+ public void init(TxnId txnId)
+ {
+ this.txnId = txnId;
+ durability = NotDurable;
+ acceptedOrCommitted = promised = Ballot.ZERO;
+ waitingOn = (txn, deps) -> null;
+ result = CommandSerializers.APPLIED;
+ }
+
public boolean isEmpty()
{
return !nextCalled;
@@ -408,19 +422,52 @@ public class SavedCommand
return count;
}
+ public Cleanup shouldCleanup(RedundantBefore redundantBefore, DurableBefore durableBefore)
+ {
+ if (!nextCalled)
+ return NO;
+
+ if (saveStatus == null || participants == null)
+ return Cleanup.EXPUNGE_PARTIAL;
+
+ return Cleanup.shouldCleanup(txnId, saveStatus, durability, participants, redundantBefore, durableBefore);
+ }
+
+ // TODO (expected): avoid allocating new builder
+ public Builder maybeCleanup(Cleanup cleanup)
+ {
+ // Do not have txnId in selected SSTables; remove
+ if (saveStatus() == null)
+ return null;
+
+ switch (cleanup)
+ {
+ case EXPUNGE:
+ case ERASE:
+ return null;
+
+ case EXPUNGE_PARTIAL:
+ return expungePartial();
+ case VESTIGIAL:
+ case INVALIDATE:
+ case TRUNCATE_WITH_OUTCOME:
+ case TRUNCATE:
+ return saveStatusOnly();
+ case NO:
+ return this;
+ default:
+ throw new IllegalStateException("Unknown cleanup: " + cleanup);}
+ }
+
public Builder expungePartial()
{
- Builder builder = new Builder();
+ Invariants.checkState(txnId != null);
+ Builder builder = new Builder(txnId);
builder.count++;
builder.nextCalled = true;
// TODO: these accesses can be abstracted away
- if (txnId != null)
- {
- builder.flags = setFieldChanged(Fields.TXN_ID, builder.flags);
- builder.txnId = txnId;
- }
if (executeAt != null)
{
builder.flags = setFieldChanged(Fields.EXECUTE_AT, builder.flags);
@@ -442,17 +489,13 @@ public class SavedCommand
public Builder saveStatusOnly()
{
- Builder builder = new Builder();
+ Invariants.checkState(txnId != null);
+ Builder builder = new Builder(txnId);
builder.count++;
builder.nextCalled = true;
// TODO: these accesses can be abstracted away
- if (txnId != null)
- {
- builder.flags = setFieldChanged(Fields.TXN_ID, builder.flags);
- builder.txnId = txnId;
- }
if (saveStatus != null)
{
builder.flags = setFieldChanged(Fields.SAVE_STATUS, builder.flags);
@@ -476,8 +519,6 @@ public class SavedCommand
out.writeInt(flags);
// We encode all changed fields unless their value is null
- if (getFieldChanged(Fields.TXN_ID, flags) && !getFieldIsNull(Fields.TXN_ID, flags))
- CommandSerializers.txnId.serialize(txnId(), out, userVersion);
if (getFieldChanged(Fields.EXECUTE_AT, flags) && !getFieldIsNull(Fields.EXECUTE_AT, flags))
CommandSerializers.timestamp.serialize(executeAt(), out, userVersion);
// TODO (desired): check if this can fold into executeAt
@@ -515,11 +556,12 @@ public class SavedCommand
@SuppressWarnings({ "rawtypes", "unchecked" })
public void deserializeNext(DataInputPlus in, int userVersion) throws IOException
{
+ Invariants.checkState(txnId != null);
final int flags = in.readInt();
nextCalled = true;
count++;
- for (Fields field : Fields.values())
+ for (Fields field : Fields.FIELDS)
{
if (getFieldChanged(field, flags))
{
@@ -531,14 +573,6 @@ public class SavedCommand
}
}
- if (getFieldChanged(Fields.TXN_ID, flags))
- {
- if (getFieldIsNull(Fields.TXN_ID, flags))
- txnId = null;
- else
- txnId = CommandSerializers.txnId.deserialize(in, userVersion);
- }
-
if (getFieldChanged(Fields.EXECUTE_AT, flags))
{
if (getFieldIsNull(Fields.EXECUTE_AT, flags))
@@ -625,6 +659,7 @@ public class SavedCommand
waitingOn = (localTxnId, deps) -> {
try
{
+ Invariants.nonNull(deps);
return WaitingOnSerializer.deserialize(localTxnId, deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, buffer);
}
catch (IOException e)
@@ -642,7 +677,6 @@ public class SavedCommand
else
writes = CommandSerializers.writes.deserialize(in, userVersion);
}
-
}
public void forceResult(Result newValue)
@@ -655,6 +689,7 @@ public class SavedCommand
if (!nextCalled)
return null;
+ Invariants.checkState(txnId != null);
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
if (partialTxn != null)
attrs.partialTxn(partialTxn);
diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
index cbcfe22f18..62cda848a6 100644
--- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
+++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
@@ -270,7 +270,6 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
}
commandStore.completeOperation(safeStore);
-
context.releaseResources(commandStore);
state(COMPLETING);
if (diffs != null || safeStore.fieldUpdates() != null)
diff --git a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java
index 41efdb245b..9a43c3eee9 100644
--- a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java
+++ b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java
@@ -18,8 +18,10 @@
package org.apache.cassandra.tcm;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@@ -30,8 +32,10 @@ import com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
+import org.apache.cassandra.tcm.log.LogReader;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.log.LogStorage;
@@ -76,9 +80,31 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
return log.waitForHighestConsecutive();
}
+ @Override
public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
{
- return log.getLocalEntries(lowEpoch);
+ try
+ {
+ LogReader.EntryHolder state = log.storage().getEntries(Epoch.EMPTY, highEpoch);
+ ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
+
+ Iterator<Entry> iter = state.iterator();
+ ImmutableList.Builder<Entry> rest = new ImmutableList.Builder<>();
+ while (iter.hasNext())
+ {
+ Entry current = iter.next();
+ if (current.epoch.isEqualOrBefore(lowEpoch))
+ metadata = current.transform.execute(metadata).success().metadata;
+ else
+ rest.add(current);
+ }
+
+ return new LogState(metadata, rest.build());
+ }
+ catch (IOException t)
+ {
+ throw new RuntimeException(t);
+ }
}
public static class InMemoryStorage implements LogStorage
@@ -133,12 +159,37 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor
@Override
public synchronized EntryHolder getEntries(Epoch since)
{
- throw new IllegalStateException("We have overridden all callers of this method, it should never be called");
+ EntryHolder entryHolder = new EntryHolder(since);
+ entries.stream().filter(e -> e.epoch.isAfter(since)).forEach(entryHolder::add);
+ return entryHolder;
+ }
+
+ @Override
+ public synchronized EntryHolder getEntries(Epoch since, Epoch until)
+ {
+ EntryHolder entryHolder = new EntryHolder(since);
+ entries.stream().filter(e -> e.epoch.isAfter(since) && e.epoch.isEqualOrBefore(until)).forEach(entryHolder::add);
+ return entryHolder;
}
- public EntryHolder getEntries(Epoch since, Epoch until)
+ public LogState getLogState(Epoch start, Epoch end)
{
- throw new IllegalStateException("We have overridden all callers of this method, it should never be called");
+ EntryHolder state = getEntries(Epoch.EMPTY);
+ ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());;
+ Iterator<Entry> iter = state.iterator();
+ ImmutableList.Builder<Entry> rest = new ImmutableList.Builder<>();
+ while (iter.hasNext())
+ {
+ Entry current = iter.next();
+ if (current.epoch.isAfter(end))
+ break;
+ if (current.epoch.isEqualOrBefore(start))
+ metadata = current.transform.execute(metadata).success().metadata;
+ else
+ rest.add(current);
+ }
+
+ return new LogState(metadata, rest.build());
}
}
diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
index c5e9b8bdf0..40012459e7 100644
--- a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
+++ b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
@@ -158,7 +158,7 @@ public class ClusterMetadataService
Processor localProcessor;
if (CassandraRelevantProperties.TCM_USE_ATOMIC_LONG_PROCESSOR.getBoolean())
{
- log = logSpec.sync().createLog();
+ log = logSpec.sync().withStorage(new AtomicLongBackedProcessor.InMemoryStorage()).createLog();
localProcessor = wrapProcessor.apply(new AtomicLongBackedProcessor(log, logSpec.isReset()));
fetchLogHandler = new FetchCMSLog.Handler((e, ignored) -> logSpec.storage().getLogState(e));
}
diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java
index 4af2194163..02c7e5eb89 100644
--- a/src/java/org/apache/cassandra/tcm/Startup.java
+++ b/src/java/org/apache/cassandra/tcm/Startup.java
@@ -509,15 +509,17 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
}
if (seeds.isEmpty())
throw new IllegalArgumentException("Can not initialize CMS without any seeds");
-
boolean hasAnyEpoch = SystemKeyspaceStorage.hasAnyEpoch();
+
// For CCM and local dev clusters
boolean isOnlySeed = DatabaseDescriptor.getSeeds().size() == 1
&& DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddressAndPort())
&& DatabaseDescriptor.getSeeds().iterator().next().getAddress().isLoopbackAddress();
boolean hasBootedBefore = SystemKeyspace.getLocalHostId() != null;
logger.info("hasAnyEpoch = {}, hasBootedBefore = {}", hasAnyEpoch, hasBootedBefore);
- if (!hasAnyEpoch && hasBootedBefore)
+ if (!hasAnyEpoch && hasBootedBefore &&
+ // Atomic long processor currently does not support upgrades
+ !CassandraRelevantProperties.TCM_USE_ATOMIC_LONG_PROCESSOR.getBoolean())
return UPGRADE;
else if (hasAnyEpoch)
return NORMAL;
diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
index 6bbd31acdd..0bcb982f1a 100644
--- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java
+++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
@@ -290,6 +290,11 @@ public abstract class LocalLog implements Closeable
assert metadata.epoch.is(Epoch.FIRST) : String.format("Epoch: %s. CMS: %s", metadata.epoch, metadata.fullCMSMembers());
}
+ public LogStorage storage()
+ {
+ return storage;
+ }
+
public ClusterMetadata metadata()
{
return committed.get();
diff --git a/src/java/org/apache/cassandra/tcm/log/LogReader.java b/src/java/org/apache/cassandra/tcm/log/LogReader.java
index b8d62e7587..b1e7ab3264 100644
--- a/src/java/org/apache/cassandra/tcm/log/LogReader.java
+++ b/src/java/org/apache/cassandra/tcm/log/LogReader.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.tcm.log;
import java.io.IOException;
import java.util.Comparator;
+import java.util.Iterator;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
@@ -195,6 +196,11 @@ public interface LogReader
return true;
}
+ public Iterator<Entry> iterator()
+ {
+ return entries.iterator();
+ }
+
private ImmutableList<Entry> immutable()
{
return ImmutableList.copyOf(entries);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
index 8478a060d7..ae2f1de4ee 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.util.concurrent.RateLimiter;
@@ -44,6 +45,7 @@ import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.distributed.shared.DistributedTestBase;
import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.utils.EstimatedHistogram;
import static java.lang.System.currentTimeMillis;
@@ -80,25 +82,32 @@ public class AccordLoadTest extends AccordTestBase
}).drop();
ICoordinator coordinator = cluster.coordinator(1);
final int repairInterval = 3000;
- final int batchSize = 1000;
+ final int compactionInterval = 3000;
+ final int flushInterval = 1000;
+ final int batchSizeLimit = 1000;
+ final long batchTime = TimeUnit.SECONDS.toNanos(10);
final int concurrency = 100;
final int ratePerSecond = 1000;
- final int keyCount = 100000;
+ final int keyCount = 1000000;
final float readChance = 0.33f;
long nextRepairAt = repairInterval;
+ long nextCompactionAt = compactionInterval;
+ long nextFlushAt = flushInterval;
final BitSet initialised = new BitSet();
Random random = new Random();
// CopyOnWriteArrayList<Throwable> exceptions = new CopyOnWriteArrayList<>();
final Semaphore inFlight = new Semaphore(concurrency);
final RateLimiter rateLimiter = RateLimiter.create(ratePerSecond);
- long testStart = System.nanoTime();
+// long testStart = System.nanoTime();
// while (NANOSECONDS.toMinutes(System.nanoTime() - testStart) < 10 && exceptions.size() < 10000)
while (true)
{
final EstimatedHistogram histogram = new EstimatedHistogram(200);
long batchStart = System.nanoTime();
- for (int i = 0 ; i < batchSize ; ++i)
+ long batchEnd = batchStart + batchTime;
+ int batchSize = 0;
+ while (batchSize < batchSizeLimit)
{
inFlight.acquire();
rateLimiter.acquire();
@@ -112,7 +121,7 @@ public class AccordLoadTest extends AccordTestBase
// else exceptions.add(fail);
}, "SELECT * FROM " + qualifiedAccordTableName + " WHERE k = ?;", ConsistencyLevel.SERIAL, k);
}
- else if (initialised.get(i))
+ else if (initialised.get(k))
{
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
@@ -122,13 +131,16 @@ public class AccordLoadTest extends AccordTestBase
}
else
{
- initialised.set(i);
+ initialised.set(k);
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
// else exceptions.add(fail);
}, "UPDATE " + qualifiedAccordTableName + " SET v = 0 WHERE k = ? IF NOT EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, k);
}
+ batchSize++;
+ if (System.nanoTime() >= batchEnd)
+ break;
}
if ((nextRepairAt -= batchSize) <= 0)
@@ -138,8 +150,22 @@ public class AccordLoadTest extends AccordTestBase
cluster.coordinator(1).instance().nodetool("repair", qualifiedAccordTableName);
}
+ if ((nextCompactionAt -= batchSize) <= 0)
+ {
+ nextCompactionAt += compactionInterval;
+ System.out.println("compacting accord...");
+ cluster.forEach(i -> i.nodetool("compact", "system_accord.journal"));
+ }
+
+ if ((nextFlushAt -= batchSize) <= 0)
+ {
+ nextFlushAt += flushInterval;
+ System.out.println("flushing journal...");
+ cluster.forEach(i -> i.runOnInstance(() -> ((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTesting()));
+ }
+
final Date date = new Date();
- System.out.printf("%tT rate: %.2f/s\n", date, (((float)batchSize * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStart)));
+ System.out.printf("%tT rate: %.2f/s\n", date, (((float)batchSizeLimit * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStart)));
System.out.printf("%tT percentiles: %d %d %d %d\n", date, histogram.percentile(.25)/1000, histogram.percentile(.5)/1000, histogram.percentile(.75)/1000, histogram.percentile(1)/1000);
class VerbCount
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 1be3fd686d..0503684e11 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationRaceTestBase.java
@@ -542,15 +542,14 @@ public abstract class AccordMigrationRaceTestBase extends AccordTestBase
if (unpauseAfterBatchLogCreatesTransaction)
{
logger.info("Creating thread to unpause after batchlog creates Accord transaction");
- new Thread(() ->
- {
+ new Thread(() -> {
try
{
// Unpause so it can route incorrectly instead of timing out waiting to fetch the epoch, need the transaction to be created first
// otherwise it will just be routed straight to non-Accord.
logger.info("Spinning waiting on a transaction");
Util.spinUntilTrue(() -> !((AccordService)AccordService.instance()).node().coordinating().isEmpty(), 20);
- logger.info("Foudn transaction, unpausing");
+ logger.info("Found transaction, unpausing");
TestChangeListener.instance.unpause();
unpaused.trySuccess(null);
}
diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java
index e64b5aba84..6aacbbc41c 100644
--- a/test/unit/org/apache/cassandra/ServerTestUtils.java
+++ b/test/unit/org/apache/cassandra/ServerTestUtils.java
@@ -28,7 +28,6 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
-import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -293,22 +292,23 @@ public final class ServerTestUtils
// log entries is always done by the dedicated log follower thread.
DatabaseDescriptor.setMetadataSnapshotFrequency(Integer.MAX_VALUE);
- Function<LocalLog, Processor> processorFactory = AtomicLongBackedProcessor::new;
IPartitioner partitioner = DatabaseDescriptor.getPartitioner();
boolean addListeners = true;
ClusterMetadata initial = new ClusterMetadata(partitioner);
if (!Keyspace.isInitialized())
Keyspace.setInitialized();
+ AtomicLongBackedProcessor.InMemoryStorage storage = new AtomicLongBackedProcessor.InMemoryStorage();
LocalLog log = LocalLog.logSpec()
.withInitialState(initial)
.withDefaultListeners(addListeners)
+ .withStorage(storage)
.createLog();
ResettableClusterMetadataService service = new ResettableClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
- processorFactory.apply(log),
+ new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
true);
diff --git a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
index 372619d145..0bdc561818 100644
--- a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
+++ b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
@@ -27,6 +27,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
@@ -502,6 +503,22 @@ public class YamlConfigurationLoaderTest
assertTrue(config.authenticator.parameters.isEmpty());
}
+ @Test
+ public void testAccordConfig()
+ {
+ Map<String, String> accordSpec = ImmutableMap.of("fast_path_update_delay", "60s",
+ "schedule_durability_frequency", "60s",
+ "durability_txnid_lag", "60s",
+ "shard_durability_cycle", "60s",
+ "global_durability_cycle", "60s");
+ AccordSpec spec = from("accord", accordSpec).accord;
+ assertThat(spec.fast_path_update_delay.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L);
+ assertThat(spec.schedule_durability_frequency.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L);
+ assertThat(spec.durability_txnid_lag.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L);
+ assertThat(spec.shard_durability_cycle.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L);
+ assertThat(spec.global_durability_cycle.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L);
+ }
+
private static Config fromYaml(Object... values)
{
assert values.length % 2 == 0 : "Map can only be created with an even number of inputs: given " + values.length;
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
index 181d2c7ca6..57ef025061 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
@@ -436,9 +436,9 @@ public class CompactionAccordIteratorsTest
{
commandStore.executeBlocking(() -> {
// clear cache and wait for post-eviction writes to complete
- long cacheSize = commandStore.capacity();
- commandStore.setCapacity(0);
- commandStore.setCapacity(cacheSize);
+ long cacheSize = commandStore.cache().capacity();
+ commandStore.cache().setCapacity(0);
+ commandStore.cache().setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
commands.forceBlockingFlush(FlushReason.UNIT_TESTS);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
index 28f54ce688..4a53512862 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
@@ -93,7 +93,7 @@ public class AccordCommandTest
public void basicCycleTest() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
- getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0)));
+ getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.cache().setCapacity(0)));
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(1);
@@ -185,7 +185,7 @@ public class AccordCommandTest
public void computeDeps() throws Throwable
{
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
- getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0)));
+ getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.cache().setCapacity(0)));
TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1);
Txn txn = createWriteTxn(2);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
index 9daa1bb9fa..7210d4a097 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
@@ -20,12 +20,27 @@ package org.apache.cassandra.service.accord;
import java.net.UnknownHostException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
-import java.util.Set;
+import java.util.Optional;
import java.util.UUID;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.schema.DistributedSchema;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.Tables;
+import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
+import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.tcm.serialization.Version;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -33,9 +48,7 @@ import org.junit.Test;
import accord.api.ConfigurationService.EpochReady;
import accord.impl.AbstractConfigurationServiceTest;
-import accord.local.Node;
import accord.local.Node.Id;
-import accord.topology.Shard;
import accord.topology.Topology;
import accord.utils.SortedArrays.SortedArrayList;
import org.apache.cassandra.SchemaLoader;
@@ -57,7 +70,6 @@ import org.apache.cassandra.utils.MockFailureDetector;
import org.apache.cassandra.utils.concurrent.Future;
import static accord.impl.AbstractConfigurationServiceTest.TestListener;
-import static com.google.common.collect.ImmutableSet.of;
import static java.lang.String.format;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
@@ -72,9 +84,8 @@ public class AccordConfigurationServiceTest
private static final Id ID2 = new Id(2);
private static final Id ID3 = new Id(3);
private static final SortedArrayList<Id> ID_LIST = new SortedArrayList<>(new Id[] { ID1, ID2, ID3 });
- private static final Set<Id> ID_SET = ImmutableSet.copyOf(ID_LIST);
- private static final TableId TBL1 = TableId.fromUUID(new UUID(0, 1));
- private static final TableId TBL2 = TableId.fromUUID(new UUID(0, 2));
+ private static final String KEYSPACE_NAME = "test_ks";
+ private static final TableId TBL_ID = TableId.fromUUID(new UUID(0, 1));
private static EndpointMapping mappingForEpoch(long epoch)
{
@@ -92,21 +103,6 @@ public class AccordConfigurationServiceTest
}
}
- private static EndpointMapping mappingForTopology(Topology topology)
- {
- try
- {
- EndpointMapping.Builder builder = EndpointMapping.builder(topology.epoch());
- for (Node.Id id : topology.nodes())
- builder.add(InetAddressAndPort.getByName("127.0.0." + id.id), id);
- return builder.build();
- }
- catch (UnknownHostException e)
- {
- throw new RuntimeException(e);
- }
- }
-
private static class Messaging implements MessageDelivery
{
static class Request
@@ -177,15 +173,17 @@ public class AccordConfigurationServiceTest
@Test
public void initialEpochTest() throws Throwable
{
+ ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
+
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
Assert.assertEquals(null, AccordKeyspace.loadEpochDiskState());
service.start();
Assert.assertEquals(null, AccordKeyspace.loadEpochDiskState());
Assert.assertTrue(executeInternal(format("SELECT * FROM %s.%s WHERE epoch=1", ACCORD_KEYSPACE_NAME, TOPOLOGIES)).isEmpty());
- Topology topology1 = new Topology(1, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, ID_SET));
+ Topology topology1 = createTopology(cms);
service.reportTopology(topology1);
- loadEpoch(1, null, (epoch, cm, topology, syncStatus, pendingSync, remoteSync, closed, redundant) -> {
+ loadEpoch(1, cms.metadata(), (epoch, cm, topology, syncStatus, pendingSync, remoteSync, closed, redundant) -> {
Assert.assertEquals(topology1, topology);
Assert.assertTrue(remoteSync.isEmpty());
});
@@ -193,37 +191,39 @@ public class AccordConfigurationServiceTest
service.receiveRemoteSyncComplete(ID1, 1);
service.receiveRemoteSyncComplete(ID2, 1);
- loadEpoch(1, null, (epoch, cm, topology, syncStatus, pendingSync, remoteSync, closed, redundant) -> {
+ loadEpoch(1, cms.metadata(), (epoch, cm, topology, syncStatus, pendingSync, remoteSync, closed, redundant) -> {
Assert.assertEquals(topology1, topology);
Assert.assertEquals(Sets.newHashSet(ID1, ID2), remoteSync);
});
}
@Test
- public void loadTest() throws Throwable
+ public void loadTest()
{
+ ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
+
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
service.start();
- Topology topology1 = new Topology(1, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, ID_SET));
- service.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
+ Topology topology1 = createTopology(cms);
+ service.updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
service.reportTopology(topology1);
service.acknowledgeEpoch(EpochReady.done(1), true);
service.receiveRemoteSyncComplete(ID1, 1);
service.receiveRemoteSyncComplete(ID2, 1);
service.receiveRemoteSyncComplete(ID3, 1);
- Topology topology2 = new Topology(2, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, of(ID1, ID2)));
+ Topology topology2 = createTopology(cms);
service.reportTopology(topology2);
service.acknowledgeEpoch(EpochReady.done(2), true);
service.receiveRemoteSyncComplete(ID1, 2);
- Topology topology3 = new Topology(3, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, of(ID1, ID2)));
+ Topology topology3 = createTopology(cms);
service.reportTopology(topology3);
service.acknowledgeEpoch(EpochReady.done(3), true);
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
- loaded.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
+ loaded.updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
AbstractConfigurationServiceTest.TestListener listener = new AbstractConfigurationServiceTest.TestListener(loaded, true);
loaded.registerListener(listener);
loaded.start();
@@ -241,29 +241,91 @@ public class AccordConfigurationServiceTest
@Test
public void truncateTest()
{
+ ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
+
AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
TestListener serviceListener = new TestListener(service, true);
service.registerListener(serviceListener);
service.start();
- Topology topology1 = new Topology(1, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, ID_SET));
- service.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
+ Topology topology1 = createTopology(cms);
+ service.updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
service.reportTopology(topology1);
- Topology topology2 = new Topology(2, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, of(ID1, ID2)));
+ Topology topology2 = createTopology(cms);
service.reportTopology(topology2);
- Topology topology3 = new Topology(3, new Shard(AccordTopology.fullRange(TBL1), ID_LIST, of(ID1, ID2)));
+ Topology topology3 = createTopology(cms);
service.reportTopology(topology3);
service.truncateTopologiesUntil(3);
Assert.assertEquals(EpochDiskState.create(3), service.diskState());
serviceListener.assertTruncates(3L);
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), AccordConfigurationService.SystemTableDiskStateManager.instance, ScheduledExecutors.scheduledTasks);
- loaded.updateMapping(mappingForEpoch(ClusterMetadata.current().epoch.getEpoch() + 1));
+ loaded.updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
TestListener loadListener = new TestListener(loaded, true);
loaded.registerListener(loadListener);
loaded.start();
loadListener.assertTopologiesFor(3L);
}
+
+ private static Topology createTopology(ValidatingClusterMetadataService cms)
+ {
+ ClusterMetadata previous = cms.metadata();
+ ClusterMetadata.Transformer next = previous.transformer();
+ maybeCreateTable(previous, next);
+
+ ClusterMetadata metadata = next.build().metadata;
+ cms.setMetadata(metadata);
+ return AccordTopology.createAccordTopology(metadata);
+ }
+
+ private static void maybeCreateTable(ClusterMetadata previous, ClusterMetadata.Transformer next)
+ {
+ Optional<KeyspaceMetadata> ks = previous.schema.getKeyspaces().get(KEYSPACE_NAME);
+ if (ks.isPresent()) return;
+ // lets create it
+ TableMetadata table = TableMetadata.builder(KEYSPACE_NAME, "tbl")
+ .id(TBL_ID)
+ .kind(TableMetadata.Kind.REGULAR)
+ .partitioner(Murmur3Partitioner.instance)
+ .addPartitionKeyColumn("pk", Int32Type.instance)
+ .build();
+ KeyspaceMetadata keyspace = KeyspaceMetadata.create(KEYSPACE_NAME, KeyspaceParams.simple(ID_LIST.size()))
+ .withSwapped(Tables.builder().add(table).build());
+
+ next.with(new DistributedSchema(previous.schema.getKeyspaces().with(keyspace)));
+
+ for (Id node : ID_LIST)
+ {
+ // not forcing the cms node id to match as they do when this logic was first added...
+ next.register(new NodeAddresses(getAddress(node)),
+ new Location("dc1", "rack1"),
+ NodeVersion.CURRENT);
+
+ next.proposeToken(new NodeId(node.id), Collections.singleton(new Murmur3Partitioner.LongToken(node.id)));
+ }
+
+ DataPlacement.Builder replication = DataPlacement.builder();
+ Range<Token> fullRange = new Range<>(Murmur3Partitioner.MINIMUM, Murmur3Partitioner.MINIMUM);
+ for (int i = 0; i < ID_LIST.size(); i++)
+ {
+ InetAddressAndPort address = getAddress(ID_LIST.get(i));
+ Replica replica = new Replica(address, fullRange, true);
+ replication.withReadReplica(next.epoch(), replica).withWriteReplica(next.epoch(), replica);
+ }
+ next.with(previous.placements.unbuild().with(keyspace.params.replication, replication.build()).build());
+ }
+
+ private static InetAddressAndPort getAddress(Id node)
+ {
+ try
+ {
+ return InetAddressAndPort.getByAddress(new byte[]{127, 0, 0, (byte) node.id});
+ }
+ catch (UnknownHostException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java
index 34a270f544..b255c00cbf 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java
@@ -83,7 +83,7 @@ public class AccordJournalOrderTest
Runnable check = () -> {
for (JournalKey key : res.keySet())
{
- SavedCommand.Builder diffs = accordJournal.loadDiffs(key.commandStoreId, (TxnId) key.timestamp);
+ SavedCommand.Builder diffs = accordJournal.loadDiffs(key.commandStoreId, (TxnId) key.id);
Assert.assertEquals(String.format("%d != %d for key %s", diffs.count(), res.get(key).intValue(), key),
diffs.count(), res.get(key).intValue());
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
index 57006fdbf2..3e74728f70 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
@@ -53,7 +53,6 @@ import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
-import accord.primitives.SaveStatus;
import accord.primitives.Ballot;
import accord.primitives.FullKeyRoute;
import accord.primitives.FullRoute;
@@ -62,8 +61,10 @@ import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Routable;
+import accord.primitives.SaveStatus;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
+import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
@@ -105,6 +106,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.primitives.Routable.Domain.Key;
import static accord.utils.async.AsyncChains.getUninterruptibly;
import static java.lang.String.format;
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
public class AccordTestUtils
{
@@ -125,6 +127,7 @@ public class AccordTestUtils
CommonAttributes.Mutable attrs = new CommonAttributes.Mutable(txnId);
attrs.partialTxn(txn);
attrs.setParticipants(StoreParticipants.all(route(txn)));
+ attrs.durability(Status.Durability.NotDurable);
return Command.SerializerSupport.preaccepted(attrs, executeAt, Ballot.ZERO);
}
@@ -401,6 +404,7 @@ public class AccordTestUtils
AccordJournal journal = new AccordJournal(new AccordSpec.JournalSpec());
journal.start(null);
+ AccordStateCache stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20, new AccordStateCacheMetrics("test"));
SingleEpochRanges holder = new SingleEpochRanges(topology.rangesForNode(node));
AccordCommandStore result = new AccordCommandStore(0,
time,
@@ -414,9 +418,7 @@ public class AccordTestUtils
}),
holder,
journal,
- loadExecutor,
- saveExecutor,
- new AccordStateCacheMetrics(AccordCommandStores.ACCORD_STATE_CACHE + System.currentTimeMillis()));
+ new AccordCommandStore.CommandStoreExecutor(stateCache, executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + 0 + ']')));
holder.set(result);
// TODO: CompactionAccordIteratorsTest relies on this
@@ -439,7 +441,7 @@ public class AccordTestUtils
Node.Id node = new Id(1);
Topology topology = new Topology(1, new Shard(range, new SortedArrayList<>(new Id[] { node }), Sets.newHashSet(node), Collections.emptySet()));
AccordCommandStore store = createAccordCommandStore(node, now, topology, loadExecutor, saveExecutor);
- store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).setCapacity(1 << 20));
+ store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).cache().setCapacity(1 << 20));
return store;
}
diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
index 9f6827f1b9..24b911a3ff 100644
--- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java
+++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
@@ -69,7 +69,7 @@ public class MockJournal implements IJournal
private final Map<Integer, FieldUpdates> fieldUpdates = new HashMap<>();
@Override
- public Command loadCommand(int store, TxnId txnId)
+ public Command loadCommand(int store, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, store);
List<LoadedDiff> saved = commands.get(key);
diff --git a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
index 1d86856922..5627029644 100644
--- a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
@@ -74,7 +74,7 @@ public class SavedCommandTest
public void simpleNullChangeCheck()
{
int flags = getFlags(null, Command.NotDefined.uninitialised(TxnId.NONE));
- EnumSet<Fields> has = EnumSet.of(Fields.TXN_ID, Fields.SAVE_STATUS, Fields.PARTICIPANTS, Fields.DURABILITY, Fields.PROMISED,
+ EnumSet<Fields> has = EnumSet.of(Fields.SAVE_STATUS, Fields.PARTICIPANTS, Fields.DURABILITY, Fields.PROMISED,
Fields.ACCEPTED /* this is Zero... which kinda means null... */);
Set<Fields> missing = Sets.difference(ALL, has);
assertHas(flags, has);
@@ -96,7 +96,7 @@ public class SavedCommandTest
out.clear();
Command orig = cmdBuilder.build(saveStatus);
SavedCommand.serialize(null, orig, out, userVersion);
- SavedCommand.Builder builder = new SavedCommand.Builder();
+ SavedCommand.Builder builder = new SavedCommand.Builder(orig.txnId());
builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), userVersion);
// We are not persisting the result, so force it for strict equality
builder.forceResult(orig.result());
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
index 06469b0bb2..78708f5538 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
@@ -52,7 +52,6 @@ import accord.primitives.FullRoute;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.RoutableKey;
-import accord.primitives.Routables;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -81,6 +80,7 @@ import org.apache.cassandra.utils.Generators;
import org.apache.cassandra.utils.Pair;
import org.assertj.core.api.Assertions;
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
import static org.apache.cassandra.utils.AccordGenerators.fromQT;
@@ -109,7 +109,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
ExecutorFactory.Global.unsafeSet(globalExecutor);
Stage.READ.unsafeSetExecutor(unorderedScheduled);
Stage.MUTATION.unsafeSetExecutor(unorderedScheduled);
- for (Stage stage : Arrays.asList(Stage.MISC, Stage.ACCORD_MIGRATION))
+ for (Stage stage : Arrays.asList(Stage.MISC, Stage.ACCORD_MIGRATION, Stage.READ, Stage.MUTATION))
stage.unsafeSetExecutor(globalExecutor.configureSequential("ignore").build());
this.updateHolder = new CommandStore.EpochUpdateHolder();
@@ -152,6 +152,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
};
+ AccordStateCache stateCache = new AccordStateCache(Stage.READ.executor(), Stage.MUTATION.executor(), 8 << 20, new AccordStateCacheMetrics("test"));
this.journal = new MockJournal();
this.store = new AccordCommandStore(0,
timeService,
@@ -179,7 +180,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}),
updateHolder,
journal,
- new AccordStateCacheMetrics("test"));
+ new AccordCommandStore.CommandStoreExecutor(stateCache, executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + 0 + ']'), Thread.currentThread().getId()));
store.cache().instances().forEach(i -> {
i.register(new AccordStateCache.Listener()
@@ -213,11 +214,6 @@ public class SimulatedAccordCommandStore implements AutoCloseable
shouldCompact = boolSource(rs.fork());
}
- public Ranges slice(Ranges ranges)
- {
- return ranges.slice(topology.ranges(), Routables.Slice.Minimal);
- }
-
private static BooleanSupplier boolSource(RandomSource rs)
{
var gen = Gens.bools().mixedDistribution().next(rs);
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java
index e3526e30b9..961a1dfdb3 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java
@@ -18,16 +18,6 @@
package org.apache.cassandra.service.accord;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.junit.Test;
-
import accord.api.RoutingKey;
import accord.primitives.FullKeyRoute;
import accord.primitives.FullRangeRoute;
@@ -37,11 +27,23 @@ import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Txn;
import accord.primitives.TxnId;
+import accord.utils.Gen;
import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.utils.Generators;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import static accord.utils.Property.qt;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
@@ -73,6 +75,40 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
});
}
+ @Test
+ public void tokenConflicts()
+ {
+ TableMetadata tbl = reverseTokenTbl;
+ int numSamples = 100;
+ Gen<ByteBuffer> rawKey = Generators.toGen(Generators.bytes(16, 16));
+
+ qt().withExamples(10).check(rs -> {
+ AccordKeyspace.unsafeClear();
+
+ ByteBuffer key = rawKey.next(rs);
+ PartitionKey pk = new PartitionKey(tbl.id, tbl.partitioner.decorateKey(key));
+ Keys keys = Keys.of(pk);
+ FullKeyRoute route = keys.toRoute(pk.toUnseekable());
+ Txn txn = createTxn(wrapInTxn("INSERT INTO " + tbl + "(pk, value) VALUES (?, ?)"), Arrays.asList(key, 42));
+
+ ByteBuffer tokenConflictKey = Murmur3Partitioner.LongToken.keyForToken((LongToken) Murmur3Partitioner.instance.decorateKey(key).getToken());
+ PartitionKey pkTokenConflict = new PartitionKey(tbl.id, tbl.partitioner.decorateKey(tokenConflictKey));
+ Keys keysTokenConflict = Keys.of(pkTokenConflict);
+ FullKeyRoute routeTokenConflict = keysTokenConflict.toRoute(pkTokenConflict.toUnseekable());
+ Txn txnTokenConflict = createTxn(wrapInTxn("INSERT INTO " + tbl + "(pk, value) VALUES (?, ?)"), Arrays.asList(tokenConflictKey, 42));
+ try (var instance = new SimulatedAccordCommandStore(rs))
+ {
+ List<TxnId> conflicts = new ArrayList<>(numSamples);
+ for (int i = 0; i < numSamples; i++)
+ {
+ instance.maybeCacheEvict(route, Ranges.EMPTY);
+ conflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), txn, route, keyConflicts(conflicts, route)));
+ conflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), txnTokenConflict, routeTokenConflict, keyConflicts(conflicts, routeTokenConflict)));
+ }
+ }
+ });
+ }
+
@Test
public void rangePartialKeyMatch()
{
@@ -86,7 +122,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
{
long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
Ranges partialRange = Ranges.of(tokenRange(tbl.id, token - 1, token));
- Ranges partialRangeSliced = instance.slice(partialRange);
+
long outOfRangeToken = token - 10;
if (outOfRangeToken == Long.MIN_VALUE) // if this wraps around that is fine, just can't be min
outOfRangeToken++;
@@ -120,7 +156,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
keyConflicts.add(id);
outOfRangeKeyConflicts.add(id);
- rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, Map.of(key, keyConflicts), rangeConflicts(rangeConflicts, partialRangeSliced)));
+ rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, Map.of(key, keyConflicts), rangeConflicts(rangeConflicts, partialRange)));
}
}
});
@@ -156,7 +192,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
{
instance.maybeCacheEvict(keyRoute, ranges);
keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keyRoute)));
- rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keyRoute), rangeConflicts(rangeConflicts, instance.slice(ranges))));
+ rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keyRoute), rangeConflicts(rangeConflicts, ranges)));
}
}
});
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java
index 9e9ac1fad7..30f0f0aa7f 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java
@@ -18,64 +18,92 @@
package org.apache.cassandra.service.accord;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.junit.Test;
-
import accord.api.RoutingKey;
import accord.primitives.FullRangeRoute;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.Ranges;
+import accord.primitives.RoutingKeys;
import accord.primitives.Txn;
import accord.primitives.TxnId;
+import accord.utils.Property;
+import accord.utils.RandomSource;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.utils.FailingConsumer;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
-import static accord.utils.Property.qt;
+import static accord.utils.Property.commands;
+import static accord.utils.Property.stateful;
import static org.apache.cassandra.dht.Murmur3Partitioner.LongToken.keyForToken;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCommandStoreTestBase
{
+ private static Property.SimpleCommand<State> insertKey(RandomSource rs, State state)
+ {
+ long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
+ RoutingKey key = new TokenKey(state.tbl.id, new LongToken(token));
+ Txn keyTxn = createTxn(wrapInTxn("INSERT INTO " + state.tbl + "(pk, value) VALUES (?, ?)"),
+ Arrays.asList(keyForToken(token), 42));
+ Keys keys = (Keys) keyTxn.keys();
+ FullRoute<RoutingKey> keyRoute = keys.toRoute(keys.get(0).toUnseekable());
+
+ return new Property.SimpleCommand<>("Write Txn: " + keys, FailingConsumer.orFail(s -> {
+ s.instance.maybeCacheEvict(keyRoute, s.wholeRange);
+ var k = assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, s.keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap());
+ s.keyConflicts.get(key).add(k);
+ }));
+ }
+
+ private static Property.SimpleCommand<State> insertRange(RandomSource rs, State state)
+ {
+ return new Property.SimpleCommand<>("Range Txn: " + state.wholeRange, FailingConsumer.orFail(s -> {
+ s.instance.maybeCacheEvict(RoutingKeys.EMPTY, s.wholeRange);
+ s.rangeConflicts.add(assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), s.rangeTxn, s.rangeRoute, s.keyConflicts, rangeConflicts(s.rangeConflicts, s.wholeRange)));
+ }));
+ }
+
+
@Test
public void keysAllOverConflictingWithRange()
{
- var tbl = reverseTokenTbl;
- Ranges wholeRange = Ranges.of(fullRange(tbl.id));
- FullRangeRoute rangeRoute = wholeRange.toRoute(wholeRange.get(0).end());
- Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, wholeRange);
- int numSamples = 300;
+ stateful().withSteps(State.steps).check(commands(() -> State::new)
+ .add(SimulatedRandomKeysWithRangeConflictTest::insertKey)
+ .add(SimulatedRandomKeysWithRangeConflictTest::insertRange)
+ .build());
+ }
- qt().withExamples(10).check(rs -> {
- AccordKeyspace.unsafeClear();
- try (var instance = new SimulatedAccordCommandStore(rs))
- {
- Map<RoutingKey, List<TxnId>> keyConflicts = new HashMap<>();
- List<TxnId> rangeConflicts = new ArrayList<>(numSamples);
- for (int i = 0; i < numSamples; i++)
- {
- long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
- RoutingKey key = new TokenKey(tbl.id, new LongToken(token));
- Txn keyTxn = createTxn(wrapInTxn("INSERT INTO " + tbl + "(pk, value) VALUES (?, ?)"),
- Arrays.asList(keyForToken(token), 42));
- Keys keys = (Keys) keyTxn.keys();
- FullRoute<RoutingKey> keyRoute = keys.toRoute(keys.get(0).toUnseekable());
+ public static class State
+ {
+ static final int steps = 300;
+ final SimulatedAccordCommandStore instance;
+ final Map<RoutingKey, List<TxnId>> keyConflicts = new HashMap<>();
+ final List<TxnId> rangeConflicts = new ArrayList<>(steps);
+
+ final TableMetadata tbl = reverseTokenTbl;
+ final Ranges wholeRange = Ranges.of(fullRange(tbl.id));
+ final FullRangeRoute rangeRoute = wholeRange.toRoute(wholeRange.get(0).end());
+ final Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, wholeRange);
- instance.maybeCacheEvict(keyRoute, wholeRange);
+ public State(RandomSource rs)
+ {
+ AccordKeyspace.unsafeClear();
+ this.instance = new SimulatedAccordCommandStore(rs);
+ }
- // the full range is (-Inf, +Inf] but the store could be [(-Inf, Number], (Number, +Inf]], so need to slice to the store to get a matching range
- Ranges wholeRangeSlicedShard = instance.slice(wholeRange);
- var k = assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap());
- keyConflicts.get(key).add(k);
- rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts, rangeConflicts(rangeConflicts, wholeRangeSlicedShard)));
- }
- }
- });
+ @Override
+ public String toString()
+ {
+ return "Storage Ranges: " + instance.topology.ranges();
+ }
}
}
diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
index 3a29e63e99..78af413da8 100644
--- a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java
@@ -97,7 +97,7 @@ public class AsyncLoaderTest
AccordCommandStore commandStore =
createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor);
AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache = commandStore.commandCache();
- commandStore.executeBlocking(() -> commandStore.setCapacity(1024));
+ commandStore.executeBlocking(() -> commandStore.cache().setCapacity(1024));
AccordStateCache.Instance<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsCache = commandStore.timestampsForKeyCache();
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
@@ -242,7 +242,7 @@ public class AsyncLoaderTest
ManualExecutor executor = new ManualExecutor();
AccordCommandStore commandStore =
createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor);
- commandStore.executor().submit(() -> commandStore.setCapacity(1024)).get();
+ commandStore.executor().submit(() -> commandStore.cache().setCapacity(1024)).get();
AccordStateCache.Instance<TxnId, Command, AccordSafeCommand> commandCache = commandStore.commandCache();
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
PartialTxn txn = createPartialTxn(0);
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 31988224b4..20f123572f 100644
--- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
@@ -223,9 +223,9 @@ public class AsyncOperationTest
// clear cache
commandStore.executeBlocking(() -> {
- long cacheSize = commandStore.capacity();
- commandStore.setCapacity(0);
- commandStore.setCapacity(cacheSize);
+ long cacheSize = commandStore.cache().capacity();
+ commandStore.cache().setCapacity(0);
+ commandStore.cache().setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
@@ -272,9 +272,9 @@ public class AsyncOperationTest
// clear cache
commandStore.executeBlocking(() -> {
- long cacheSize = commandStore.capacity();
- commandStore.setCapacity(0);
- commandStore.setCapacity(cacheSize);
+ long cacheSize = commandStore.cache().capacity();
+ commandStore.cache().setCapacity(0);
+ commandStore.cache().setCapacity(cacheSize);
commandStore.cache().awaitSaveResults();
});
@@ -365,7 +365,7 @@ public class AsyncOperationTest
// all txn use the same key; 0
Keys keys = keys(Schema.instance.getTableMetadata("ks", "tbl"), 0);
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
- commandStore.executeBlocking(() -> commandStore.setCapacity(0));
+ commandStore.executeBlocking(() -> commandStore.cache().setCapacity(0));
Gen<TxnId> txnIdGen = rs -> txnId(1, clock.incrementAndGet(), 1);
qt().withPure(false)
diff --git a/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java b/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
index fd37346aa0..128fdeca7b 100644
--- a/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
+++ b/test/unit/org/apache/cassandra/tcm/ValidatingClusterMetadataService.java
@@ -20,11 +20,17 @@ package org.apache.cassandra.tcm;
import java.io.IOException;
import java.util.List;
+import java.util.TreeMap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.tcm.log.Entry;
+import org.apache.cassandra.tcm.log.LogState;
+import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializers;
import org.apache.cassandra.tcm.serialization.Version;
@@ -33,6 +39,7 @@ import org.assertj.core.api.Assertions;
public class ValidatingClusterMetadataService extends StubClusterMetadataService
{
private final List<Version> supportedVersions;
+ private final TreeMap<Epoch, ClusterMetadata> epochs = new TreeMap<>();
private ValidatingClusterMetadataService(List<Version> supportedVersions)
{
@@ -102,6 +109,62 @@ public class ValidatingClusterMetadataService extends StubClusterMetadataService
if (!metadata.epoch.equals(metadata().epoch.nextEpoch()))
throw new AssertionError("Epochs were not sequential: expected " + metadata().epoch.nextEpoch() + " but given " + metadata.epoch);
testSerde(ClusterMetadata.serializer, metadata);
+ epochs.put(metadata.epoch, metadata);
super.setMetadata(metadata);
}
+
+ @Override
+ public Processor processor()
+ {
+ Processor delegate = super.processor();
+ return new Processor()
+ {
+ @Override
+ public Commit.Result commit(Entry.Id entryId, Transformation transform, Epoch lastKnown, Retry.Deadline retryPolicy)
+ {
+ return delegate.commit(entryId, transform, lastKnown, retryPolicy);
+ }
+
+ @Override
+ public ClusterMetadata fetchLogAndWait(Epoch waitFor, Retry.Deadline retryPolicy)
+ {
+ return delegate.fetchLogAndWait(waitFor, retryPolicy);
+ }
+
+ @Override
+ public LogState reconstruct(Epoch lowEpoch, Epoch highEpoch, Retry.Deadline retryPolicy)
+ {
+ if (!epochs.containsKey(lowEpoch))
+ throw new AssertionError("Unknown epoch: " + lowEpoch);
+ ClusterMetadata base = epochs.get(lowEpoch);
+ ImmutableList.Builder<Entry> entries = ImmutableList.builder();
+ int id = 0;
+ for (ClusterMetadata cm : epochs.subMap(lowEpoch, false, highEpoch, true).values())
+ entries.add(new Entry(new Entry.Id(id++), cm.epoch, new MockTransformer(cm)));
+ return new LogState(base, entries.build());
+ }
+ };
+ }
+
+ private static class MockTransformer implements Transformation
+ {
+ private final ClusterMetadata result;
+
+ private MockTransformer(ClusterMetadata result)
+ {
+ this.result = result;
+ }
+
+ @Override
+ public Kind kind()
+ {
+ return null;
+ }
+
+ @Override
+ public Result execute(ClusterMetadata metadata)
+ {
+ return new Success(result, LockedRanges.AffectedRanges.EMPTY, ImmutableSet.of());
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org