natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Benedict Elliott Smith 2024-10-11 23:09:22 +01:00 committed by David Capwell
parent 4e9110e881
commit 5a12875b3a
13 changed files with 60 additions and 228 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit a63cac24a2198a5893874cdf72946073854a8d4d
Subproject commit d77bdd1a4cd96120868279b665e0abe4ab509a80

95
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -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<K, V>() {
instance.register(new AccordStateCache.Listener<>() {
private final IdentityHashMap<AccordCachingState<?, ?>, 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<Pair<Range, Deps>> loaded = journal.loadHistoricalTransactions(epoch, id);
// synchronously load and register historical, so we don't have unlimited numbers of epochs in flight
for (Pair<Range, Deps> 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<Void> 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<TxnId, Command> 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);
}
}

14
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -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<Pair<Range, Deps>> 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;

60
src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
View File

@ -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<List<Pair<Range, Deps>>, Pair<Range, Deps>>
{
public HistoricalTransactionsAccumulator()
{
super(new ArrayList<>());
}
@Override
protected List<Pair<Range, Deps>> accumulate(List<Pair<Range, Deps>> oldValue, Pair<Range, Deps> deps)
{
accumulated.add(deps); // we can keep it mutable
return accumulated;
}
}
public static class HistoricalTransactionsSerializer implements FlyweightSerializer<Pair<Range, Deps>, HistoricalTransactionsAccumulator>
{
@Override
public HistoricalTransactionsAccumulator mergerFor(JournalKey key)
{
return new HistoricalTransactionsAccumulator();
}
@Override
public void serialize(JournalKey key, Pair<Range, Deps> 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<Range, Deps> 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)));
}
}
}
}

26
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
View File

@ -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<AccordSafeC
return super.redundantBefore();
}
@Override
public void registerHistoricalTransactions(long epoch, Range range, Deps deps)
{
ensureFieldUpdates().addHistoricalTransactions = new HistoricalTransactions(epoch, range, deps);
// TODO (required): it is potentially unsafe to propagate this synchronously, as if we fail to write to the journal we may be in an inconsistent state
// however, we can and should retire the concept of historical transactions in favour of ExclusiveSyncPoints ensuring their deps are known
super.registerHistoricalTransactions(epoch, range, deps);
}
private FieldUpdates ensureFieldUpdates()
{
if (fieldUpdates == null) fieldUpdates = new FieldUpdates();
@ -380,20 +369,5 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
public NavigableMap<TxnId, Ranges> newBootstrapBeganAt;
public NavigableMap<Timestamp, Ranges> 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;
}
}
}

17
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -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<? extends Request> requestHandler;
private final AccordResponseVerbHandler<? extends Reply> 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;
}

27
src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
View File

@ -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<TxnId, Command>
{
private final RoutesSearcher searcher = new RoutesSearcher();
//TODO (now, durability): find solution for this...
private final NavigableMap<TxnId, Ranges> historicalTransaction = new TreeMap<>();
private final AccordCommandStore store;
private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
private final ObjectHashSet<TxnId> 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<TxnId,
public AsyncResult<Pair<Watcher, NavigableMap<TxnId, Summary>>> 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<TxnId,
{
assert range instanceof TokenRange : "Require TokenRange but given " + range.getClass();
Set<TxnId> intersects = searcher.intersects(store.id(), (TokenRange) range, minTxnId, maxTxnId);
if (!historicalTransaction.isEmpty())
if (!transitive.isEmpty())
{
if (intersects.isEmpty())
intersects = new ObjectHashSet<>();
for (Map.Entry<TxnId, Ranges> e : historicalTransaction.tailMap(minTxnId, true).entrySet())
for (Map.Entry<TxnId, Ranges> 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<TxnId,
return new Summary(cmd.txnId, cmd.executeAt, saveStatus, ranges, null, false);
}
public void mergeHistoricalTransaction(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
{
historicalTransaction.merge(txnId, ranges, remappingFunction);
transitive.merge(txnId, ranges, remappingFunction);
}
public void gcBefore(TxnId gcBefore, Ranges ranges)
{
Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
while (iterator.hasNext())
{
Map.Entry<TxnId, Ranges> e = iterator.next();
Ranges newRanges = e.getValue().without(ranges);
if (newRanges.isEmpty())
iterator.remove();
e.setValue(newRanges);
}
}
public static class Summary

5
src/java/org/apache/cassandra/service/accord/IJournal.java
View File

@ -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<TxnId, Ranges> loadBootstrapBeganAt(int commandStoreId);
NavigableMap<Timestamp, Ranges> loadSafeToRead(int commandStoreId);
CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int commandStoreId);
List<Pair<Range, Deps>> loadHistoricalTransactions(long epoch, int store);
void appendCommand(int store, SavedCommand.DiffWriter value, Runnable onFlush);
Persister<DurableBefore, DurableBefore> durableBeforePersister();

11
src/java/org/apache/cassandra/service/accord/JournalKey.java
View File

@ -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);
}
}

7
src/java/org/apache/cassandra/service/accord/api/AccordScheduler.java
View File

@ -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)
{

10
test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java
View File

@ -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<Timestamp, Ranges> safeToReadAtAccumulator = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY);
NavigableMap<TxnId, Ranges> bootstrapBeganAtAccumulator = ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY);
RangesForEpoch.Snapshot rangesForEpochAccumulator = null;
HistoricalTransactionsAccumulator historicalTransactionsAccumulator = new HistoricalTransactionsAccumulator();
Gen<RedundantBefore> redundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner());
Gen<DurableBefore> 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<Pair<Range, Deps>> historical = historicalTransactionsAccumulator.get();
Collections.reverse(historical);
Assert.assertEquals(historical, journal.loadHistoricalTransactions(0l, 1));
}
finally
{

13
test/unit/org/apache/cassandra/service/accord/MockJournal.java
View File

@ -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<NavigableMap<TxnId, Ranges>> bootstrapBeganAtAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY));
final IdentityAccumulator<NavigableMap<Timestamp, Ranges>> safeToReadAccumulator = new IdentityAccumulator<>(ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY));
final IdentityAccumulator<CommandStores.RangesForEpoch.Snapshot> 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<Pair<Range, Deps>> 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();

1
test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
View File

@ -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