From 22df42c7bec23b77de670ce1773e76424225ac00 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Fri, 11 Oct 2024 16:27:33 +0100 Subject: [PATCH] Store historical transactions per epoch update durability scheduling and majority deps fetching do not deserialize deps in CommandsForRangesLoader unless required AccordJournalPurger should use shouldCleanupPartial load historical transactions when loading topology --- modules/accord | 2 +- .../apache/cassandra/concurrent/Stage.java | 33 +++---- .../apache/cassandra/config/AccordSpec.java | 22 +++-- .../cassandra/config/DatabaseDescriptor.java | 28 +++++- .../org/apache/cassandra/journal/Journal.java | 1 + .../org/apache/cassandra/net/Message.java | 39 ++++++-- .../org/apache/cassandra/schema/TableId.java | 17 +++- .../cassandra/service/CassandraDaemon.java | 3 +- .../service/accord/AccordCachingState.java | 7 +- .../service/accord/AccordCommandStore.java | 78 ++++++++++----- .../service/accord/AccordCommandStores.java | 18 ---- .../accord/AccordConfigurationService.java | 8 +- .../accord/AccordFastPathCoordinator.java | 2 +- .../service/accord/AccordJournal.java | 33 ++++++- .../accord/AccordJournalValueSerializers.java | 28 ++++-- .../service/accord/AccordMessageSink.java | 30 +++--- .../accord/AccordResponseVerbHandler.java | 16 ++++ .../accord/AccordSafeCommandStore.java | 23 ++++- .../service/accord/AccordService.java | 7 +- .../service/accord/AccordStateCache.java | 36 +++++-- .../service/accord/AccordVerbHandler.java | 9 +- .../accord/CommandsForRangesLoader.java | 77 ++++++++++++--- .../cassandra/service/accord/IJournal.java | 5 +- .../cassandra/service/accord/JournalKey.java | 9 +- .../service/accord/SavedCommand.java | 90 +++++++++++++++--- .../service/accord/api/AccordRoutingKey.java | 4 +- .../service/accord/api/PartitionKey.java | 7 +- .../service/accord/async/AsyncLoader.java | 94 +++++++++++++------ .../service/accord/async/AsyncOperation.java | 11 ++- .../apache/cassandra/utils/NoSpamLogger.java | 15 ++- .../cassandra/utils/memory/BufferPool.java | 5 +- .../test/accord/AccordLoadTest.java | 5 +- .../accord/AccordJournalCompactionTest.java | 11 ++- .../config/YamlConfigurationLoaderTest.java | 6 +- .../service/accord/EpochSyncTest.java | 9 +- .../cassandra/service/accord/MockJournal.java | 19 +++- .../service/accord/SavedCommandTest.java | 4 +- 37 files changed, 602 insertions(+), 209 deletions(-) diff --git a/modules/accord b/modules/accord index dd04c81214..d914ee6981 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit dd04c81214be9213cf503025bf6ed4cbf118be00 +Subproject commit d914ee69816ebfdf88b2120ff1d8e0bc16edecbc diff --git a/src/java/org/apache/cassandra/concurrent/Stage.java b/src/java/org/apache/cassandra/concurrent/Stage.java index 135b5d078e..910ac40a5b 100644 --- a/src/java/org/apache/cassandra/concurrent/Stage.java +++ b/src/java/org/apache/cassandra/concurrent/Stage.java @@ -43,22 +43,23 @@ import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized; public enum Stage { - READ (false, "ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage), - MUTATION (true, "MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage), - COUNTER_MUTATION (true, "CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage), - VIEW_MUTATION (true, "ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage), - ACCORD_MIGRATION (false, "AccordMigrationReadStage", "request", DatabaseDescriptor::getConcurrentAccordOps, DatabaseDescriptor::setConcurrentAccordOps, Stage::multiThreadedLowSignalStage), - GOSSIP (true, "GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage), - REQUEST_RESPONSE (false, "RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage), - ANTI_ENTROPY (false, "AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage), - MIGRATION (false, "MigrationStage", "internal", () -> 1, null, Stage::migrationStage), - MISC (false, "MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage), - TRACING (false, "TracingStage", "internal", () -> 1, null, Stage::tracingStage), - INTERNAL_RESPONSE (false, "InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), - IMMEDIATE (false, "ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor), - PAXOS_REPAIR (false, "PaxosRepairStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), - INTERNAL_METADATA (false, "InternalMetadataStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), - FETCH_LOG (false, "MetadataFetchLogStage", "internal", () -> 1, null, Stage::singleThreadedStage) + READ (false, "ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage), + MUTATION (true, "MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage), + COUNTER_MUTATION (true, "CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage), + VIEW_MUTATION (true, "ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage), + ACCORD_MIGRATION (false, "AccordMigrationStage", "request", DatabaseDescriptor::getConcurrentAccordOps, DatabaseDescriptor::setConcurrentAccordOps, Stage::multiThreadedLowSignalStage), + GOSSIP (true, "GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage), + REQUEST_RESPONSE (false, "RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage), + ANTI_ENTROPY (false, "AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage), + MIGRATION (false, "MigrationStage", "internal", () -> 1, null, Stage::migrationStage), + MISC (false, "MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage), + TRACING (false, "TracingStage", "internal", () -> 1, null, Stage::tracingStage), + INTERNAL_RESPONSE (false, "InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), + IMMEDIATE (false, "ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor), + PAXOS_REPAIR (false, "PaxosRepairStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), + INTERNAL_METADATA (false, "InternalMetadataStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), + FETCH_LOG (false, "MetadataFetchLogStage", "internal", () -> 1, null, Stage::singleThreadedStage), + ACCORD_RANGE_LOADER(false, "AccordRangeLoader", "internal", () -> 4, null, Stage::multiThreadedStage), ; public final String jmxName; private final Supplier executorSupplier; diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index 3efcac6566..74504326ab 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -18,14 +18,14 @@ package org.apache.cassandra.config; -import accord.primitives.Routable; -import accord.primitives.Txn; +import java.util.concurrent.TimeUnit; + import accord.primitives.TxnId; import com.fasterxml.jackson.annotation.JsonIgnore; import org.apache.cassandra.journal.Params; import org.apache.cassandra.service.consensus.TransactionalMode; -import java.util.concurrent.TimeUnit; +import static accord.primitives.Routable.Domain.Range; public class AccordSpec { @@ -39,13 +39,13 @@ public class AccordSpec // TODO (expected): we should be able to support lower recover delays, at least for txns public volatile DurationSpec.IntMillisecondsBound recover_delay = new DurationSpec.IntMillisecondsBound(5000); - public volatile DurationSpec.IntMillisecondsBound range_sync_recover_delay = new DurationSpec.IntMillisecondsBound(10000); + public volatile DurationSpec.IntMillisecondsBound range_sync_recover_delay = new DurationSpec.IntMillisecondsBound("5m"); public String slowPreAccept = "30ms <= p50*2 <= 100ms"; public String slowRead = "30ms <= p50*2 <= 100ms"; public long recoveryDelayFor(TxnId txnId, TimeUnit unit) { - if (txnId.kind() == Txn.Kind.SyncPoint && txnId.domain() == Routable.Domain.Range) + if (txnId.isSyncPoint() && txnId.is(Range)) return range_sync_recover_delay.to(unit); return recover_delay.to(unit); } @@ -65,13 +65,15 @@ public class AccordSpec public DurationSpec.IntMillisecondsBound range_barrier_timeout = new DurationSpec.IntMillisecondsBound("2m"); - public volatile DurationSpec.IntSecondsBound fast_path_update_delay = new DurationSpec.IntSecondsBound("3600s"); + public volatile DurationSpec.IntSecondsBound fast_path_update_delay = new DurationSpec.IntSecondsBound("60m"); - public volatile DurationSpec.IntSecondsBound gc_delay = new DurationSpec.IntSecondsBound(300); - public volatile DurationSpec.IntSecondsBound schedule_durability_frequency = new DurationSpec.IntSecondsBound(120); - public volatile DurationSpec.IntSecondsBound durability_txnid_lag = new DurationSpec.IntSecondsBound(10); - public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(5, TimeUnit.MINUTES); + public volatile DurationSpec.IntSecondsBound gc_delay = new DurationSpec.IntSecondsBound("5m"); + public volatile int shard_durability_target_splits = 128; + public volatile DurationSpec.IntSecondsBound durability_txnid_lag = new DurationSpec.IntSecondsBound(5); + public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(15, TimeUnit.MINUTES); public volatile DurationSpec.IntSecondsBound global_durability_cycle = new DurationSpec.IntSecondsBound(10, TimeUnit.MINUTES); + public volatile DurationSpec.IntSecondsBound default_durability_retry_delay = new DurationSpec.IntSecondsBound(10, TimeUnit.SECONDS); + public volatile DurationSpec.IntSecondsBound max_durability_retry_delay = new DurationSpec.IntSecondsBound(10, TimeUnit.MINUTES); public enum TransactionalRangeMigration { diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 1bd8f0f335..05f15148d8 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -5343,14 +5343,14 @@ public class DatabaseDescriptor return conf.accord.gc_delay.to(unit); } - public static long getAccordScheduleDurabilityFrequency(TimeUnit unit) + public static int getAccordShardDurabilityTargetSplits() { - return conf.accord.schedule_durability_frequency.to(unit); + return conf.accord.shard_durability_target_splits; } - public static void setAccordScheduleDurabilityFrequencySeconds(long seconds) + public static void setAccordShardDurabilityTargetSplits(int number) { - conf.accord.schedule_durability_frequency = new DurationSpec.IntSecondsBound(seconds); + conf.accord.shard_durability_target_splits = number; } public static long getAccordScheduleDurabilityTxnIdLag(TimeUnit unit) @@ -5373,6 +5373,26 @@ public class DatabaseDescriptor conf.accord.global_durability_cycle = new DurationSpec.IntSecondsBound(seconds); } + public static long getAccordDefaultDurabilityRetryDelay(TimeUnit unit) + { + return conf.accord.default_durability_retry_delay.to(unit); + } + + public static void setAccordDefaultDurabilityRetryDelaySeconds(long seconds) + { + conf.accord.default_durability_retry_delay = new DurationSpec.IntSecondsBound(seconds); + } + + public static long getAccordMaxDurabilityRetryDelay(TimeUnit unit) + { + return conf.accord.max_durability_retry_delay.to(unit); + } + + public static void setAccordMaxDurabilityRetryDelaySeconds(long seconds) + { + conf.accord.max_durability_retry_delay = new DurationSpec.IntSecondsBound(seconds); + } + public static long getAccordShardDurabilityCycle(TimeUnit unit) { return conf.accord.shard_durability_cycle.to(unit); diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 5501146d8d..34aff492ea 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -130,6 +130,7 @@ public class Journal implements Shutdownable @Override public void onFlush(long segment, int position) { + // TODO (required): this seems to be a big source of allocations waitingFor.drain(drained::add); List remaining = new ArrayList<>(); for (WaitingFor wait : drained) diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java index 77edd1d0a2..4bf114972b 100644 --- a/src/java/org/apache/cassandra/net/Message.java +++ b/src/java/org/apache/cassandra/net/Message.java @@ -91,6 +91,12 @@ public class Message implements ResponseContext this.payloadSerializer = verb().serializer(); } + /** Whether the message has crossed the node boundary, that is whether it originated from another node. */ + public boolean isCrossNode() + { + return !from().equals(getBroadcastAddressAndPort()); + } + /** Sender of the message. */ @Override public InetAddressAndPort from() @@ -98,12 +104,6 @@ public class Message implements ResponseContext return header.from; } - /** Whether the message has crossed the node boundary, that is whether it originated from another node. */ - public boolean isCrossNode() - { - return !from().equals(getBroadcastAddressAndPort()); - } - /** * id of the request/message. In 4.0+ can be shared between multiple messages of the same logical request, * whilst in versions above a new id would be allocated for each message sent. @@ -520,7 +520,7 @@ public class Message implements ResponseContext * Split into a separate object to allow partial message deserialization without wasting work and allocation * afterwards, if the entire message is necessary and available. */ - public static class Header + public static class Header implements ResponseContext { public final long id; public final Epoch epoch; @@ -607,6 +607,31 @@ public class Message implements ResponseContext return respondTo; } + /** Sender of the message. */ + @Override + public InetAddressAndPort from() + { + return from; + } + + @Override + public long id() + { + return id; + } + + @Override + public Verb verb() + { + return verb; + } + + @Override + public long expiresAtNanos() + { + return expiresAtNanos; + } + @Nullable public TimeUUID traceSession() { diff --git a/src/java/org/apache/cassandra/schema/TableId.java b/src/java/org/apache/cassandra/schema/TableId.java index 03fd3dc490..5cdaf1c2b6 100644 --- a/src/java/org/apache/cassandra/schema/TableId.java +++ b/src/java/org/apache/cassandra/schema/TableId.java @@ -22,9 +22,12 @@ import java.io.DataOutput; import java.io.IOException; import java.nio.ByteBuffer; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.commons.lang3.ArrayUtils; @@ -53,6 +56,8 @@ public class TableId implements Comparable public static final long MAGIC = 1956074401491665062L; public static final long EMPTY_SIZE = ObjectSizes.measureDeep(new UUID(0, 0)); + private static final ConcurrentHashMap internCache = new ConcurrentHashMap<>(); + private final UUID id; private TableId(UUID id) @@ -200,10 +205,10 @@ public class TableId implements Comparable return new TableId(new UUID(accessor.getLong(src, offset), accessor.getLong(src, offset + TypeSizes.LONG_SIZE))); } - public TableId tryIntern() + public TableId intern() { - TableMetadata metadata = Schema.instance.getTableMetadata(this); - return metadata == null ? this : metadata.id; + TableId interned = internCache.putIfAbsent(this, this); + return interned == null ? this : interned; } @Override @@ -253,4 +258,10 @@ public class TableId implements Comparable return t.serializedSize(); } }; + + public static void scheduleCachePruning() + { + ScheduledExecutors.scheduledFastTasks.scheduleSelfRecurring(internCache::clear, 1, TimeUnit.HOURS); + } + } diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 45740f1916..7f628f8931 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -77,6 +77,7 @@ import org.apache.cassandra.metrics.DefaultNameFactory; import org.apache.cassandra.net.StartupClusterConnectivityChecker; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.TableId; import org.apache.cassandra.security.ThreadAwareSecurityManager; import org.apache.cassandra.service.accord.AccordOperations; import org.apache.cassandra.service.paxos.PaxosState; @@ -422,7 +423,7 @@ public class CassandraDaemon logger.info("Prewarming of auth caches is disabled"); PaxosState.startAutoRepairs(); - + TableId.scheduleCachePruning(); completeSetup(); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCachingState.java b/src/java/org/apache/cassandra/service/accord/AccordCachingState.java index b8bb61e00c..52d6b7c090 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCachingState.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCachingState.java @@ -547,7 +547,12 @@ public class AccordCachingState extends IntrusiveLinkedListNode Saving(V current, Runnable saveRunnable) { - super(() -> { saveRunnable.run(); return null; }); + this(current, () -> { saveRunnable.run(); return null; }); + } + + Saving(V current, Callable saveCallable) + { + super(saveCallable); this.current = current; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index a7a89698dd..3157fb8355 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -28,6 +28,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.IntFunction; @@ -38,6 +39,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.Agent; +import accord.api.ConfigurationService; import accord.api.DataStore; import accord.api.LocalListeners; import accord.api.ProgressLog; @@ -49,6 +51,7 @@ import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.Commands; import accord.local.KeyHistory; +import accord.local.Node; import accord.local.NodeCommandStoreService; import accord.local.PreLoadContext; import accord.local.RedundantBefore; @@ -67,23 +70,30 @@ import accord.primitives.TxnId; import accord.utils.Invariants; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; +import accord.utils.async.AsyncResult; +import accord.utils.async.AsyncResults; import org.apache.cassandra.cache.CacheSize; 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.service.accord.SavedCommand.MinimalCommand; 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; import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Promise; import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; +import static accord.api.ConfigurationService.EpochReady.DONE; +import static accord.local.KeyHistory.COMMANDS; import static accord.primitives.Status.Committed; import static accord.primitives.Status.Invalidated; import static accord.primitives.Status.Truncated; import static accord.utils.Invariants.checkState; +import static org.apache.cassandra.service.accord.SavedCommand.Load.MINIMAL; public class AccordCommandStore extends CommandStore { @@ -218,7 +228,6 @@ public class AccordCommandStore extends CommandStore loadBootstrapBeganAt(journal.loadBootstrapBeganAt(id())); loadSafeToRead(journal.loadSafeToRead(id())); loadRangesForEpoch(journal.loadRangesForEpoch(id())); - loadHistoricalTransactions(journal.loadHistoricalTransactions(id())); executor.execute(() -> CommandStore.register(this)); } @@ -488,7 +497,41 @@ public class AccordCommandStore extends CommandStore { } - public void registerHistoricalTransactions(Deps deps, SafeCommandStore safeStore) + protected ConfigurationService.EpochReady syncInternal(Node node, Ranges ranges, long epoch, boolean isLoad) + { + if (!isLoad) + return super.syncInternal(node, ranges, epoch, false); + + List> loaded = journal.loadHistoricalTransactions(epoch, id); + // synchronously load and register historical, so we don't have unlimited numbers of epochs in flight + for (Pair pair : loaded) + { + cancelFetch(pair.left, epoch); + try + { + logger.info("Restoring sync'd deps for {} at epoch {}", pair.left, epoch); + AsyncChains.getBlocking(submit(PreLoadContext.contextFor(null, pair.right.keyDeps.keys(), COMMANDS), safeStore -> { + registerHistoricalTransactions(pair.left, pair.right, safeStore); + return null; + }).beginAsResult(), 5L, TimeUnit.MINUTES); + } + catch (InterruptedException | TimeoutException | ExecutionException e) + { + throw new RuntimeException(e); + } + ranges = ranges.without(Ranges.of(pair.left)); + } + + if (ranges.isEmpty()) + { + AsyncResult done = AsyncResults.success(null); + return new ConfigurationService.EpochReady(epoch, DONE, done, done, done); + } + + return super.syncInternal(node, ranges, epoch, false); + } + + public void registerHistoricalTransactions(Range range, Deps deps, SafeCommandStore safeStore) { if (deps.isEmpty()) return; @@ -509,19 +552,19 @@ public class AccordCommandStore extends CommandStore }); for (int i = 0; i < deps.rangeDeps.rangeCount(); i++) { - Range range = deps.rangeDeps.range(i); - if (!allRanges.intersects(range)) + Range r = deps.rangeDeps.range(i); + if (!allRanges.intersects(r)) continue; - deps.rangeDeps.forEach(range, txnId -> { + deps.rangeDeps.forEach(r, txnId -> { // TODO (desired, efficiency): this can be made more efficient by batching by epoch - if (ranges.coordinates(txnId).intersects(range)) + if (ranges.coordinates(txnId).intersects(r)) return; // already coordinates, no need to replicate - if (!ranges.allBefore(txnId.epoch()).intersects(range)) + if (!ranges.allBefore(txnId.epoch()).intersects(r)) return; // TODO (required): this is potentially not safe - it should not be persisted until we save in journal // but, preferable to retire historical transactions as a concept entirely, and rely on ExclusiveSyncPoints instead - diskCommandsForRanges().mergeHistoricalTransaction(txnId, Ranges.single(range).slice(allRanges), Ranges::with); + diskCommandsForRanges().mergeHistoricalTransaction(txnId, Ranges.single(r).slice(allRanges), Ranges::with); }); } } @@ -542,6 +585,11 @@ public class AccordCommandStore extends CommandStore return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore()); } + public MinimalCommand loadMinimal(TxnId txnId) + { + return journal.loadMinimal(id, txnId, MINIMAL, unsafeGetRedundantBefore(), durableBefore()); + } + public interface Loader { Promise load(Command next); @@ -581,7 +629,7 @@ public class AccordCommandStore extends CommandStore TxnId txnId = command.txnId(); AsyncPromise future = new AsyncPromise<>(); - execute(context(command, KeyHistory.COMMANDS), + execute(context(command, COMMANDS), safeStore -> { Command local = command; if (local.status() != Truncated && local.status() != Invalidated) @@ -663,18 +711,6 @@ public class AccordCommandStore extends CommandStore unsafeSetRangesForEpoch(new CommandStores.RangesForEpoch(rangesForEpoch.epochs, rangesForEpoch.ranges, this)); } - void loadHistoricalTransactions(List deps) - { - if (deps != null) - { - execute(PreLoadContext.empty(), - safeStore -> { - for (Deps dep : deps) - registerHistoricalTransactions(dep, safeStore); - }); - } - } - public static class CommandStoreExecutor implements CacheSize { final AccordStateCache stateCache; diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index 620bad8b16..af0bebdb4e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -22,10 +22,8 @@ import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; import accord.api.Agent; -import accord.api.ConfigurationService.EpochReady; import accord.api.DataStore; import accord.api.LocalListeners; import accord.api.ProgressLog; @@ -143,22 +141,6 @@ public class AccordCommandStores extends CommandStores implements CacheSize executor.execute(() -> executor.setCapacity(perExecutor)); } - @Override - public synchronized Supplier updateTopology(Node node, Topology newTopology, boolean startSync) - { - Supplier start = super.updateTopology(node, newTopology, startSync); - return () -> { - EpochReady ready = start.get(); - ready.metadata.addCallback(() -> { - synchronized (this) - { - refreshCacheSizes(); - } - }); - return ready; - }; - } - public void waitForQuiescense() { boolean hadPending; diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java index 0a2a192aa1..4f6f3d1666 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java @@ -234,7 +234,8 @@ public class AccordConfigurationService extends AbstractConfigurationService previousRef = new AtomicReference<>(null); diskState = diskStateManager.loadTopologies(((epoch, metadata, topology, syncStatus, pendingSyncNotify, remoteSyncComplete, closed, redundant) -> { updateMapping(metadata); - reportTopology(topology, syncStatus == SyncStatus.NOT_STARTED); + reportTopology(topology, syncStatus == SyncStatus.NOT_STARTED, true); + Topology previous = previousRef.get(); if (previous != null) { @@ -326,6 +327,11 @@ public class AccordConfigurationService extends AbstractConfigurationService onTopologyUpdate(Topology topology, boolean startSync) + public AsyncResult onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync) { updatePeers(topology); return SUCCESS; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 26b868e5d1..330c78e32c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -41,6 +41,7 @@ import accord.local.Node; import accord.local.RedundantBefore; import accord.local.cfk.CommandsForKey; import accord.primitives.Deps; +import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.SaveStatus; import accord.primitives.Timestamp; @@ -65,11 +66,13 @@ import org.apache.cassandra.service.accord.AccordJournalValueSerializers.Histori import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator; import org.apache.cassandra.service.accord.JournalKey.JournalKeySupport; import org.apache.cassandra.utils.ExecutorUtils; +import org.apache.cassandra.utils.Pair; 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; +import static org.apache.cassandra.service.accord.JournalKey.keyForHistoricalTransactions; public class AccordJournal implements IJournal, Shutdownable { @@ -193,6 +196,21 @@ public class AccordJournal implements IJournal, Shutdownable return builder.construct(); } + @Override + public SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId, SavedCommand.Load load, RedundantBefore redundantBefore, DurableBefore durableBefore) + { + SavedCommand.Builder builder = loadDiffs(commandStoreId, txnId, load); + Cleanup cleanup = builder.shouldCleanup(redundantBefore, durableBefore); + switch (cleanup) + { + case EXPUNGE_PARTIAL: + case EXPUNGE: + case ERASE: + return null; + } + return builder.asMinimal(); + } + @VisibleForTesting public RedundantBefore loadRedundantBefore(int store) { @@ -222,9 +240,9 @@ public class AccordJournal implements IJournal, Shutdownable } @Override - public List loadHistoricalTransactions(int store) + public List> loadHistoricalTransactions(long epoch, int store) { - HistoricalTransactionsAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store)); + HistoricalTransactionsAccumulator accumulator = readAll(keyForHistoricalTransactions(epoch, store)); return accumulator.get(); } @@ -287,7 +305,7 @@ public class AccordJournal implements IJournal, Shutdownable if (fieldUpdates.newRangesForEpoch != null) pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), fieldUpdates.newRangesForEpoch); if (fieldUpdates.addHistoricalTransactions != null) - pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.HISTORICAL_TRANSACTIONS, store), fieldUpdates.addHistoricalTransactions); + pointer = appendInternal(JournalKey.keyForHistoricalTransactions(fieldUpdates.addHistoricalTransactions.epoch, store), Pair.create(fieldUpdates.addHistoricalTransactions.range, fieldUpdates.addHistoricalTransactions.deps)); if (onFlush == null) return; @@ -299,14 +317,19 @@ public class AccordJournal implements IJournal, Shutdownable } @VisibleForTesting - public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId) + public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId, SavedCommand.Load load) { JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId); - SavedCommand.Builder builder = new SavedCommand.Builder(txnId); + SavedCommand.Builder builder = new SavedCommand.Builder(txnId, load); journalTable.readAll(key, builder::deserializeNext); return builder; } + public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId) + { + return loadDiffs(commandStoreId, txnId, SavedCommand.Load.ALL); + } + private BUILDER readAll(JournalKey key) { BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key); diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java index 60a1ef4f58..f232b9213e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java @@ -28,6 +28,7 @@ import com.google.common.collect.ImmutableSortedMap; import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.primitives.Deps; +import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Timestamp; import accord.primitives.TxnId; @@ -36,8 +37,10 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers; import org.apache.cassandra.service.accord.serializers.KeySerializers; +import org.apache.cassandra.utils.Pair; import static accord.local.CommandStores.RangesForEpoch; +import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL; import static org.apache.cassandra.service.accord.serializers.DepsSerializer.deps; // TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger @@ -63,7 +66,7 @@ public class AccordJournalValueSerializers @Override public SavedCommand.Builder mergerFor(JournalKey journalKey) { - return new SavedCommand.Builder(journalKey.id); + return new SavedCommand.Builder(journalKey.id, ALL); } @Override @@ -338,7 +341,7 @@ public class AccordJournalValueSerializers } } - public static class HistoricalTransactionsAccumulator extends Accumulator, Deps> + public static class HistoricalTransactionsAccumulator extends Accumulator>, Pair> { public HistoricalTransactionsAccumulator() { @@ -346,14 +349,14 @@ public class AccordJournalValueSerializers } @Override - protected List accumulate(List oldValue, Deps deps) + protected List> accumulate(List> oldValue, Pair deps) { accumulated.add(deps); // we can keep it mutable return accumulated; } } - public static class HistoricalTransactionsSerializer implements FlyweightSerializer + public static class HistoricalTransactionsSerializer implements FlyweightSerializer, HistoricalTransactionsAccumulator> { @Override public HistoricalTransactionsAccumulator mergerFor(JournalKey key) @@ -362,18 +365,22 @@ public class AccordJournalValueSerializers } @Override - public void serialize(JournalKey key, Deps from, DataOutputPlus out, int userVersion) throws IOException + public void serialize(JournalKey key, Pair from, DataOutputPlus out, int userVersion) throws IOException { out.writeUnsignedVInt32(1); - deps.serialize(from, out, messagingVersion); + TokenRange.serializer.serialize((TokenRange) from.left, out, messagingVersion); + deps.serialize(from.right, out, messagingVersion); } @Override public void reserialize(JournalKey key, HistoricalTransactionsAccumulator from, DataOutputPlus out, int userVersion) throws IOException { out.writeUnsignedVInt32(from.get().size()); - for (Deps d : from.get()) - deps.serialize(d, out, messagingVersion); + for (Pair d : from.get()) + { + TokenRange.serializer.serialize((TokenRange) d.left, out, messagingVersion); + deps.serialize(d.right, out, messagingVersion); + } } @Override @@ -381,7 +388,10 @@ public class AccordJournalValueSerializers { int count = in.readUnsignedVInt32(); for (int i = 0; i < count; i++) - into.update(deps.deserialize(in, messagingVersion)); + { + Range range = TokenRange.serializer.deserialize(in, messagingVersion); + into.update(Pair.create(range, deps.deserialize(in, messagingVersion))); + } } } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java index 756de92e3f..e42b08b993 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java +++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java @@ -26,36 +26,42 @@ import java.util.List; import java.util.Map; import java.util.Set; -import accord.impl.RequestCallbacks; -import accord.messages.*; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; - -import org.apache.cassandra.config.AccordSpec; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.metrics.ClientRequestsMetricsHolder; -import org.apache.cassandra.service.TimeoutStrategy; -import org.apache.cassandra.service.TimeoutStrategy.LatencySourceFactory; -import org.apache.cassandra.utils.Clock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.Agent; import accord.api.MessageSink; +import accord.impl.RequestCallbacks; import accord.local.AgentExecutor; import accord.local.Node; +import accord.messages.Callback; +import accord.messages.Commit; +import accord.messages.MessageType; +import accord.messages.Reply; +import accord.messages.ReplyContext; +import accord.messages.Request; +import accord.messages.TxnRequest; +import org.apache.cassandra.config.AccordSpec; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.ClientRequestsMetricsHolder; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessageDelivery; import org.apache.cassandra.net.MessageFlag; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.ResponseContext; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.service.TimeoutStrategy; +import org.apache.cassandra.service.TimeoutStrategy.LatencySourceFactory; +import org.apache.cassandra.utils.Clock; import static accord.messages.MessageType.Kind.REMOTE; +import static accord.primitives.Routable.Domain.Range; import static java.util.concurrent.TimeUnit.NANOSECONDS; public class AccordMessageSink implements MessageSink @@ -249,10 +255,10 @@ public class AccordMessageSink implements MessageSink return false; TxnRequest txnRequest = (TxnRequest) request; - if (!txnRequest.txnId.kind().isSyncPoint()) + if (!txnRequest.txnId.isSyncPoint()) return false; - return txnRequest.txnId.domain().isRange(); + return txnRequest.txnId.is(Range); } // TODO (expected): permit bulk send to save esp. on callback registration (and combine records) @@ -265,7 +271,7 @@ public class AccordMessageSink implements MessageSink long nowNanos = Clock.Global.nanoTime(); long delayedAtNanos = Long.MAX_VALUE; long expiresAtNanos; - if (isRangeBarrier(request) || verb == Verb.ACCORD_CALCULATE_DEPS_REQ) + if (isRangeBarrier(request)) expiresAtNanos = nowNanos + DatabaseDescriptor.getAccordRangeBarrierTimeoutNanos(); else expiresAtNanos = nowNanos + verb.expiresAfterNanos(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java index f5a06f9be8..6ee810fb04 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java +++ b/src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java @@ -17,6 +17,11 @@ */ package org.apache.cassandra.service.accord; +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import accord.coordinate.Timeout; import accord.impl.RequestCallbacks; import accord.local.Node; @@ -27,12 +32,17 @@ import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.tracing.Tracing; +import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime; class AccordResponseVerbHandler implements IVerbHandler { + private static final Logger logger = LoggerFactory.getLogger(AccordResponseVerbHandler.class); + private static final NoSpamLogStatement dropping = NoSpamLogger.getStatement(logger, "Dropping response {} from {}", 1L, TimeUnit.SECONDS); + private final RequestCallbacks callbacks; private final AccordEndpointMapper endpointMapper; @@ -45,6 +55,12 @@ class AccordResponseVerbHandler implements IVerbHandler @Override public void doVerb(Message message) { + if (!((AccordService)AccordService.instance()).shouldAcceptMessages()) + { + dropping.debug(message.verb(), message.from()); + return; + } + Node.Id from = endpointMapper.mappedId(message.from()); if (message.isFailureResponse()) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 4025049d39..c260a013c6 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -42,6 +42,7 @@ import accord.local.cfk.CommandsForKey; import accord.primitives.AbstractKeys; import accord.primitives.AbstractRanges; import accord.primitives.Deps; +import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Routables; import accord.primitives.Timestamp; @@ -336,12 +337,12 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore newBootstrapBeganAt; public NavigableMap newSafeToRead; public RangesForEpoch.Snapshot newRangesForEpoch; - public Deps addHistoricalTransactions; + public HistoricalTransactions addHistoricalTransactions; + } + + public static class HistoricalTransactions + { + public final long epoch; + public final Range range; + public final Deps deps; + + public HistoricalTransactions(long epoch, Range range, Deps deps) + { + this.epoch = epoch; + this.range = range; + this.deps = deps; + } } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 95188cddd4..dcb6fdf70d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -463,7 +463,7 @@ public class AccordService implements IAccordService, Shutdownable ? tcmLoadRange(optMaxEpoch.getAsLong(), Long.MAX_VALUE) : discoverHistoric(node, cms); for (ClusterMetadata m : historic) - configService.reportMetadataInternal(m); + configService.reportMetadataInternal(m, true); })); ClusterMetadata current = cms.metadata(); if (!ref.historic.isEmpty()) @@ -499,10 +499,12 @@ public class AccordService implements IAccordService, Shutdownable fastPathCoordinator.start(); cms.log().addListener(fastPathCoordinator); + durabilityScheduling.setDefaultRetryDelay(Ints.checkedCast(DatabaseDescriptor.getAccordDefaultDurabilityRetryDelay(SECONDS)), SECONDS); + durabilityScheduling.setMaxRetryDelay(Ints.checkedCast(DatabaseDescriptor.getAccordMaxDurabilityRetryDelay(SECONDS)), SECONDS); + durabilityScheduling.setTargetShardSplits(Ints.checkedCast(DatabaseDescriptor.getAccordShardDurabilityTargetSplits())); durabilityScheduling.setGlobalCycleTime(Ints.checkedCast(DatabaseDescriptor.getAccordGlobalDurabilityCycle(SECONDS)), SECONDS); durabilityScheduling.setShardCycleTime(Ints.checkedCast(DatabaseDescriptor.getAccordShardDurabilityCycle(SECONDS)), SECONDS); durabilityScheduling.setTxnIdLag(Ints.checkedCast(DatabaseDescriptor.getAccordScheduleDurabilityTxnIdLag(SECONDS)), TimeUnit.SECONDS); - durabilityScheduling.setFrequency(Ints.checkedCast(DatabaseDescriptor.getAccordScheduleDurabilityFrequency(SECONDS)), SECONDS); durabilityScheduling.start(); state = State.STARTED; } @@ -1267,6 +1269,7 @@ public class AccordService implements IAccordService, Shutdownable @Override public boolean awaitTermination(long timeout, TimeUnit units) { + // TODO (required): expose awaitTermination in Node // node doesn't offer return true; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java index ccbc90fdb2..4c3f068976 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java @@ -28,6 +28,8 @@ import java.util.function.Function; import java.util.function.ToLongFunction; import java.util.stream.Stream; +import javax.annotation.Nullable; + import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -365,7 +367,7 @@ public class AccordStateCache extends IntrusiveLinkedList l.onAdd(finalNode)); } } - AccordCachingState acquired = acquireExisting(node, true); + AccordCachingState acquired = acquireExisting(node, true, null); Invariants.checkState(acquired != null, "%s could not be acquired", node); return safeRefFactory.apply(acquired); } @@ -378,7 +380,7 @@ public class AccordStateCache extends IntrusiveLinkedList node = acquire(key, false); - return safeRefFactory.apply(node); + return acquire(key, null); } public S acquireIfLoaded(K key) { - AccordCachingState node = acquire(key, true); + return acquireIfLoaded(key, null); + } + + public S acquire(K key, @Nullable ExecutorPlus loadExecutor) + { + AccordCachingState node = acquire(key, false, loadExecutor); + return safeRefFactory.apply(node); + } + + public S acquireIfLoaded(K key, @Nullable ExecutorPlus loadExecutor) + { + AccordCachingState node = acquire(key, true, loadExecutor); if (node == null) return null; return safeRefFactory.apply(node); } - private AccordCachingState acquire(K key, boolean onlyIfLoaded) + private AccordCachingState acquire(K key, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor) { incrementCacheQueries(); @SuppressWarnings("unchecked") AccordCachingState node = (AccordCachingState) cache.get(key); return node == null - ? acquireAbsent(key, onlyIfLoaded) - : acquireExisting(node, onlyIfLoaded); + ? acquireAbsent(key, onlyIfLoaded, loadExecutor) + : acquireExisting(node, onlyIfLoaded, loadExecutor); } /* * Can only return a LOADING Node (or null) */ - private AccordCachingState acquireAbsent(K key, boolean onlyIfLoaded) + private AccordCachingState acquireAbsent(K key, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor) { incrementCacheMisses(); if (onlyIfLoaded) return null; AccordCachingState node = nodeFactory.create(key, index); + if (loadExecutor == null) + loadExecutor = AccordStateCache.this.loadExecutor; node.load(loadExecutor, loadFunction); node.references++; @@ -448,7 +462,7 @@ public class AccordStateCache extends IntrusiveLinkedList acquireExisting(AccordCachingState node, boolean onlyIfLoaded) + private AccordCachingState acquireExisting(AccordCachingState node, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor) { Status status = node.status(); // status() completes @@ -462,6 +476,8 @@ public class AccordStateCache extends IntrusiveLinkedList implements IVerbHandler { private static final Logger logger = LoggerFactory.getLogger(AccordVerbHandler.class); + private static final NoSpamLogger.NoSpamLogStatement dropping = NoSpamLogger.getStatement(logger, "Dropping message {} from {}", 1L, TimeUnit.SECONDS); private final Node node; private final AccordEndpointMapper endpointMapper; @@ -45,7 +48,7 @@ public class AccordVerbHandler implements IVerbHandler { if (!((AccordService)AccordService.instance()).shouldAcceptMessages()) { - logger.debug("Dropping message {} from {}", message.verb(), message.from()); + dropping.debug(message.verb(), message.from()); return; } @@ -61,12 +64,12 @@ public class AccordVerbHandler implements IVerbHandler long waitForEpoch = request.waitForEpoch(); if (node.topology().hasAtLeastEpoch(waitForEpoch)) - request.process(node, fromNodeId, message); + request.process(node, fromNodeId, message.header); else node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> { if (withEpochFailure != null) throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure); - request.process(node, fromNodeId, message); + request.process(node, fromNodeId, message.header); }); } } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java index 45f69e67f1..f4162d02b8 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java @@ -18,7 +18,6 @@ package org.apache.cassandra.service.accord; -import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Map; @@ -60,6 +59,7 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener historicalTransaction = new TreeMap<>(); private final AccordCommandStore store; private final ObjectHashSet cachedRangeTxns = new ObjectHashSet<>(); + // TODO (required): make this configurable, or perhaps backed by READ stage with concurrency limit public CommandsForRangesLoader(AccordCommandStore store) { @@ -91,7 +91,7 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener before = ImmutableMap.copyOf(watcher.get()); - return AsyncChains.ofCallable(Stage.READ.executor(), () -> get(ranges, before, findAsDep, minTxnId, maxTxnId, redundantBefore)) + return AsyncChains.ofCallable(Stage.ACCORD_RANGE_LOADER.executor(), () -> get(ranges, before, findAsDep, minTxnId, maxTxnId, redundantBefore)) .map(map -> Pair.create(watcher, map), store) .beginAsResult(); } @@ -229,13 +229,27 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener { + Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> { if (e.gcBefore.compareTo(cmd.txnId()) < 0) return accum; - accum.add(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end)); - return accum; - }, new ArrayList(), ignore -> false).toArray(Range[]::new)); - Ranges newRanges = ranges.without(durableAlready); + return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end))); + }, ranges, ignore -> false); if (newRanges.isEmpty()) return null; @@ -279,6 +291,43 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener { + if (e.gcBefore.compareTo(cmd.txnId) < 0) + return accum; + return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end))); + }, ranges, ignore -> false); + + if (newRanges.isEmpty()) + return null; + } + + return new Summary(cmd.txnId, cmd.executeAt, saveStatus, ranges, null, false); + } + public void mergeHistoricalTransaction(TxnId txnId, Ranges ranges, BiFunction remappingFunction) { historicalTransaction.merge(txnId, ranges, remappingFunction); diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java index 7de9d68602..635e5e2d15 100644 --- a/src/java/org/apache/cassandra/service/accord/IJournal.java +++ b/src/java/org/apache/cassandra/service/accord/IJournal.java @@ -26,20 +26,23 @@ import accord.local.CommandStores; import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.primitives.Deps; +import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Timestamp; import accord.primitives.TxnId; import accord.utils.PersistentField.Persister; +import org.apache.cassandra.utils.Pair; public interface IJournal { Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore); + SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId, SavedCommand.Load load, RedundantBefore redundantBefore, DurableBefore durableBefore); RedundantBefore loadRedundantBefore(int commandStoreId); NavigableMap loadBootstrapBeganAt(int commandStoreId); NavigableMap loadSafeToRead(int commandStoreId); CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int commandStoreId); - List loadHistoricalTransactions(int store); + List> loadHistoricalTransactions(long epoch, int store); void appendCommand(int store, SavedCommand.DiffWriter value, Runnable onFlush); Persister durableBeforePersister(); diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java index 99e068ca26..8266b3ac64 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -23,8 +23,11 @@ import java.nio.ByteBuffer; import java.util.Objects; import java.util.zip.Checksum; +import accord.local.Node; import accord.local.Node.Id; +import accord.primitives.Routable; import accord.primitives.Timestamp; +import accord.primitives.Txn; import accord.primitives.TxnId; import accord.utils.Invariants; import org.apache.cassandra.io.util.DataInputPlus; @@ -277,5 +280,9 @@ public final class JournalKey } } - + public static JournalKey keyForHistoricalTransactions(long epoch, int store) + { + TxnId txnId = new TxnId(epoch, 0l, Txn.Kind.LocalOnly, Routable.Domain.Range, Node.Id.NONE); + return new JournalKey(txnId, JournalKey.Type.HISTORICAL_TRANSACTIONS, store); + } } diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java index be2f9c6289..0bd21ef056 100644 --- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java +++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java @@ -59,7 +59,12 @@ import static accord.primitives.Known.KnownDeps.NoDeps; import static accord.primitives.SaveStatus.TruncatedApplyWithOutcome; import static accord.primitives.Status.Durability.NotDurable; import static accord.utils.Invariants.illegalState; +import static org.apache.cassandra.service.accord.SavedCommand.Fields.DURABILITY; +import static org.apache.cassandra.service.accord.SavedCommand.Fields.EXECUTE_AT; import static org.apache.cassandra.service.accord.SavedCommand.Fields.PARTICIPANTS; +import static org.apache.cassandra.service.accord.SavedCommand.Fields.SAVE_STATUS; +import static org.apache.cassandra.service.accord.SavedCommand.Fields.WRITES; +import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL; public class SavedCommand { @@ -231,8 +236,8 @@ public class SavedCommand 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); - flags = collectFlags(before, after, Command::durability, false, Fields.DURABILITY, flags); + flags = collectFlags(before, after, Command::saveStatus, false, SAVE_STATUS, flags); + flags = collectFlags(before, after, Command::durability, false, DURABILITY, flags); flags = collectFlags(before, after, Command::acceptedOrCommitted, false, Fields.ACCEPTED, flags); flags = collectFlags(before, after, Command::promised, false, Fields.PROMISED, flags); @@ -243,7 +248,7 @@ public class SavedCommand flags = collectFlags(before, after, SavedCommand::getWaitingOn, false, Fields.WAITING_ON, flags); - flags = collectFlags(before, after, Command::writes, false, Fields.WRITES, flags); + flags = collectFlags(before, after, Command::writes, false, WRITES, flags); return flags; } @@ -323,8 +328,51 @@ public class SavedCommand return oldFlags & ~(1 << field.ordinal()); } + public enum Load + { + ALL(0), + PURGEABLE(SAVE_STATUS, PARTICIPANTS, DURABILITY, EXECUTE_AT, WRITES), + MINIMAL(SAVE_STATUS, PARTICIPANTS, EXECUTE_AT); + + final int mask; + + Load(int mask) + { + this.mask = mask; + } + + Load(Fields ... fields) + { + int mask = -1; + for (Fields field : fields) + mask &= ~(1<< field.ordinal()); + this.mask = mask; + } + } + + public static class MinimalCommand + { + public final TxnId txnId; + public final SaveStatus saveStatus; + public final StoreParticipants participants; + public final Status.Durability durability; + public final Timestamp executeAt; + public final Writes writes; + + public MinimalCommand(TxnId txnId, SaveStatus saveStatus, StoreParticipants participants, Status.Durability durability, Timestamp executeAt, Writes writes) + { + this.txnId = txnId; + this.saveStatus = saveStatus; + this.participants = participants; + this.durability = durability; + this.executeAt = executeAt; + this.writes = writes; + } + } + public static class Builder { + final int mask; int flags; TxnId txnId; @@ -350,13 +398,25 @@ public class SavedCommand boolean nextCalled; int count; + public Builder(TxnId txnId, Load load) + { + this.mask = load.mask; + init(txnId); + } + public Builder(TxnId txnId) { - init(txnId); + this(txnId, ALL); + } + + public Builder(Load load) + { + this.mask = load.mask; } public Builder() { + this(ALL); } public TxnId txnId() @@ -484,7 +544,7 @@ public class SavedCommand if (saveStatus == null || participants == null) return Cleanup.NO; - Cleanup cleanup = Cleanup.shouldCleanup(txnId, saveStatus, durability, participants, redundantBefore, durableBefore); + Cleanup cleanup = Cleanup.shouldCleanupPartial(txnId, saveStatus, durability, participants, redundantBefore, durableBefore); if (this.cleanup != null && this.cleanup.compareTo(cleanup) > 0) cleanup = this.cleanup; return cleanup; @@ -522,13 +582,13 @@ public class SavedCommand public Builder expungePartial(Cleanup cleanup, SaveStatus saveStatus, boolean includeOutcome) { Invariants.checkState(txnId != null); - Builder builder = new Builder(txnId); + Builder builder = new Builder(txnId, ALL); builder.count++; builder.nextCalled = true; Invariants.checkState(saveStatus != null); - builder.flags = setFieldChanged(Fields.SAVE_STATUS, builder.flags); + builder.flags = setFieldChanged(SAVE_STATUS, builder.flags); builder.saveStatus = saveStatus; builder.flags = setFieldChanged(Fields.CLEANUP, builder.flags); builder.cleanup = cleanup; @@ -539,7 +599,7 @@ public class SavedCommand } if (durability != null) { - builder.flags = setFieldChanged(Fields.DURABILITY, builder.flags); + builder.flags = setFieldChanged(DURABILITY, builder.flags); builder.durability = durability; } if (participants != null) @@ -549,7 +609,7 @@ public class SavedCommand } if (includeOutcome && builder.writes != null) { - builder.flags = setFieldChanged(Fields.WRITES, builder.flags); + builder.flags = setFieldChanged(WRITES, builder.flags); builder.writes = writes; } @@ -559,7 +619,7 @@ public class SavedCommand public Builder saveStatusOnly() { Invariants.checkState(txnId != null); - Builder builder = new Builder(txnId); + Builder builder = new Builder(txnId, ALL); builder.count++; builder.nextCalled = true; @@ -567,7 +627,7 @@ public class SavedCommand // TODO: these accesses can be abstracted away if (saveStatus != null) { - builder.flags = setFieldChanged(Fields.SAVE_STATUS, builder.flags); + builder.flags = setFieldChanged(SAVE_STATUS, builder.flags); builder.saveStatus = saveStatus; } @@ -583,6 +643,11 @@ public class SavedCommand } } + public MinimalCommand asMinimal() + { + return new MinimalCommand(txnId, saveStatus, participants, durability, executeAt, writes); + } + public static Route deserializeRouteOrNull(DataInputPlus in, int userVersion) throws IOException { int flags = in.readInt(); @@ -595,6 +660,7 @@ public class SavedCommand public void serialize(DataOutputPlus out, int userVersion) throws IOException { + Invariants.checkState(mask == 0); out.writeInt(validateFlags(flags)); int iterable = toIterableSetFields(flags); @@ -665,7 +731,7 @@ public class SavedCommand while (iterable != 0) { Fields field = nextSetField(iterable); - if (getFieldChanged(field, this.flags)) + if (getFieldChanged(field, this.flags) || getFieldIsNull(field, mask)) { if (!getFieldIsNull(field, flags)) skip(field, in, userVersion); diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java index 6d8d2b8184..68ab848885 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java @@ -287,14 +287,14 @@ public abstract class AccordRoutingKey extends AccordRoutableKey implements Rout @Override public TokenKey deserialize(DataInputPlus in, int version) throws IOException { - TableId table = TableId.deserialize(in); + TableId table = TableId.deserialize(in).intern(); Token token = Token.compactSerializer.deserialize(in, getPartitioner(), version); return new TokenKey(table, token); } public TokenKey fromBytes(ByteBuffer bytes, IPartitioner partitioner) { - TableId tableId = TableId.deserialize(bytes, ByteBufferAccessor.instance, 0); + TableId tableId = TableId.deserialize(bytes, ByteBufferAccessor.instance, 0).intern(); bytes.position(tableId.serializedSize()); Token token = Token.compactSerializer.deserialize(bytes, partitioner); return new TokenKey(tableId, token); diff --git a/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java b/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java index fc78fe6692..a01ae38284 100644 --- a/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java @@ -146,15 +146,14 @@ public final class PartitionKey extends AccordRoutableKey implements Key @Override public void skip(DataInputPlus in, int version) throws IOException { - TableId tableId = TableId.deserialize(in); - IPartitioner partitioner = Schema.instance.getExistingTablePartitioner(tableId); + in.skipBytesFully(TableId.staticSerializedSize()); ByteBufferUtil.skipShortLength(in); } @Override public PartitionKey deserialize(DataInputPlus in, int version) throws IOException { - TableId tableId = TableId.deserialize(in); + TableId tableId = TableId.deserialize(in).intern(); IPartitioner partitioner = Schema.instance.getExistingTablePartitioner(tableId); DecoratedKey key = partitioner.decorateKey(ByteBufferUtil.readWithShortLength(in)); return new PartitionKey(tableId, key); @@ -162,7 +161,7 @@ public final class PartitionKey extends AccordRoutableKey implements Key public PartitionKey deserialize(V src, ValueAccessor accessor, int offset) throws IOException { - TableId tableId = TableId.deserialize(src, accessor, offset); + TableId tableId = TableId.deserialize(src, accessor, offset).intern(); offset += tableId.serializedSize(); TableMetadata metadata = Schema.instance.getTableMetadata(tableId); int numBytes = accessor.getShort(src, offset); diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java index 71758981a6..7f6b30fb7c 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java @@ -17,34 +17,53 @@ */ package org.apache.cassandra.service.accord.async; -import accord.api.RoutingKey; -import accord.local.cfk.CommandsForKey; -import accord.local.KeyHistory; -import accord.local.PreLoadContext; -import accord.primitives.*; -import accord.utils.async.AsyncChain; -import accord.utils.async.AsyncChains; -import accord.utils.async.AsyncResult; -import accord.utils.async.Observable; -import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.ImmutableList; -import com.google.common.collect.ImmutableSet; -import com.google.common.collect.Iterables; -import org.apache.cassandra.service.accord.*; -import org.apache.cassandra.service.accord.api.AccordRoutingKey; -import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey; -import org.apache.cassandra.utils.NoSpamLogger; -import org.apache.cassandra.utils.Pair; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.stream.Collectors; import javax.annotation.Nullable; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Iterables; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.api.RoutingKey; +import accord.local.KeyHistory; +import accord.local.PreLoadContext; +import accord.local.cfk.CommandsForKey; +import accord.primitives.AbstractKeys; +import accord.primitives.AbstractRanges; +import accord.primitives.Range; +import accord.primitives.Ranges; +import accord.primitives.TxnId; +import accord.primitives.Unseekables; +import accord.utils.async.AsyncChain; +import accord.utils.async.AsyncChains; +import accord.utils.async.AsyncResult; +import accord.utils.async.Observable; +import org.apache.cassandra.concurrent.ExecutorPlus; +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.service.accord.AccordCachingState; +import org.apache.cassandra.service.accord.AccordCommandStore; +import org.apache.cassandra.service.accord.AccordKeyspace; +import org.apache.cassandra.service.accord.AccordSafeCommandsForRanges; +import org.apache.cassandra.service.accord.AccordSafeState; +import org.apache.cassandra.service.accord.AccordStateCache; +import org.apache.cassandra.service.accord.CommandsForRangesLoader; +import org.apache.cassandra.service.accord.api.AccordRoutingKey; +import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey; +import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.Pair; + public class AsyncLoader { private static final Logger logger = LoggerFactory.getLogger(AsyncLoader.class); @@ -89,7 +108,16 @@ public class AsyncLoader AccordStateCache.Instance cache, List> listenChains) { - S safeRef = cache.acquire(key); + referenceAndAssembleReadsForKey(key, context, cache, listenChains, null); + } + + private static > void referenceAndAssembleReadsForKey(K key, + Map context, + AccordStateCache.Instance cache, + List> listenChains, + @Nullable ExecutorPlus loadExecutor) + { + S safeRef = cache.acquire(key, loadExecutor); if (context.putIfAbsent(key, safeRef) != null) { noSpamLogger.warn("Context {} contained key {} more than once", context, key); @@ -120,16 +148,24 @@ public class AsyncLoader private void referenceAndAssembleReadsForKey(RoutingKey key, AsyncOperation.Context context, List> listenChains) + { + referenceAndAssembleReadsForKey(key, context, listenChains, null); + } + + private void referenceAndAssembleReadsForKey(RoutingKey key, + AsyncOperation.Context context, + List> listenChains, + @Nullable ExecutorPlus loadExecutor) { // recovery operations also need the deps data for their preaccept logic switch (keyHistory) { case TIMESTAMPS: - referenceAndAssembleReadsForKey(key, context.timestampsForKey, commandStore.timestampsForKeyCache(), listenChains); + referenceAndAssembleReadsForKey(key, context.timestampsForKey, commandStore.timestampsForKeyCache(), listenChains, loadExecutor); break; case COMMANDS: case RECOVERY: - referenceAndAssembleReadsForKey(key, context.commandsForKey, commandStore.commandsForKeyCache(), listenChains); + referenceAndAssembleReadsForKey(key, context.commandsForKey, commandStore.commandsForKeyCache(), listenChains, loadExecutor); case NONE: break; default: throw new IllegalArgumentException("Unhandled keyhistory: " + keyHistory); @@ -168,11 +204,15 @@ public class AsyncLoader private AsyncChain referenceAndDispatchReadsForRange(@Nullable TxnId primaryTxnId, AsyncOperation.Context context) { + if (keyHistory == KeyHistory.NONE) + return AsyncChains.success(null); + Ranges ranges = ((AbstractRanges) keysOrRanges).toRanges(); List> root = new ArrayList<>(ranges.size() + 1); class Watcher implements AccordStateCache.Listener { + // TODO (required): streams prohibited in hot path private final Set cached = commandStore.commandsForKeyCache().stream() .map(n -> (TokenKey) n.key()) .filter(ranges::contains) @@ -196,7 +236,7 @@ public class AsyncLoader return AsyncChains.success(null); Set set = ImmutableSet.builder().addAll(watcher.cached).addAll(keys).build(); List> chains = new ArrayList<>(); - set.forEach(key -> referenceAndAssembleReadsForKey(key, context, chains)); + set.forEach(key -> referenceAndAssembleReadsForKey(key, context, chains, Stage.ACCORD_RANGE_LOADER.executor())); return chains.isEmpty() ? AsyncChains.success(null) : AsyncChains.reduce(chains, (a, b) -> null); }, commandStore)); 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 463350adf8..51dbb0a6c2 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java @@ -37,6 +37,7 @@ import accord.primitives.TxnId; import accord.primitives.Unseekables; import accord.utils.Invariants; import accord.utils.async.AsyncChains; +import accord.utils.async.Cancellable; import org.agrona.collections.Object2ObjectHashMap; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.service.accord.AccordCommandStore; @@ -57,7 +58,7 @@ import static org.apache.cassandra.service.accord.async.AsyncOperation.State.LOA import static org.apache.cassandra.service.accord.async.AsyncOperation.State.PREPARING; import static org.apache.cassandra.service.accord.async.AsyncOperation.State.RUNNING; -public abstract class AsyncOperation extends AsyncChains.Head implements Runnable, Function +public abstract class AsyncOperation extends AsyncChains.Head implements Runnable, Function, Cancellable { private static final Logger logger = LoggerFactory.getLogger(AsyncOperation.class); @@ -360,12 +361,18 @@ public abstract class AsyncOperation extends AsyncChains.Head implements R } @Override - public void start(BiConsumer callback) + public Cancellable start(BiConsumer callback) { Invariants.checkState(this.callback == null); this.callback = callback; if (!commandStore.inStore() || preRun()) commandStore.executor().execute(this); + return this; + } + + @Override + public void cancel() + { } static class ForFunction extends AsyncOperation diff --git a/src/java/org/apache/cassandra/utils/NoSpamLogger.java b/src/java/org/apache/cassandra/utils/NoSpamLogger.java index e16caa930f..94c5e7d103 100644 --- a/src/java/org/apache/cassandra/utils/NoSpamLogger.java +++ b/src/java/org/apache/cassandra/utils/NoSpamLogger.java @@ -46,7 +46,7 @@ public class NoSpamLogger */ public enum Level { - INFO, WARN, ERROR + DEBUG, INFO, WARN, ERROR } @VisibleForTesting @@ -99,6 +99,9 @@ public class NoSpamLogger { switch (l) { + case DEBUG: + wrapped.debug(statement, objects); + break; case INFO: wrapped.info(statement, objects); break; @@ -114,6 +117,16 @@ public class NoSpamLogger return true; } + public boolean debug(long nowNanos, Object... objects) + { + return NoSpamLogStatement.this.log(Level.DEBUG, nowNanos, objects); + } + + public boolean debug(Object... objects) + { + return NoSpamLogStatement.this.debug(CLOCK.nanoTime(), objects); + } + public boolean info(long nowNanos, Object... objects) { return NoSpamLogStatement.this.log(Level.INFO, nowNanos, objects); diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java index e46b0e4d69..90c95228f9 100644 --- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -130,6 +130,7 @@ public class BufferPool public static final int TINY_CHUNK_SIZE = NORMAL_ALLOCATION_UNIT; public static final int TINY_ALLOCATION_UNIT = TINY_CHUNK_SIZE / 64; public static final int TINY_ALLOCATION_LIMIT = TINY_CHUNK_SIZE / 2; + private static final boolean REF_TRACE_ENABLED = Ref.TRACE_ENABLED; private static final Logger logger = LoggerFactory.getLogger(BufferPool.class); private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 15L, TimeUnit.MINUTES); @@ -1330,7 +1331,7 @@ public class BufferPool void setAttachment(ByteBuffer buffer) { - if (Ref.TRACE_ENABLED) + if (REF_TRACE_ENABLED) MemoryUtil.setAttachment(buffer, new DirectBufferRef<>(this, null)); else MemoryUtil.setAttachment(buffer, this); @@ -1342,7 +1343,7 @@ public class BufferPool if (attachment == null) return false; - if (Ref.TRACE_ENABLED) + if (REF_TRACE_ENABLED) ((DirectBufferRef) attachment).release(); return true; 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 4c249a1a28..2dd741f5b2 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java @@ -66,7 +66,8 @@ public class AccordLoadTest extends AccordTestBase CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis()))); // AccordTestBase.setupCluster(builder -> builder, 3); AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config - .set("accord.schedule_durability_frequency", "5s") + .set("accord.shard_durability_target_splits", "64") + .set("accord.shard_durability_cycle", "5m") .set("accord.ephemeral_read_enabled", "true") .set("accord.gc_delay", "5s")), 3); } @@ -105,7 +106,7 @@ public class AccordLoadTest extends AccordTestBase final long batchTime = TimeUnit.SECONDS.toNanos(10); final int concurrency = 100; final int ratePerSecond = 1000; - final int keyCount = 1000000; + final int keyCount = 1_000_000; final float readChance = 0.33f; long nextRepairAt = repairInterval; long nextCompactionAt = compactionInterval; diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java index 3367c72611..578c6a84b9 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java @@ -34,6 +34,7 @@ import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.primitives.Deps; import accord.primitives.KeyDeps; +import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Timestamp; import accord.primitives.TxnId; @@ -52,6 +53,7 @@ import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsAccumulator; import org.apache.cassandra.utils.AccordGenerators; +import org.apache.cassandra.utils.Pair; import static accord.local.CommandStores.RangesForEpoch; import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator; @@ -98,6 +100,7 @@ public class AccordJournalCompactionTest Gen durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner()); Gen> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner()); Gen rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner()); + Gen rangeGen = AccordGenerators.range(DatabaseDescriptor.getPartitioner()); Gen historicalTransactionsGen = depsGen(); AccordJournal journal = new AccordJournal(new TestParams() @@ -134,7 +137,7 @@ public class AccordJournalCompactionTest // updates.newRedundantBefore = redundantBefore = RedundantBefore.merge(redundantBefore, updates.addRedundantBefore); updates.newSafeToRead = safeToReadGen.next(rs); updates.newRangesForEpoch = rangesForEpochGen.next(rs); - updates.addHistoricalTransactions = historicalTransactionsGen.next(rs); + updates.addHistoricalTransactions = new AccordSafeCommandStore.HistoricalTransactions(0l, rangeGen.next(rs), historicalTransactionsGen.next(rs)); journal.durableBeforePersister().persist(addDurableBefore, null); journal.persistStoreState(1, updates, null); @@ -147,7 +150,7 @@ public class AccordJournalCompactionTest safeToReadAtAccumulator = updates.newSafeToRead; if (updates.newRangesForEpoch != null) rangesForEpochAccumulator = updates.newRangesForEpoch; - historicalTransactionsAccumulator.update(updates.addHistoricalTransactions); + historicalTransactionsAccumulator.update(Pair.create(updates.addHistoricalTransactions.range, updates.addHistoricalTransactions.deps)); if (i % 100 == 0) journal.closeCurrentSegmentForTestingIfNonEmpty(); @@ -160,9 +163,9 @@ public class AccordJournalCompactionTest Assert.assertEquals(bootstrapBeganAtAccumulator, journal.loadBootstrapBeganAt(1)); Assert.assertEquals(safeToReadAtAccumulator, journal.loadSafeToRead(1)); Assert.assertEquals(rangesForEpochAccumulator, journal.loadRangesForEpoch(1)); - List historical = historicalTransactionsAccumulator.get(); + List> historical = historicalTransactionsAccumulator.get(); Collections.reverse(historical); - Assert.assertEquals(historical, journal.loadHistoricalTransactions(1)); + Assert.assertEquals(historical, journal.loadHistoricalTransactions(0l, 1)); } finally { diff --git a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java index 0bdc561818..6a4546f435 100644 --- a/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java +++ b/test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java @@ -507,13 +507,15 @@ public class YamlConfigurationLoaderTest public void testAccordConfig() { Map accordSpec = ImmutableMap.of("fast_path_update_delay", "60s", - "schedule_durability_frequency", "60s", + "default_durability_retry_delay", "60s", + "max_durability_retry_delay", "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.default_durability_retry_delay.to(TimeUnit.NANOSECONDS)).isEqualTo(60000000000L); + assertThat(spec.max_durability_retry_delay.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); diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java index 8344c09dc6..b252d13f76 100644 --- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java +++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java @@ -36,6 +36,7 @@ import java.util.Set; import java.util.SortedSet; import java.util.TreeSet; import java.util.concurrent.Callable; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.function.Predicate; @@ -65,6 +66,7 @@ import accord.utils.RandomSource; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; +import accord.utils.async.Cancellable; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; import org.apache.cassandra.concurrent.SimulatedExecutorFactory; import org.apache.cassandra.concurrent.Stage; @@ -629,9 +631,9 @@ public class EpochSyncTest return new AsyncChains.Head<>() { @Override - protected void start(BiConsumer callback) + protected Cancellable start(BiConsumer callback) { - scheduler.schedule(() -> { + Future future = scheduler.schedule(() -> { T value; try { @@ -644,6 +646,7 @@ public class EpochSyncTest } callback.accept(value, null); }, time, unit); + return () -> future.cancel(true); } }; } @@ -672,7 +675,7 @@ public class EpochSyncTest config.registerListener(new ConfigurationService.Listener() { @Override - public AsyncResult onTopologyUpdate(Topology topology, boolean startSync) + public AsyncResult onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync) { // EpochReady ready = EpochReady.done(topology.epoch()); AsyncResult metadata = schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable) () -> null).beginAsResult(); diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java index 64caa97216..ef0e264def 100644 --- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java +++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java @@ -36,6 +36,7 @@ import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.local.StoreParticipants; import accord.primitives.Known; +import accord.primitives.Range; import accord.primitives.SaveStatus; import accord.primitives.Status; import accord.primitives.Ballot; @@ -54,7 +55,10 @@ import org.apache.cassandra.service.accord.AccordJournalValueSerializers.Durable import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsAccumulator; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.RedundantBeforeAccumulator; +import org.apache.cassandra.service.accord.SavedCommand.Load; +import org.apache.cassandra.service.accord.SavedCommand.MinimalCommand; import org.apache.cassandra.service.accord.serializers.CommandSerializers; +import org.apache.cassandra.utils.Pair; public class MockJournal implements IJournal { @@ -81,6 +85,17 @@ public class MockJournal implements IJournal return reconstructFromDiff(new ArrayList<>(saved)); } + @Override + public MinimalCommand loadMinimal(int store, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore) + { + JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, store); + List saved = commands.get(key); + if (saved == null) + return null; + Command command = reconstructFromDiff(new ArrayList<>(saved)); + return new MinimalCommand(command.txnId(), command.saveStatus(), command.participants(), command.durability(), command.executeAt(), command.writes()); + } + @Override public RedundantBefore loadRedundantBefore(int store) { @@ -126,7 +141,7 @@ public class MockJournal implements IJournal } @Override - public List loadHistoricalTransactions(int store) + public List> loadHistoricalTransactions(long epoch, int store) { return fieldUpdates(store).historicalTransactionsAccumulator.get(); } @@ -163,7 +178,7 @@ public class MockJournal implements IJournal if (fieldUpdates.newRangesForEpoch != null) updates.rangesForEpochAccumulator.update(fieldUpdates.newRangesForEpoch); if (fieldUpdates.addHistoricalTransactions != null) - updates.historicalTransactionsAccumulator.update(fieldUpdates.addHistoricalTransactions); + updates.historicalTransactionsAccumulator.update(Pair.create(fieldUpdates.addHistoricalTransactions.range, fieldUpdates.addHistoricalTransactions.deps)); onFlush.run(); } diff --git a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java index 99963c3af0..3cfc1c91c5 100644 --- a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java @@ -40,12 +40,14 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.SavedCommand.Fields; +import org.apache.cassandra.service.accord.SavedCommand.Load; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.utils.AccordGenerators; import org.assertj.core.api.SoftAssertions; import static accord.utils.Property.qt; import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse; +import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL; import static org.apache.cassandra.service.accord.SavedCommand.getFlags; public class SavedCommandTest @@ -96,7 +98,7 @@ public class SavedCommandTest out.clear(); Command orig = cmdBuilder.build(saveStatus); SavedCommand.serialize(null, orig, out, userVersion); - SavedCommand.Builder builder = new SavedCommand.Builder(orig.txnId()); + SavedCommand.Builder builder = new SavedCommand.Builder(orig.txnId(), Load.ALL); builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), userVersion); // We are not persisting the result, so force it for strict equality builder.forceResult(orig.result());