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

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
This commit is contained in:
Benedict Elliott Smith 2024-10-11 16:27:33 +01:00 committed by David Capwell
parent 91def31284
commit 22df42c7be
37 changed files with 602 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit dd04c81214be9213cf503025bf6ed4cbf118be00
Subproject commit d914ee69816ebfdf88b2120ff1d8e0bc16edecbc

33
src/java/org/apache/cassandra/concurrent/Stage.java
View File

@ -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<ExecutorPlus> executorSupplier;

22
src/java/org/apache/cassandra/config/AccordSpec.java
View File

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

28
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

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

1
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -130,6 +130,7 @@ public class Journal<K, V> 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<WaitingFor> remaining = new ArrayList<>();
for (WaitingFor wait : drained)

39
src/java/org/apache/cassandra/net/Message.java
View File

@ -91,6 +91,12 @@ public class Message<T> 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<T> 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<T> 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<T> 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()
{

17
src/java/org/apache/cassandra/schema/TableId.java
View File

@ -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<TableId>
public static final long MAGIC = 1956074401491665062L;
public static final long EMPTY_SIZE = ObjectSizes.measureDeep(new UUID(0, 0));
private static final ConcurrentHashMap<TableId, TableId> internCache = new ConcurrentHashMap<>();
private final UUID id;
private TableId(UUID id)
@ -200,10 +205,10 @@ public class TableId implements Comparable<TableId>
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<TableId>
return t.serializedSize();
}
};
public static void scheduleCachePruning()
{
ScheduledExecutors.scheduledFastTasks.scheduleSelfRecurring(internCache::clear, 1, TimeUnit.HOURS);
}
}

3
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

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

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

@ -547,7 +547,12 @@ public class AccordCachingState<K, V> extends IntrusiveLinkedListNode
Saving(V current, Runnable saveRunnable)
{
super(() -> { saveRunnable.run(); return null; });
this(current, () -> { saveRunnable.run(); return null; });
}
Saving(V current, Callable<Void> saveCallable)
{
super(saveCallable);
this.current = current;
}

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

@ -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<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;
@ -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> deps)
{
if (deps != null)
{
execute(PreLoadContext.empty(),
safeStore -> {
for (Deps dep : deps)
registerHistoricalTransactions(dep, safeStore);
});
}
}
public static class CommandStoreExecutor implements CacheSize
{
final AccordStateCache stateCache;

18
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -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<EpochReady> updateTopology(Node node, Topology newTopology, boolean startSync)
{
Supplier<EpochReady> start = super.updateTopology(node, newTopology, startSync);
return () -> {
EpochReady ready = start.get();
ready.metadata.addCallback(() -> {
synchronized (this)
{
refreshCacheSizes();
}
});
return ready;
};
}
public void waitForQuiescense()
{
boolean hadPending;

8
src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
View File

@ -234,7 +234,8 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
AtomicReference<Topology> 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<Acc
}
synchronized void reportMetadataInternal(ClusterMetadata metadata)
{
reportMetadataInternal(metadata, false);
}
synchronized void reportMetadataInternal(ClusterMetadata metadata, boolean isLoad)
{
updateMapping(metadata);
Topology topology = AccordTopology.createAccordTopology(metadata);

2
src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java
View File

@ -332,7 +332,7 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
}
@Override
public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean startSync)
public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync)
{
updatePeers(topology);
return SUCCESS;

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

@ -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<Deps> loadHistoricalTransactions(int store)
public List<Pair<Range, Deps>> 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> BUILDER readAll(JournalKey key)
{
BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key);

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

@ -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<List<Deps>, Deps>
public static class HistoricalTransactionsAccumulator extends Accumulator<List<Pair<Range, Deps>>, Pair<Range, Deps>>
{
public HistoricalTransactionsAccumulator()
{
@ -346,14 +349,14 @@ public class AccordJournalValueSerializers
}
@Override
protected List<Deps> accumulate(List<Deps> oldValue, Deps deps)
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<Deps, HistoricalTransactionsAccumulator>
public static class HistoricalTransactionsSerializer implements FlyweightSerializer<Pair<Range, Deps>, 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<Range, Deps> 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<Range, Deps> 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)));
}
}
}
}

30
src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
View File

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

16
src/java/org/apache/cassandra/service/accord/AccordResponseVerbHandler.java
View File

@ -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<T extends Reply> implements IVerbHandler<T>
{
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<T extends Reply> implements IVerbHandler<T>
@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())
{

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

@ -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<AccordSafeC
}
@Override
protected void registerHistoricalTransactions(Deps deps)
public void registerHistoricalTransactions(long epoch, Range range, Deps deps)
{
ensureFieldUpdates().addHistoricalTransactions = 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(deps);
super.registerHistoricalTransactions(epoch, range, deps);
}
private FieldUpdates ensureFieldUpdates()
@ -379,6 +380,20 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
public NavigableMap<TxnId, Ranges> newBootstrapBeganAt;
public NavigableMap<Timestamp, Ranges> 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;
}
}
}

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

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

