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 08:40:01 UTC
(cassandra) branch cep-15-accord updated: Follow-up to CASSANDRA-19967 and CASSANDRA-19869
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new 59713bec80 Follow-up to CASSANDRA-19967 and CASSANDRA-19869
59713bec80 is described below
commit 59713bec8046535ad5dabdfbb0dee60985a53b8f
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Tue Oct 1 10:29:45 2024 +0200
Follow-up to CASSANDRA-19967 and CASSANDRA-19869
---
modules/accord | 2 +-
.../org/apache/cassandra/config/AccordSpec.java | 12 +-
.../db/compaction/CompactionIterator.java | 77 +++----
.../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 | 21 +-
.../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 | 11 +-
.../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, 865 insertions(+), 474 deletions(-)
diff --git a/modules/accord b/modules/accord
index e2e72287a7..bb43d32178 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit e2e72287a70bfff4cf6a5639d4a60a0c9e8086b4
+Subproject commit bb43d3217809a7ba3d5e37d2db6becea0363070f
diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java
index 0350e587c6..c39ed9d11d 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();
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 17160c4fa8..4afe936836 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,30 @@ 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(recordColumn.name.toString(), commandBuilder.asByteBuffer(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 +1122,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 +1141,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
{
throw new RuntimeException(e);
}
- return null;
}
@Override
@@ -1157,7 +1151,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..778ae506ad 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;
@@ -80,10 +81,11 @@ public class AccordLoadTest extends AccordTestBase
}).drop();
ICoordinator coordinator = cluster.coordinator(1);
final int repairInterval = 3000;
- final int batchSize = 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;
final BitSet initialised = new BitSet();
@@ -92,13 +94,15 @@ public class AccordLoadTest extends AccordTestBase
// 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 +116,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 +126,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)
@@ -139,7 +146,7 @@ public class AccordLoadTest extends AccordTestBase
}
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..9e84ff191d 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
@@ -81,6 +81,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 +110,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 +153,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 +181,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 +215,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