From 5a12875b3a7a5c7c6b152a8fc19e1b2b2b6c46e8 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Fri, 11 Oct 2024 23:09:22 +0100 Subject: [PATCH] Use ExclusiveSyncPoints to join a new topology For correctness, the dependencies we adopt on joining a new topology must exclude the possibility of respondents accepting additional transactions with a lower TxnId, so proxying on the existing `ExclusiveSyncPoint` mechanisms is logical for the time-being. This patch removes the `FetchMajorityDeps` logic in favour of simply waiting for a suitable `ExclusiveSyncPoint` to be proposed. patch by Benedict, reviewed by Alex Petrov for CASSANDRA-20056 --- modules/accord | 2 +- .../service/accord/AccordCommandStore.java | 95 +++++-------------- .../service/accord/AccordJournal.java | 14 --- .../accord/AccordJournalValueSerializers.java | 60 ------------ .../accord/AccordSafeCommandStore.java | 26 ----- .../service/accord/AccordService.java | 17 ++-- .../accord/CommandsForRangesLoader.java | 27 ++++-- .../cassandra/service/accord/IJournal.java | 5 - .../cassandra/service/accord/JournalKey.java | 11 --- .../service/accord/api/AccordScheduler.java | 7 ++ .../accord/AccordJournalCompactionTest.java | 10 -- .../cassandra/service/accord/MockJournal.java | 13 --- .../service/accord/SavedCommandTest.java | 1 - 13 files changed, 60 insertions(+), 228 deletions(-) diff --git a/modules/accord b/modules/accord index a63cac24a2..d77bdd1a4c 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit a63cac24a2198a5893874cdf72946073854a8d4d +Subproject commit d77bdd1a4cd96120868279b665e0abe4ab509a80 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index c46c953a61..c7ee8aa6cd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -28,7 +28,6 @@ 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; @@ -39,7 +38,6 @@ 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; @@ -51,16 +49,14 @@ 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; import accord.local.SafeCommand; import accord.local.SafeCommandStore; import accord.local.cfk.CommandsForKey; -import accord.primitives.Deps; import accord.primitives.Participants; -import accord.primitives.Range; +import accord.primitives.RangeDeps; import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.RoutableKey; @@ -70,8 +66,6 @@ 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; @@ -82,12 +76,10 @@ 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.SaveStatus.Applying; import static accord.primitives.Status.Committed; @@ -117,7 +109,7 @@ public class AccordCommandStore extends CommandStore { if (!DatabaseDescriptor.getAccordStateCacheListenerJFREnabled()) return; - instance.register(new AccordStateCache.Listener() { + instance.register(new AccordStateCache.Listener<>() { private final IdentityHashMap, CacheEvents.Evict> pendingEvicts = new IdentityHashMap<>(); @Override @@ -249,6 +241,7 @@ public class AccordCommandStore extends CommandStore { store.snapshot(ranges, globalSyncId); super.markShardDurable(safeStore, globalSyncId, ranges); + commandsForRangesLoader.gcBefore(globalSyncId, ranges); } @Override @@ -506,75 +499,37 @@ public class AccordCommandStore extends CommandStore { } - protected ConfigurationService.EpochReady syncInternal(Node node, Ranges ranges, long epoch, boolean isLoad) + public void registerTransitive(SafeCommandStore safeStore, RangeDeps rangeDeps) { - 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; + if (rangeDeps.isEmpty()) + return; + RedundantBefore redundantBefore = unsafeGetRedundantBefore(); CommandStores.RangesForEpoch ranges = safeStore.ranges(); // used in places such as accord.local.CommandStore.fetchMajorityDeps // We find a set of dependencies for a range then update CommandsFor to know about them Ranges allRanges = safeStore.ranges().all(); - deps.keyDeps.keys().forEach(allRanges, key -> { - // TODO (desired): batch register to minimise GC - deps.keyDeps.forEach(key, (txnId, txnIdx) -> { - if (ranges.coordinates(txnId).contains(key)) - return; // already coordinates, no need to replicate - if (!ranges.allBefore(txnId.epoch()).contains(key)) - return; - - safeStore.get(key).registerHistorical(safeStore, txnId); - }); - }); - for (int i = 0; i < deps.rangeDeps.rangeCount(); i++) + Ranges coordinateRanges = Ranges.EMPTY; + long coordinateEpoch = -1; + for (int i = 0; i < rangeDeps.txnIdCount(); i++) { - Range r = deps.rangeDeps.range(i); - if (!allRanges.intersects(r)) + TxnId txnId = rangeDeps.txnId(i); + AccordCachingState state = commandCache.getUnsafe(txnId); + if (state != null && state.isLoaded() && state.get() != null && state.get().known().isDefinitionKnown()) continue; - deps.rangeDeps.forEach(r, txnId -> { - // TODO (desired, efficiency): this can be made more efficient by batching by epoch - if (ranges.coordinates(txnId).intersects(r)) - return; // already coordinates, no need to replicate - 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(r).slice(allRanges), Ranges::with); - }); + Ranges addRanges = rangeDeps.ranges(i).slice(allRanges); + if (addRanges.isEmpty()) continue; + + if (coordinateEpoch != txnId.epoch()) + { + coordinateEpoch = txnId.epoch(); + coordinateRanges = ranges.allAt(txnId.epoch()); + } + if (addRanges.intersects(coordinateRanges)) continue; + addRanges = redundantBefore.removeShardRedundant(txnId, txnId, addRanges); + if (addRanges.isEmpty()) continue; + diskCommandsForRanges().mergeTransitive(txnId, addRanges, Ranges::with); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index a575da3f8b..aa79a492a9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -40,8 +40,6 @@ import accord.local.DurableBefore; 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; @@ -62,17 +60,14 @@ import org.apache.cassandra.journal.Params; import org.apache.cassandra.journal.RecordPointer; 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 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 { @@ -239,13 +234,6 @@ public class AccordJournal implements IJournal, Shutdownable return accumulator.get(); } - @Override - public List> loadHistoricalTransactions(long epoch, int store) - { - HistoricalTransactionsAccumulator accumulator = readAll(keyForHistoricalTransactions(epoch, store)); - return accumulator.get(); - } - @Override public void appendCommand(int store, SavedCommand.DiffWriter value, Runnable onFlush) { @@ -304,8 +292,6 @@ public class AccordJournal implements IJournal, Shutdownable pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.SAFE_TO_READ, store), fieldUpdates.newSafeToRead); if (fieldUpdates.newRangesForEpoch != null) pointer = appendInternal(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store), fieldUpdates.newRangesForEpoch); - if (fieldUpdates.addHistoricalTransactions != null) - pointer = appendInternal(JournalKey.keyForHistoricalTransactions(fieldUpdates.addHistoricalTransactions.epoch, store), Pair.create(fieldUpdates.addHistoricalTransactions.range, fieldUpdates.addHistoricalTransactions.deps)); if (onFlush == null) return; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java index f232b9213e..c6a2a46bf4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java @@ -19,16 +19,12 @@ package org.apache.cassandra.service.accord; import java.io.IOException; -import java.util.ArrayList; -import java.util.List; import java.util.NavigableMap; 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; @@ -37,11 +33,9 @@ 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 // TODO (required): alert on metadata size @@ -340,58 +334,4 @@ public class AccordJournalValueSerializers into.update(new RangesForEpoch.Snapshot(epochs, ranges)); } } - - public static class HistoricalTransactionsAccumulator extends Accumulator>, Pair> - { - public HistoricalTransactionsAccumulator() - { - super(new ArrayList<>()); - } - - @Override - protected List> accumulate(List> oldValue, Pair deps) - { - accumulated.add(deps); // we can keep it mutable - return accumulated; - } - } - - public static class HistoricalTransactionsSerializer implements FlyweightSerializer, HistoricalTransactionsAccumulator> - { - @Override - public HistoricalTransactionsAccumulator mergerFor(JournalKey key) - { - return new HistoricalTransactionsAccumulator(); - } - - @Override - public void serialize(JournalKey key, Pair from, DataOutputPlus out, int userVersion) throws IOException - { - out.writeUnsignedVInt32(1); - 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 (Pair d : from.get()) - { - TokenRange.serializer.serialize((TokenRange) d.left, out, messagingVersion); - deps.serialize(d.right, out, messagingVersion); - } - } - - @Override - public void deserialize(JournalKey key, HistoricalTransactionsAccumulator into, DataInputPlus in, int userVersion) throws IOException - { - int count = in.readUnsignedVInt32(); - for (int i = 0; i < count; i++) - { - 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/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index c260a013c6..eef12d6a7e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -41,8 +41,6 @@ import accord.local.RedundantBefore; 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,15 +334,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore newBootstrapBeganAt; public NavigableMap newSafeToRead; public RangesForEpoch.Snapshot newRangesForEpoch; - 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 167428a759..36710717d0 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -66,7 +66,6 @@ import accord.coordinate.TopologyMismatch; import accord.coordinate.tracking.AllTracker; import accord.coordinate.tracking.RequestStatus; import accord.impl.AbstractConfigurationService; -import accord.impl.CoordinateDurabilityScheduling; import accord.impl.DefaultLocalListeners; import accord.impl.DefaultRemoteListeners; import accord.impl.RequestCallbacks; @@ -202,7 +201,6 @@ public class AccordService implements IAccordService, Shutdownable private final AccordScheduler scheduler; private final AccordDataStore dataStore; private final AccordJournal journal; - private final CoordinateDurabilityScheduling durabilityScheduling; private final AccordVerbHandler requestHandler; private final AccordResponseVerbHandler responseHandler; private final LocalConfig configuration; @@ -448,7 +446,6 @@ public class AccordService implements IAccordService, Shutdownable journal.durableBeforePersister(), configuration); this.nodeShutdown = toShutdownable(node); - this.durabilityScheduling = new CoordinateDurabilityScheduling(node); this.requestHandler = new AccordVerbHandler<>(node, configService); this.responseHandler = new AccordResponseVerbHandler<>(callbacks, configService); } @@ -504,13 +501,13 @@ 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.start(); + node.durabilityScheduling().setDefaultRetryDelay(Ints.checkedCast(DatabaseDescriptor.getAccordDefaultDurabilityRetryDelay(SECONDS)), SECONDS); + node.durabilityScheduling().setMaxRetryDelay(Ints.checkedCast(DatabaseDescriptor.getAccordMaxDurabilityRetryDelay(SECONDS)), SECONDS); + node.durabilityScheduling().setTargetShardSplits(Ints.checkedCast(DatabaseDescriptor.getAccordShardDurabilityTargetSplits())); + node.durabilityScheduling().setGlobalCycleTime(Ints.checkedCast(DatabaseDescriptor.getAccordGlobalDurabilityCycle(SECONDS)), SECONDS); + node.durabilityScheduling().setShardCycleTime(Ints.checkedCast(DatabaseDescriptor.getAccordShardDurabilityCycle(SECONDS)), SECONDS); + node.durabilityScheduling().setTxnIdLag(Ints.checkedCast(DatabaseDescriptor.getAccordScheduleDurabilityTxnIdLag(SECONDS)), TimeUnit.SECONDS); + node.durabilityScheduling().start(); state = State.STARTED; } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java index f4162d02b8..c338e1116b 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java @@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord; import java.util.Collection; import java.util.Collections; +import java.util.Iterator; import java.util.Map; import java.util.NavigableMap; import java.util.Set; @@ -55,9 +56,8 @@ import static accord.primitives.Txn.Kind.ExclusiveSyncPoint; public class CommandsForRangesLoader implements AccordStateCache.Listener { private final RoutesSearcher searcher = new RoutesSearcher(); - //TODO (now, durability): find solution for this... - private final NavigableMap historicalTransaction = new TreeMap<>(); private final AccordCommandStore store; + private final NavigableMap transitive = new TreeMap<>(); private final ObjectHashSet cachedRangeTxns = new ObjectHashSet<>(); // TODO (required): make this configurable, or perhaps backed by READ stage with concurrency limit @@ -86,7 +86,7 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener>> get(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Ranges ranges) { RedundantBefore redundantBefore = store.unsafeGetRedundantBefore(); - TxnId minTxnId = redundantBefore.minGcBefore(ranges); + TxnId minTxnId = redundantBefore.min(ranges, e -> e.gcBefore); Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVERY || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId; TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVERY ? primaryTxnId : null; Watcher watcher = fromCache(findAsDep, ranges, minTxnId, maxTxnId, redundantBefore); @@ -110,11 +110,11 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener intersects = searcher.intersects(store.id(), (TokenRange) range, minTxnId, maxTxnId); - if (!historicalTransaction.isEmpty()) + if (!transitive.isEmpty()) { if (intersects.isEmpty()) intersects = new ObjectHashSet<>(); - for (Map.Entry e : historicalTransaction.tailMap(minTxnId, true).entrySet()) + for (Map.Entry e : transitive.tailMap(minTxnId, true).entrySet()) { if (e.getValue().intersects(range)) intersects.add(e.getKey()); @@ -328,9 +328,22 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener remappingFunction) + public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction remappingFunction) { - historicalTransaction.merge(txnId, ranges, remappingFunction); + transitive.merge(txnId, ranges, remappingFunction); + } + + public void gcBefore(TxnId gcBefore, Ranges ranges) + { + Iterator> iterator = transitive.headMap(gcBefore).entrySet().iterator(); + while (iterator.hasNext()) + { + Map.Entry e = iterator.next(); + Ranges newRanges = e.getValue().without(ranges); + if (newRanges.isEmpty()) + iterator.remove(); + e.setValue(newRanges); + } } public static class Summary diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java index 635e5e2d15..849173c9b5 100644 --- a/src/java/org/apache/cassandra/service/accord/IJournal.java +++ b/src/java/org/apache/cassandra/service/accord/IJournal.java @@ -18,20 +18,16 @@ package org.apache.cassandra.service.accord; -import java.util.List; import java.util.NavigableMap; import accord.local.Command; 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 { @@ -42,7 +38,6 @@ public interface IJournal NavigableMap loadBootstrapBeganAt(int commandStoreId); NavigableMap loadSafeToRead(int commandStoreId); CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int commandStoreId); - 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 8266b3ac64..4292387ee7 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -23,11 +23,8 @@ 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; @@ -37,7 +34,6 @@ import org.apache.cassandra.service.accord.AccordJournalValueSerializers.Bootstr import org.apache.cassandra.service.accord.AccordJournalValueSerializers.CommandDiffSerializer; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeSerializer; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer; -import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsSerializer; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.RedundantBeforeSerializer; import org.apache.cassandra.utils.ByteArrayUtil; @@ -237,7 +233,6 @@ public final class JournalKey SAFE_TO_READ (3, new SafeToReadSerializer()), BOOTSTRAP_BEGAN_AT (4, new BootstrapBeganAtSerializer()), RANGES_FOR_EPOCH (5, new RangesForEpochSerializer()), - HISTORICAL_TRANSACTIONS (6, new HistoricalTransactionsSerializer()) ; public final int id; @@ -279,10 +274,4 @@ public final class JournalKey return type; } } - - 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/api/AccordScheduler.java b/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java index 50720952a4..dec0cbb225 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java @@ -68,6 +68,13 @@ public class AccordScheduler implements Scheduler, Shutdownable return new ScheduledFutureWrapper(future); } + @Override + public Scheduled selfRecurring(Runnable run, long delay, TimeUnit units) + { + ScheduledFuture future = scheduledExecutor.scheduleSelfRecurring(run, delay, units); + return new ScheduledFutureWrapper(future); + } + @Override public void now(Runnable task) { diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java index 578c6a84b9..ac17ddad15 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java @@ -19,8 +19,6 @@ package org.apache.cassandra.service.accord; import java.nio.file.Files; -import java.util.Collections; -import java.util.List; import java.util.NavigableMap; import java.util.concurrent.atomic.AtomicInteger; @@ -51,9 +49,7 @@ import org.apache.cassandra.io.util.File; import org.apache.cassandra.journal.TestParams; 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; @@ -94,7 +90,6 @@ public class AccordJournalCompactionTest NavigableMap safeToReadAtAccumulator = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY); NavigableMap bootstrapBeganAtAccumulator = ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY); RangesForEpoch.Snapshot rangesForEpochAccumulator = null; - HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator(); Gen redundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner()); Gen durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner()); @@ -137,7 +132,6 @@ public class AccordJournalCompactionTest // updates.newRedundantBefore = redundantBefore = RedundantBefore.merge(redundantBefore, updates.addRedundantBefore); updates.newSafeToRead = safeToReadGen.next(rs); updates.newRangesForEpoch = rangesForEpochGen.next(rs); - updates.addHistoricalTransactions = new AccordSafeCommandStore.HistoricalTransactions(0l, rangeGen.next(rs), historicalTransactionsGen.next(rs)); journal.durableBeforePersister().persist(addDurableBefore, null); journal.persistStoreState(1, updates, null); @@ -150,7 +144,6 @@ public class AccordJournalCompactionTest safeToReadAtAccumulator = updates.newSafeToRead; if (updates.newRangesForEpoch != null) rangesForEpochAccumulator = updates.newRangesForEpoch; - historicalTransactionsAccumulator.update(Pair.create(updates.addHistoricalTransactions.range, updates.addHistoricalTransactions.deps)); if (i % 100 == 0) journal.closeCurrentSegmentForTestingIfNonEmpty(); @@ -163,9 +156,6 @@ 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(); - Collections.reverse(historical); - Assert.assertEquals(historical, journal.loadHistoricalTransactions(0l, 1)); } finally { diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java index b7b2fecb57..7d7c49153e 100644 --- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java +++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java @@ -36,11 +36,9 @@ 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; -import accord.primitives.Deps; import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.Ranges; @@ -52,13 +50,11 @@ import accord.utils.PersistentField.Persister; import accord.utils.async.AsyncResult; import accord.utils.async.AsyncResults; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator; -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 { @@ -70,7 +66,6 @@ public class MockJournal implements IJournal final IdentityAccumulator> bootstrapBeganAtAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY)); final IdentityAccumulator> safeToReadAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY)); final IdentityAccumulator rangesForEpochAccumulator = new IdentityAccumulator<>(null); - final HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator(); } final DurableBeforeAccumulator durableBeforeAccumulator = new DurableBeforeAccumulator(); @@ -140,12 +135,6 @@ public class MockJournal implements IJournal return fieldUpdates(store).rangesForEpochAccumulator.get(); } - @Override - public List> loadHistoricalTransactions(long epoch, int store) - { - return fieldUpdates(store).historicalTransactionsAccumulator.get(); - } - @Override public void appendCommand(int store, SavedCommand.DiffWriter diff, Runnable onFlush) { @@ -177,8 +166,6 @@ public class MockJournal implements IJournal updates.safeToReadAccumulator.update(fieldUpdates.newSafeToRead); if (fieldUpdates.newRangesForEpoch != null) updates.rangesForEpochAccumulator.update(fieldUpdates.newRangesForEpoch); - if (fieldUpdates.addHistoricalTransactions != null) - updates.historicalTransactionsAccumulator.update(Pair.create(fieldUpdates.addHistoricalTransactions.range, fieldUpdates.addHistoricalTransactions.deps)); if (onFlush != null) 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 0108cd9cac..0e2c57bb97 100644 --- a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java @@ -47,7 +47,6 @@ 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