36
src/java/org/apache/cassandra/service/accord/AccordStateCache.java
View File

@ -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<AccordCachingState<?,?
listeners.forEach(l -> l.onAdd(finalNode));
}
}
AccordCachingState<K, V> acquired = acquireExisting(node, true);
AccordCachingState<K, V> 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<AccordCachingState<?,?
if (node == null)
return null;
return safeRefFactory.apply(acquireExisting(node, false));
return safeRefFactory.apply(acquireExisting(node, false, null));
}
public void maybeLoad(K key, V initial)
@ -401,37 +403,49 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
public S acquire(K key)
{
AccordCachingState<K, V> node = acquire(key, false);
return safeRefFactory.apply(node);
return acquire(key, null);
}
public S acquireIfLoaded(K key)
{
AccordCachingState<K, V> node = acquire(key, true);
return acquireIfLoaded(key, null);
}
public S acquire(K key, @Nullable ExecutorPlus loadExecutor)
{
AccordCachingState<K, V> node = acquire(key, false, loadExecutor);
return safeRefFactory.apply(node);
}
public S acquireIfLoaded(K key, @Nullable ExecutorPlus loadExecutor)
{
AccordCachingState<K, V> node = acquire(key, true, loadExecutor);
if (node == null)
return null;
return safeRefFactory.apply(node);
}
private AccordCachingState<K, V> acquire(K key, boolean onlyIfLoaded)
private AccordCachingState<K, V> acquire(K key, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor)
{
incrementCacheQueries();
@SuppressWarnings("unchecked")
AccordCachingState<K, V> node = (AccordCachingState<K, V>) 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<K, V> acquireAbsent(K key, boolean onlyIfLoaded)
private AccordCachingState<K, V> acquireAbsent(K key, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor)
{
incrementCacheMisses();
if (onlyIfLoaded)
return null;
AccordCachingState<K, V> 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<AccordCachingState<?,?
/*
* Can't return EVICTED or INITIALIZED
*/
private AccordCachingState<K, V> acquireExisting(AccordCachingState<K, V> node, boolean onlyIfLoaded)
private AccordCachingState<K, V> acquireExisting(AccordCachingState<K, V> node, boolean onlyIfLoaded, @Nullable ExecutorPlus loadExecutor)
{
Status status = node.status(); // status() completes
@ -462,6 +476,8 @@ public class AccordStateCache extends IntrusiveLinkedList<AccordCachingState<?,?
if (node.references == 0)
{
if (loadExecutor == null)
loadExecutor = AccordStateCache.this.loadExecutor;
if (status == FAILED_TO_LOAD || status == EVICTED)
node.reset().load(loadExecutor, loadFunction);

9
src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -26,10 +27,12 @@ import accord.local.Node;
import accord.messages.Request;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.utils.NoSpamLogger;
public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
{
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<T extends Request> implements IVerbHandler<T>
{
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<T extends Request> implements IVerbHandler<T>
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);
});
}
}

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

@ -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<TxnId,
private final NavigableMap<TxnId, Ranges> historicalTransaction = new TreeMap<>();
private final AccordCommandStore store;
private final ObjectHashSet<TxnId> 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<TxnId,
TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVERY ? primaryTxnId : null;
Watcher watcher = fromCache(findAsDep, ranges, minTxnId, maxTxnId, redundantBefore);
ImmutableMap<TxnId, Summary> 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<TxnId,
{
if (cacheHits.containsKey(txnId))
continue;
Command cmd = store.loadCommand(txnId);
if (cmd == null)
continue; // unknown command
Summary summary = create(cmd, ranges, findAsDep, redundantBefore);
if (summary == null)
continue;
map.put(txnId, summary);
if (findAsDep == null)
{
SavedCommand.MinimalCommand cmd = store.loadMinimal(txnId);
if (cmd == null)
continue; // unknown command
Summary summary = create(cmd, ranges, redundantBefore);
if (summary == null)
continue;
map.put(txnId, summary);
}
else
{
Command cmd = store.loadCommand(txnId);
if (cmd == null)
continue; // unknown command
Summary summary = create(cmd, ranges, findAsDep, redundantBefore);
if (summary == null)
continue;
map.put(txnId, summary);
}
}
return map;
}
@ -262,13 +276,11 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener<TxnId,
if (redundantBefore != null)
{
Ranges durableAlready = Ranges.of(redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
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<Range>(), 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<TxnId,
return new Summary(cmd.txnId(), cmd.executeAt(), saveStatus, ranges, findAsDep, hasAsDep);
}
private static Summary create(SavedCommand.MinimalCommand cmd, Ranges cacheRanges, @Nullable RedundantBefore redundantBefore)
{
//TODO (required, correctness): C* did Invalidated, accord-core did Erased... what is correct?
SaveStatus saveStatus = cmd.saveStatus;
if (saveStatus == null
|| saveStatus == SaveStatus.Invalidated
|| saveStatus == SaveStatus.Erased
|| !saveStatus.hasBeen(Status.PreAccepted))
return null;
if (cmd.participants == null)
return null;
Ranges keysOrRanges = cmd.participants.touches().toRanges();
if (keysOrRanges.domain() != Domain.Range)
throw new AssertionError(String.format("Txn keys are not range for %s", cmd.participants));
Ranges ranges = (Ranges) keysOrRanges;
ranges = ranges.slice(cacheRanges, Routables.Slice.Minimal);
if (ranges.isEmpty())
return null;
if (redundantBefore != null)
{
Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
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<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
{
historicalTransaction.merge(txnId, ranges, remappingFunction);

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

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

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

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

90
src/java/org/apache/cassandra/service/accord/SavedCommand.java
View File

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

4
src/java/org/apache/cassandra/service/accord/api/AccordRoutingKey.java
View File

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

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

@ -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 <V> PartitionKey deserialize(V src, ValueAccessor<V> 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);

94
src/java/org/apache/cassandra/service/accord/async/AsyncLoader.java
View File

@ -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<K, V, S> cache,
List<AsyncChain<?>> listenChains)
{
S safeRef = cache.acquire(key);
referenceAndAssembleReadsForKey(key, context, cache, listenChains, null);
}
private static <K, V, S extends AccordSafeState<K, V>> void referenceAndAssembleReadsForKey(K key,
Map<K, S> context,
AccordStateCache.Instance<K, V, S> cache,
List<AsyncChain<?>> 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<AsyncChain<?>> listenChains)
{
referenceAndAssembleReadsForKey(key, context, listenChains, null);
}
private void referenceAndAssembleReadsForKey(RoutingKey key,
AsyncOperation.Context context,
List<AsyncChain<?>> 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<AsyncChain<?>> root = new ArrayList<>(ranges.size() + 1);
class Watcher implements AccordStateCache.Listener<RoutingKey, CommandsForKey>
{
// TODO (required): streams prohibited in hot path
private final Set<TokenKey> cached = commandStore.commandsForKeyCache().stream()
.map(n -> (TokenKey) n.key())
.filter(ranges::contains)
@ -196,7 +236,7 @@ public class AsyncLoader
return AsyncChains.success(null);
Set<? extends RoutingKey> set = ImmutableSet.<RoutingKey>builder().addAll(watcher.cached).addAll(keys).build();
List<AsyncChain<?>> 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));

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

@ -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<R> extends AsyncChains.Head<R> implements Runnable, Function<SafeCommandStore, R>
public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements Runnable, Function<SafeCommandStore, R>, Cancellable
{
private static final Logger logger = LoggerFactory.getLogger(AsyncOperation.class);
@ -360,12 +361,18 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
}
@Override
public void start(BiConsumer<? super R, Throwable> callback)
public Cancellable start(BiConsumer<? super R, Throwable> 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<R> extends AsyncOperation<R>

15
src/java/org/apache/cassandra/utils/NoSpamLogger.java
View File

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

5
src/java/org/apache/cassandra/utils/memory/BufferPool.java
View File

@ -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<Chunk>) attachment).release();
return true;

5
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

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

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

@ -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<DurableBefore> durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner());
Gen<NavigableMap<Timestamp, Ranges>> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner());
Gen<RangesForEpoch.Snapshot> rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner());
Gen<Range> rangeGen = AccordGenerators.range(DatabaseDescriptor.getPartitioner());
Gen<Deps> 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<Deps> historical = historicalTransactionsAccumulator.get();
List<Pair<Range, Deps>> historical = historicalTransactionsAccumulator.get();
Collections.reverse(historical);
Assert.assertEquals(historical, journal.loadHistoricalTransactions(1));
Assert.assertEquals(historical, journal.loadHistoricalTransactions(0l, 1));
}
finally
{

6
test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
View File

@ -507,13 +507,15 @@ public class YamlConfigurationLoaderTest
public void testAccordConfig()
{
Map<String, String> 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);

9
test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
View File

@ -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<? super T, Throwable> callback)
protected Cancellable start(BiConsumer<? super T, Throwable> 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<Void> onTopologyUpdate(Topology topology, boolean startSync)
public AsyncResult<Void> onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync)
{
// EpochReady ready = EpochReady.done(topology.epoch());
AsyncResult<Void> metadata = schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult();

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

@ -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<LoadedDiff> 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<Deps> loadHistoricalTransactions(int store)
public List<Pair<Range, Deps>> 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();
}

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

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