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

Improve Rebootstrap: 

- Distinguish corrupt/incomplete cases; latter case only refuses uncertain data
     - Ensures quorum durability before marking own log faulty
     - Refuse unsafe operations more selectively once log marked faulty
     - Simplify and merge logic with standard bootstrap
This commit is contained in:
Benedict Elliott Smith 2026-05-19 15:07:59 +01:00
parent 988235c736
commit ba683a02fd
39 changed files with 1946 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitmodules vendored
View File

@ -1,4 +1,4 @@
[submodule "modules/accord"]
path = modules/accord
url = https://github.com/belliottsmith/cassandra-accord.git
branch = executor-fair
branch = rerebootstrap

2
doc/modules/cassandra/pages/managing/configuration/configuration.adoc
View File

@ -153,7 +153,7 @@ configuration changes in the near future.
`@Replaces` is the annotation to be used when you make changes to any configuration parameters in `Config` class and `cassandra.yaml`, and you want to add backward
compatibility with previous Cassandra versions. `Converters` class enumerates the different methods used for backward compatibility.
`IDENTITY` is the one used for name change only. For more information about the other Converters, please, check the JavaDoc in the class.
For backward compatibility virtual table `Settings` contains both the old and the new
For backward compatibility virtual table `LoadSettings` contains both the old and the new
parameters with the old and the new value format. Only exception at the moment are the following three parameters: `key_cache_save_period`,
`row_cache_save_period` and `counter_cache_save_period` which appear only once with the new value format.
The old names and value format still can be used at least until the next major release. Deprecation warning is emitted on startup.

2
modules/accord

@ -1 +1 @@
Subproject commit 138e96dd0f0b9e315be21a8659f1aa81ec48637a
Subproject commit 673cf2ff93a4b32a392c91feacda09bb48ac16e7

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

@ -47,7 +47,7 @@ public enum Stage
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::getAccordConcurrentOps, DatabaseDescriptor::setConcurrentAccordOps, Stage::multiThreadedLowSignalStage),
ACCORD_MIGRATION (false, "AccordMigrationStage", "request", DatabaseDescriptor::getAccordConcurrentMigrationOps, DatabaseDescriptor::setConcurrentAccordMigrationOps, 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),

42
src/java/org/apache/cassandra/config/AccordConfig.java
View File

@ -34,7 +34,7 @@ import org.apache.cassandra.journal.Params;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.config.AccordConfig.CatchupMode.NORMAL;
import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.HLC_FIFO;
import static org.apache.cassandra.config.AccordConfig.QueueShardModel.THREAD_POOL_PER_SHARD;
import static org.apache.cassandra.config.AccordConfig.QueueSubmissionModel.SYNC;
import static org.apache.cassandra.config.AccordConfig.RangeIndexMode.in_memory;
@ -214,6 +214,10 @@ public class AccordConfig
* once this number of tasks are blocked behind it, regardless of batch_size.
*/
public Integer queue_nonsync_blocked_limit;
/**
* The number of threads that may be used to execute distributed requests for migration tasks
*/
public volatile OptionaldPositiveInt migration_concurrency = OptionaldPositiveInt.UNDEFINED;
/**
* If set, the signal loop does not match park/unpark pairs, but instead consumers perform timed-park spin waits
@ -258,6 +262,7 @@ public class AccordConfig
public String expire_epoch_wait = "10s";
// we don't want to wait ages for durability as it blocks other durability progress; even this might be too long, as we can always retry
public String expire_durability = "10s*attempts <= 30s";
public String slow_durability = "10s";
public String slow_syncpoint_preaccept = "10s";
public String slow_txn_preaccept = "30ms <= p50*2 <= 1000ms";
public String slow_read = "30ms <= p50*2 <= 1000ms";
@ -294,12 +299,12 @@ public class AccordConfig
*/
public volatile TransactionalRangeMigration range_migration = TransactionalRangeMigration.auto;
public enum CatchupMode
public enum CatchupFallbackMode
{
DISABLED,
NORMAL,
FALLBACK_TO_HARD,
HARD
IGNORE,
EXIT,
REBOOTSTRAP,
REBOOTSTRAP_AND_CATCHUP
}
/**
@ -344,16 +349,19 @@ public class AccordConfig
public int commands_for_key_prune_interval = 64;
public DurationSpec.IntSecondsBound max_conflicts_prune_delta = new DurationSpec.IntSecondsBound(1);
public int catchup_on_start_max_attempts = 5;
public boolean catchup_on_start = true;
public DurationSpec.IntSecondsBound catchup_on_start_success_latency = new DurationSpec.IntSecondsBound(60);
public DurationSpec.IntSecondsBound catchup_on_start_fail_latency = new DurationSpec.IntSecondsBound(900);
public int catchup_on_start_max_attempts = 5;
// TODO (required): roll this back to catchup_on_start_exit_on_failure: true
public boolean catchup_on_start_exit_on_failure = false;
public CatchupMode catchup_on_start = NORMAL;
// TODO (required): default this to EXIT or REBOOTSTRAP
public CatchupFallbackMode catchup_on_start_on_timeout = CatchupFallbackMode.IGNORE;
public CatchupFallbackMode catchup_on_start_on_error = CatchupFallbackMode.IGNORE;
public CatchupFallbackMode catchup_on_start_on_rebootstrap_fallback = CatchupFallbackMode.IGNORE;
public DurationSpec.IntSecondsBound shutdown_grace_period = new DurationSpec.IntSecondsBound(15 * 60);
public boolean execute_waiting_on_start = true;
public DurationSpec.IntSecondsBound execute_waiting_on_start_timeout = new DurationSpec.IntSecondsBound(0);
public boolean execute_waiting_on_start_fail_on_timeout = false;
public DurationSpec.IntSecondsBound shutdown_grace_period = new DurationSpec.IntSecondsBound(15 * 60);
public enum RangeIndexMode { in_memory, journal_sai }
public RangeIndexMode range_index_mode = in_memory;
@ -391,7 +399,17 @@ public class AccordConfig
* Replay journal entries for commands that are not durable to the data or command stores.
* THIS MODE IS NOT YET SAFE TO RUN
*/
NON_DURABLE
NON_DURABLE,
/**
* Don't replay, simply rebootstrap, marking our log as incomplete.
*/
REBOOTSTRAP_INCOMPLETE,
/**
* Don't replay, simply rebootstrap, marking our log as corrupted/unavailable.
*/
REBOOTSTRAP_RESET
}
public enum ReplaySavePoint

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

@ -2967,6 +2967,20 @@ public class DatabaseDescriptor
conf.accord.queue_thread_count = new OptionaldPositiveInt(concurrent_operations);
}
public static int getAccordConcurrentMigrationOps()
{
return conf.accord.migration_concurrency.or(2 * FBUtilities.getAvailableProcessors());
}
public static void setConcurrentAccordMigrationOps(int concurrent_operations)
{
if (concurrent_operations < 0)
{
throw new IllegalArgumentException("Concurrent accord operations must be non-negative");
}
conf.accord.migration_concurrency = new OptionaldPositiveInt(concurrent_operations);
}
public static int getFlushWriters()
{
return conf.memtable_flush_writers;

27
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -62,7 +62,8 @@ import accord.impl.CommandChange;
import accord.impl.progresslog.DefaultProgressLog;
import accord.impl.progresslog.DefaultProgressLog.ModeFlag;
import accord.impl.progresslog.TxnStateKind;
import accord.local.CatchupHard;
import accord.local.BootstrapReason;
import accord.local.Catchup;
import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommandStore;
@ -166,6 +167,8 @@ import org.apache.cassandra.utils.concurrent.Future;
import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
import static accord.impl.CommandChange.Field.ACCEPTED;
import static accord.impl.CommandChange.Field.PROMISED;
import static accord.local.BootstrapReason.LOG_CORRUPTED;
import static accord.local.BootstrapReason.LOG_INCOMPLETE;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
@ -183,6 +186,7 @@ import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
import static org.apache.cassandra.db.virtual.AbstractLazyVirtualTable.OnTimeout.BEST_EFFORT;
import static org.apache.cassandra.db.virtual.AbstractLazyVirtualTable.OnTimeout.FAIL;
import static org.apache.cassandra.db.virtual.AccordDebugKeyspace.CommandStoreOpsTable.CommandStoreOp.REBOOTSTRAP_CORRUPTED;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.ASC;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.SORTED;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.UNSORTED;
@ -1988,8 +1992,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace
UNSET_PROGRESS_LOG_MODE("Unset the specified progress log mode."),
TRY_EXECUTE_LISTENING("Try to execute all of the transactions (and their dependencies) that have registered listeners on other transactions."),
REPLAY("Run journal replay for all transactions"),
REBOOTSTRAP("Rebootstrap the command store. This invalidates the local journal, synchronises its data via data repair and rejoins the distributed state machine."),
HARD_CATCHUP("Hard catchup the command store. This invalidates the local journal for any ranges not up to date with some quorum, synchronises its data via data repair and rejoins the distributed state machine."),
// TODO (expected): specify ranges
REBOOTSTRAP_CORRUPTED("Rebootstrap the command store. This invalidates the local journal, synchronises its data via data repair and rejoins the distributed state machine."),
REBOOTSTRAP_INCOMPLETE("Rebootstrap the command store. This invalidates the local journal, synchronises its data via data repair and rejoins the distributed state machine."),
REBOOTSTRAP_IF_BEHIND("Hard catchup the command store. This invalidates the local journal for any ranges not up to date with some quorum, synchronises its data via data repair and rejoins the distributed state machine."),
;
final String description;
@ -2082,13 +2088,15 @@ public class AccordDebugKeyspace extends VirtualKeyspace
};
break;
}
case REBOOTSTRAP:
allFunction = () -> node.commandStores().rebootstrap(node);
function = commandStore -> commandStore.rebootstrap(node);
case REBOOTSTRAP_CORRUPTED:
case REBOOTSTRAP_INCOMPLETE:
BootstrapReason reason = op == REBOOTSTRAP_CORRUPTED ? LOG_CORRUPTED : LOG_INCOMPLETE;
allFunction = () -> node.commandStores().rebootstrap(node, reason);
function = commandStore -> commandStore.rebootstrap(node, reason);
break;
case HARD_CATCHUP:
allFunction = () -> CatchupHard.catchup(node, Arrays.asList(node.commandStores().all())).beginAsResult();
function = commandStore -> CatchupHard.catchup(node, Collections.singletonList(commandStore)).beginAsResult();
case REBOOTSTRAP_IF_BEHIND:
allFunction = () -> Catchup.rebootstrapIfBehind(node, Arrays.asList(node.commandStores().all())).beginAsResult();
function = commandStore -> Catchup.rebootstrapIfBehind(node, Collections.singletonList(commandStore)).beginAsResult();
break;
}
@ -2484,7 +2492,6 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
throw new InvalidRequestException("Unknown bucket_mode '" + value + '\'');
}
}
private static int checkNonNegative(Object value, String field, int ifNull)

11
src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java
View File

@ -72,10 +72,13 @@ public class ExceptionSerializer
static String getMessageWithOriginatingHost(Throwable t, boolean isFirstException)
{
if (isFirstException)
return "Remote exception from host " + FBUtilities.getBroadcastAddressAndPort().toString() + (t.getLocalizedMessage() != null ? " - " + t.getLocalizedMessage() : "");
else
return t.getLocalizedMessage();
String message;
if (isFirstException) message = "Remote exception from host " + FBUtilities.getBroadcastAddressAndPort().toString() + (t.getLocalizedMessage() != null ? " - " + t.getLocalizedMessage() : "");
else message = t.getLocalizedMessage();
if (message.length() > Short.MAX_VALUE)
message = message.substring(0, Short.MAX_VALUE);
return message;
}
private static final IVersionedSerializer<StackTraceElement> stackTraceElementSerializer = new IVersionedSerializer<StackTraceElement>()

7
src/java/org/apache/cassandra/journal/Compactor.java
View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.journal;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@ -29,6 +28,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.journal.SegmentCompactor.Stopped;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
@ -100,9 +100,9 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
compacted.signalAll();
}
catch (IOException e)
catch (Stopped e)
{
throw new RuntimeException("Could not compact segments: " + toCompact);
logger.info("Cancelling compaction of {}", toCompact);
}
}
@ -118,6 +118,7 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
logger.debug("Shutting down " + executor);
if (scheduled != null)
scheduled.cancel(false);
segmentCompactor.stop();
executor.shutdown();
}

6
src/java/org/apache/cassandra/journal/SegmentCompactor.java
View File

@ -17,18 +17,20 @@
*/
package org.apache.cassandra.journal;
import java.io.IOException;
import java.util.Collection;
public interface SegmentCompactor<K, V>
{
SegmentCompactor<?, ?> NOOP = (SegmentCompactor<Object, Object>) (segments) -> segments;
class Stopped extends RuntimeException {}
static <K, V> SegmentCompactor<K, V> noop()
{
//noinspection unchecked
return (SegmentCompactor<K, V>) NOOP;
}
Collection<StaticSegment<K, V>> compact(Collection<StaticSegment<K, V>> segments) throws IOException;
Collection<StaticSegment<K, V>> compact(Collection<StaticSegment<K, V>> segments) throws Stopped;
default void stop() {}
}

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

@ -254,8 +254,10 @@ public class AccordCommandStore extends CommandStore
maybeLoadRedundantBefore(journal.loadRedundantBefore(id()));
maybeLoadBootstrapBeganAt(journal.loadBootstrapBeganAt(id()));
maybeLoadSafeToRead(journal.loadSafeToRead(id()));
tableId = (TableId)rangesForEpoch.all().stream().map(r -> r.start().prefix()).reduce((a, b) -> {
maybeLoadRangesForEpoch(journal.loadRangesForEpoch(id()));
RangesForEpoch ranges = this.rangesForEpoch;
Invariants.require(ranges != null && !ranges.all().isEmpty(), "CommandStore %d created with no ranges", id);
tableId = (TableId)ranges.all().stream().map(r -> r.start().prefix()).reduce((a, b) -> {
Invariants.require(a.equals(b), "CommandStore created with multiple distinct TableId (%s and %s)", a, b);
return a;
}).orElseThrow(() -> Invariants.illegalState("CommandStore %d created with no ranges", id));
@ -824,7 +826,7 @@ public class AccordCommandStore extends CommandStore
{
File rjbf = new File(savePoint, "reject_before");
if (rjbf.exists())
mxc = mxc.with(readOne(rjbf, rejectBefore));
mxc = mxc.update(readOne(rjbf, rejectBefore));
}
dll = readList(new File(savePoint, "listeners"), txnListener);
dpl = readList(new File(savePoint, "progress_log"), progressLogState);

81
src/java/org/apache/cassandra/service/accord/AccordDataStore.java
View File

@ -20,24 +20,23 @@ package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.DataStore;
import accord.local.CommandStore;
import accord.local.CommandStores.RestrictedStoreSelector;
import accord.local.MapReduceConsumeCommandStores;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.primitives.Ranges;
import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.Reduce;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.ScheduledExecutors;
@ -56,10 +55,9 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordDurableOnFlush.ReportDurable;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressListener;
import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.utils.Invariants.illegalArgument;
public class AccordDataStore implements DataStore
@ -94,12 +92,12 @@ public class AccordDataStore implements DataStore
AccordDurableOnFlush.notifyOnDurable(cfs, commandStore, ReportDurable.of(redundantBefore, flags));
}
public FetchResult image(Node node, SafeCommandStore safeStore, Ranges ranges, SyncPoint syncPoint, FetchRanges callback)
public FetchResult image(Node node, SafeCommandStore safeStore, Ranges ranges, TxnId atLeast, SortedArrayList<Node.Id> readable, FetchRanges callback)
{
AccordFetchCoordinator coordinator;
try
{
coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
coordinator = new AccordFetchCoordinator(node, ranges, atLeast, readable, callback, safeStore.commandStore());
}
catch (Throwable t)
{
@ -110,11 +108,10 @@ public class AccordDataStore implements DataStore
return coordinator.result();
}
public FetchResult sync(Node node, SafeCommandStore safeStore, Map<TxnId, Ranges> rangesById, FetchRanges callback)
@Override
public FetchResult sync(Node node, SafeCommandStore safeStore, Ranges ranges, TxnId atLeast, SortedArrayList<Node.Id> readable, FetchRanges callback)
{
TableId tableId = ((AccordCommandStore)safeStore.commandStore()).tableId();
Invariants.require(rangesById.values().stream().flatMap(Ranges::stream).allMatch(r -> tableId.equals(r.prefix())));
Ranges ranges = rangesById.values().stream().reduce(Ranges::with).get();
ClusterMetadata cm = ClusterMetadata.current();
TableMetadata tableMetadata = cm.schema.getTableMetadata(tableId);
@ -137,27 +134,22 @@ public class AccordDataStore implements DataStore
// TODO (expected): add some automatic slicing of ranges and retry/back-off logic; but for now,
// since this is done at the command store level, and this is already a slice of a node, this should be fine
SyncResult syncResult = new SyncResult();
ProgressListener listener = new ProgressListener()
{
StartingRangeFetch starting = callback.starting(ranges);
{ Invariants.require(starting != null); }
logger.info("Requesting quorum durability before initiating repair");
List<AsyncResult<Void>> syncs = new ArrayList<>();
for (Map.Entry<TxnId, Ranges> e : rangesById.entrySet())
syncs.add(node.durability().sync("Sync Data Store", ExclusiveSyncPoint, e.getKey(), e.getValue(), NoLocal, Quorum, 1L, TimeUnit.HOURS));
AsyncResults.reduce(syncs, Reduce.toNull()).invoke((success, fail) -> {
if (fail != null)
@Override
public void progress(String tag, ProgressEvent event)
{
logger.error("{} failed to achieve quorum durability before repair for rebootstrap of {}", safeStore.commandStore(), ranges);
syncResult.tryFailure(fail);
return;
}
RepairCoordinator coord = StorageService.instance.newRepairCoordinator(tableMetadata.keyspace, options(tableMetadata, ranges));
coord.addProgressListener((tag, event) -> {
switch (event.getType())
{
default: throw new UnhandledEnum(event.getType());
case SUCCESS:
callback.fetched(ranges);
syncResult.trySuccess(null);
case START:
callback.starting(ranges);
reportStarted();
break;
case PROGRESS:
case COMPLETE:
@ -169,17 +161,32 @@ public class AccordDataStore implements DataStore
callback.fail(ranges, ex);
syncResult.tryFailure(ex);
break;
case SUCCESS:
callback.fetched(ranges);
syncResult.trySuccess(null);
break;
}
});
}
ScheduledExecutors.optionalTasks.submit(coord).addCallback((s, f) -> {
if (f != null)
syncResult.tryFailure(f);
});
private void reportStarted()
{
StartingRangeFetch start = this.starting;
if (start == null)
return;
this.starting = null;
node.commandStores().mapReduceConsume(new RestrictedStoreSelector(ranges, 0, Long.MAX_VALUE), new MapReduceConsumeCommandStores<Ranges, Timestamp>(ranges)
{
@Override public Timestamp reduce(Timestamp o1, Timestamp o2) { return Timestamp.max(o1, o2); }
@Override public void accept(Timestamp result, Throwable failure) { start.started(result); }
@Override public TxnId primaryTxnId() { return null; }
@Override public String reason() { return "Compute MaxConflict to report for fetch"; }
@Override protected Timestamp applyInternal(SafeCommandStore safeStore) { return safeStore.commandStore().maxConflict(TxnId.NONE, ranges); }
});
}
};
RepairCoordinator coord = StorageService.instance.newRepairCoordinator(tableMetadata.keyspace, options(tableMetadata, ranges));
coord.addProgressListener(listener);
ScheduledExecutors.optionalTasks.submit(coord).addCallback((s, f) -> {
if (f != null)
syncResult.tryFailure(f);
});
return syncResult;

29
src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java
View File

@ -38,7 +38,6 @@ import accord.impl.AbstractFetchCoordinator;
import accord.local.CommandStore;
import accord.local.Node;
import accord.local.SafeCommandStore;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
import accord.primitives.Range;
@ -46,12 +45,12 @@ import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.TopologyException;
import accord.utils.Invariants;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
@ -395,9 +394,9 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
private final Map<TimeUUID, IncomingStream> streams = new HashMap<>();
public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore) throws TopologyException
public AccordFetchCoordinator(Node node, Ranges ranges, TxnId atLeast, SortedArrayList<Node.Id> readable, DataStore.FetchRanges fetchRanges, CommandStore commandStore) throws TopologyException
{
super(node, node.someExclusiveExecutor(), ranges, syncPoint, fetchRanges, commandStore);
super(node, node.someExclusiveExecutor(), ranges, atLeast, readable, fetchRanges, commandStore);
}
@Override
@ -431,13 +430,6 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
super.onDone(success, failure);
}
@Override
protected PartialTxn rangeReadTxn(Ranges ranges)
{
StreamingRead read = new StreamingRead(FBUtilities.getBroadcastAddressAndPort(), ranges);
return new PartialTxn.InMemory(Txn.Kind.Read, ranges, read, noopQuery, null, TableMetadatasAndKeys.none(Routable.Domain.Range));
}
@Override
protected synchronized void onReadOk(Node.Id from, CommandStore commandStore, Data data, Ranges received)
{
@ -463,9 +455,9 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
public static class AccordFetchRequest extends FetchRequest
{
public AccordFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn)
public AccordFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialTxn partialTxn)
{
super(sourceEpoch, syncId, ranges, partialDeps, partialTxn);
super(sourceEpoch, syncId, ranges, partialTxn);
}
@Override
@ -479,8 +471,15 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
}
@Override
protected FetchRequest newFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges, PartialDeps partialDeps, PartialTxn partialTxn)
protected FetchRequest newFetchRequest(long sourceEpoch, TxnId syncId, Ranges ranges)
{
return new AccordFetchRequest(sourceEpoch, syncId, ranges, partialDeps, partialTxn);
return new AccordFetchRequest(sourceEpoch, syncId, ranges, rangeReadTxn(ranges));
}
private PartialTxn rangeReadTxn(Ranges ranges)
{
StreamingRead read = new StreamingRead(FBUtilities.getBroadcastAddressAndPort(), ranges);
return new PartialTxn.InMemory(Txn.Kind.Read, ranges, read, noopQuery, null, TableMetadatasAndKeys.none(Routable.Domain.Range));
}
}

3
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
View File

@ -106,6 +106,7 @@ import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.schema.Types;
import org.apache.cassandra.schema.UserFunctions;
import org.apache.cassandra.schema.Views;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.utils.AbstractIterator;
@ -341,7 +342,7 @@ public class AccordKeyspace
{
ByteBuffer bytes;
if (serialized instanceof ByteBuffer) bytes = (ByteBuffer) serialized;
else bytes = Serialize.toBytesWithoutKey(commandsForKey.maximalPrune()); // TODO (expected): we only need to strip pruned, not prune additional txns
else bytes = Serialize.toBytesWithoutKey(commandsForKey.maybePrune(0, AccordAgent.cfkHlcPruneDelta(DatabaseDescriptor.getAccord()))); // TODO (expected): we only need to strip pruned, not prune additional txns
return makeUpdate(storeId, key, timestampMicros, bytes);
}

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

@ -254,6 +254,14 @@ public class AccordMessageSink implements MessageSink
slowAtNanos = nowNanos + slow.computeWait(attempt, NANOSECONDS);
break;
}
case ACCORD_WAIT_UNTIL_APPLIED_REQ:
{
TimeoutStrategy slow = slowPreaccept(txnId);
if (slow != null)
slowAtNanos = nowNanos + slow.computeWait(attempt, NANOSECONDS);
break;
}
}
Message<Request> message = Message.out(verb, request, expiresAtNanos);

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

@ -201,7 +201,9 @@ public class AccordResult<V> extends AsyncFuture<V> implements BiConsumer<V, Thr
}
else
{
logger.error("Unexpected exception", fail);
AccordAgent.handleException(fail);
if (!AccordAgent.expectedException(fail))
logger.error("Unexpected exception", fail);
JVMStabilityInspector.inspectThrowable(fail);
report = bookkeeping.newFailed(txnId, keysOrRanges);
}

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

@ -56,8 +56,9 @@ import accord.impl.RequestCallbacks;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.progresslog.DefaultProgressLog;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.BootstrapReason;
import accord.local.Catchup;
import accord.local.CatchupHard;
import accord.local.Catchup.Unsuccessful;
import accord.local.CommandStores;
import accord.local.Node;
import accord.local.Node.Id;
@ -92,7 +93,6 @@ import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.AccordConfig;
import org.apache.cassandra.config.AccordConfig.CatchupMode;
import org.apache.cassandra.config.AccordConfig.JournalConfig.ReplaySavePoint;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -166,6 +166,8 @@ import static accord.api.Journal.TopologyUpdate;
import static accord.api.ProtocolModifiers.FastExecution.MAY_BYPASS_SAFESTORE;
import static accord.coordinate.Coordination.CoordinationKind.Client;
import static accord.impl.progresslog.DefaultProgressLog.ModeFlag.CATCH_UP;
import static accord.local.BootstrapReason.LOG_CORRUPTED;
import static accord.local.BootstrapReason.LOG_INCOMPLETE;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
@ -179,8 +181,9 @@ import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.concurrent.ExecutorFactory.SimulatorThreadTag.JOB;
import static org.apache.cassandra.concurrent.ExecutorFactory.SystemThreadTag.DAEMON;
import static org.apache.cassandra.config.AccordConfig.CatchupMode.FALLBACK_TO_HARD;
import static org.apache.cassandra.config.AccordConfig.CatchupMode.HARD;
import static org.apache.cassandra.config.AccordConfig.CatchupFallbackMode.REBOOTSTRAP;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.REBOOTSTRAP_INCOMPLETE;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.REBOOTSTRAP_RESET;
import static org.apache.cassandra.config.AccordConfig.JournalConfig.ReplayMode.RESET;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccord;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordGlobalDurabilityCycle;
@ -534,7 +537,16 @@ public class AccordService implements IAccordService, Shutdownable
if (state != State.INIT)
return;
boolean rebootstrap = false;
BootstrapReason rebootstrap = null;
if (getAccord().journal.replay == REBOOTSTRAP_RESET)
{
rebootstrap = LOG_CORRUPTED;
}
else if (getAccord().journal.replay == REBOOTSTRAP_INCOMPLETE)
{
rebootstrap = LOG_INCOMPLETE;
}
else
{
long startMarker = ReplayMarkers.readStartMarker();
long stopMarker = ReplayMarkers.readStopMarker();
@ -553,7 +565,7 @@ public class AccordService implements IAccordService, Shutdownable
case REBOOTSTRAP:
logger.info("Stop marker is older than start marker ({}<{}). Rebootstrapping.", stopMarker, startMarker);
rebootstrap = true;
rebootstrap = LOG_INCOMPLETE;
}
}
}
@ -594,7 +606,7 @@ public class AccordService implements IAccordService, Shutdownable
// restore save points before starting the journal so we can validate consistency between journal and save point state (where possible)
Long2LongHashMap minSegments;
if (rebootstrap) minSegments = null;
if (rebootstrap != null) minSegments = null;
else
{
switch (getAccord().journal.replaySavePoint)
@ -616,11 +628,11 @@ public class AccordService implements IAccordService, Shutdownable
}
node.commandStores().forAllUnsafe(cs -> cs.unsafeProgressLog().start());
if (rebootstrap)
if (rebootstrap != null)
{
// rebootstrap expects the durability service to process visibility sync points for command stores
node.durability().start();
getBlocking(node.commandStores().rebootstrap(node));
getBlocking(node.commandStores().rebootstrap(node, rebootstrap));
}
else
{
@ -707,81 +719,100 @@ public class AccordService implements IAccordService, Shutdownable
void catchup()
{
AccordConfig spec = DatabaseDescriptor.getAccord();
if (spec.catchup_on_start == CatchupMode.DISABLED)
if (!spec.catchup_on_start)
{
logger.info("Catchup disabled; continuing to startup");
return;
}
CatchupMode mode = spec.catchup_on_start;
BootstrapState bootstrapState = SystemKeyspace.getBootstrapState();
if (bootstrapState == COMPLETED)
{
node.commandStores().forAllUnsafe(commandStore -> ((DefaultProgressLog)commandStore.unsafeProgressLog()).setMode(CATCH_UP));
try
{
AccordConfig.CatchupFallbackMode onError = spec.catchup_on_start_on_error;
AccordConfig.CatchupFallbackMode onTimeout = spec.catchup_on_start_on_timeout;
long maxLatencyNanos = spec.catchup_on_start_fail_latency.toNanoseconds();
int attempts = 1;
while (true)
{
logger.info("Catchup ({}) with quorum...", mode);
logger.info("Catchup with quorum...");
long start = nanoTime();
long failAt = start + maxLatencyNanos;
Future<Void> f;
Future<Unsuccessful> f;
{
AsyncChain<Void> submit;
if (mode == HARD)
{
if (!node.durability().isStarted())
node.durability().start();
submit = CatchupHard.catchup(node);
}
else submit = Catchup.catchup(node);
AsyncChain<Unsuccessful> submit = Catchup.catchup(node, failAt, NANOSECONDS);
f = toFuture(submit);
}
if (!f.awaitUntilThrowUncheckedOnInterrupt(failAt))
{
if (spec.catchup_on_start_exit_on_failure)
{
logger.error("Catchup exceeded maximum latency of {}ns; shutting down", maxLatencyNanos);
throw new RuntimeException("Could not catchup with peers");
}
logger.error("Catchup exceeded maximum latency of {}ns; continuing to startup", maxLatencyNanos);
break;
}
f.awaitThrowUncheckedOnInterrupt();
Throwable failed = f.cause();
Unsuccessful result = f.getNow();
if (failed != null)
{
if (spec.catchup_on_start_exit_on_failure)
throw new RuntimeException("Could not catchup with peers", failed);
logger.error("Could not catchup with peers; continuing to startup");
break;
switch (onError)
{
default: throw new UnhandledEnum(onError);
case EXIT:
logger.error("Could not catchup with peers; exiting", failed);
throw new RuntimeException("Could not catchup with peers", failed);
case IGNORE:
logger.error("Could not catchup with peers; continuing to startup", failed);
return;
case REBOOTSTRAP:
case REBOOTSTRAP_AND_CATCHUP:
logger.error("Could not catchup with peers; rebootstrapping", failed);
getBlocking(node.commandStores().rebootstrap(node, LOG_INCOMPLETE));
if (onError == REBOOTSTRAP)
return;
++attempts;
onError = onTimeout = spec.catchup_on_start_on_rebootstrap_fallback;
continue;
}
}
long end = nanoTime();
double seconds = NANOSECONDS.toMillis(end - start)/1000.0;
logger.info("Finished catchup with all quorums. {}s elapsed.", String.format("%.2f", seconds));
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
break;
if (++attempts > spec.catchup_on_start_max_attempts)
if (result == null)
{
if (spec.catchup_on_start_exit_on_failure)
{
logger.error("Catchup was slow, aborting after {} attempts and shutting down", attempts);
throw new RuntimeException("Could not catchup with peers");
}
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
return;
logger.info("Catchup was slow; continuing to startup after {} attempts.", attempts - 1);
break;
if (attempts < spec.catchup_on_start_max_attempts)
{
logger.info("Catchup was slow, so we may behind again; retrying");
++attempts;
continue;
}
}
logger.info("Catchup was slow, so we may behind again; retrying");
if (mode == FALLBACK_TO_HARD)
mode = HARD;
switch (onTimeout)
{
default: throw new UnhandledEnum(onTimeout);
case EXIT:
if (result == null) logger.error("Catchup was slow; aborting after {} attempts and shutting down", attempts);
else logger.error("Catchup was incomplete for {}; aborting after {} attempts and shutting down", result.ranges, attempts);
throw new RuntimeException("Could not catchup with peers");
case IGNORE:
if (result == null) logger.info("Catchup was slow, continuing to startup after {} attempts", attempts);
else logger.info("Catchup was incomplete for {}, continuing to startup after {} attempts", result.ranges, attempts);
return;
case REBOOTSTRAP:
case REBOOTSTRAP_AND_CATCHUP:
if (result == null) logger.info("Catchup was slow, rebootstrapping after {} attempts", attempts);
else logger.info("Catchup was incomplete for {}, rebootstrapping after {} attempts", result.ranges, attempts);
getBlocking(node.commandStores().rebootstrap(node, result == null ? null : result.ranges, LOG_INCOMPLETE));
if (onTimeout == REBOOTSTRAP)
return;
++attempts;
onError = onTimeout = spec.catchup_on_start_on_rebootstrap_fallback;
continue;
}
}
}
finally
@ -954,7 +985,7 @@ public class AccordService implements IAccordService, Shutdownable
}
@Override
public AsyncResult<Void> sync(Object requestedBy, TxnId minBound, Ranges ranges, @Nullable Collection<Id> include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
public AsyncResult<?> sync(Object requestedBy, TxnId minBound, Ranges ranges, @Nullable Collection<Id> include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
{
return node.durability().sync(requestedBy, ExclusiveSyncPoint, minBound, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits);
}

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

@ -88,8 +88,8 @@ public interface IAccordService
IVerbHandler<? extends Request> requestHandler();
IVerbHandler<? extends Reply> responseHandler();
AsyncResult<Void> sync(Object requestedBy, @Nullable TxnId minBound, Ranges ranges, @Nullable Collection<Id> include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits);
AsyncChain<Void> sync(@Nullable TxnId minBound, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote);
AsyncResult<?> sync(Object requestedBy, @Nullable TxnId minBound, Ranges ranges, @Nullable Collection<Id> include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits);
AsyncChain<?> sync(@Nullable TxnId minBound, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote);
AsyncChain<Timestamp> maxConflict(Ranges ranges);
@Nonnull
@ -402,13 +402,13 @@ public interface IAccordService
}
@Override
public AsyncResult<Void> sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection<Id> include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
public AsyncResult<?> sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection<Id> include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
{
return delegate.sync(requestedBy, onOrAfter, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits);
}
@Override
public AsyncChain<Void> sync(@Nullable TxnId onOrAfter, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote)
public AsyncChain<?> sync(@Nullable TxnId onOrAfter, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote)
{
return delegate.sync(onOrAfter, keys, syncLocal, syncRemote);
}

62
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -40,9 +40,6 @@ import accord.api.RoutingKey;
import accord.api.Tracing;
import accord.coordinate.Coordination;
import accord.coordinate.CoordinationFailed;
import accord.coordinate.Exhausted;
import accord.coordinate.Preempted;
import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.LogUnavailableException;
@ -51,6 +48,7 @@ import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.TimeService;
import accord.messages.MessageType;
import accord.primitives.Ballot;
import accord.primitives.Keys;
import accord.primitives.Participants;
import accord.primitives.Ranges;
@ -266,6 +264,11 @@ public class AccordAgent implements Agent, OwnershipEventListener
// TODO (expected): we probably want additional configuration here so we can prune on shorter time horizons when we have a lot of transactions on a single key
@Override
public long cfkHlcPruneDelta()
{
return cfkHlcPruneDelta(config);
}
public static long cfkHlcPruneDelta(AccordConfig config)
{
return config.commands_for_key_prune_delta.to(MICROSECONDS);
}
@ -325,24 +328,12 @@ public class AccordAgent implements Agent, OwnershipEventListener
@Override
public long slowCoordinatorDelay(Node node, SafeCommandStore safeStore, TxnId txnId, TimeUnit units, int attempt)
{
SafeCommand safeCommand = safeStore.unsafeGetNoCleanup(txnId);
if (safeCommand == null)
{
noSpamLogger.warn("{} invoked slowCoordinatorDelay for {} without having it in cache", safeStore.commandStore(), txnId, new RuntimeException());
return recover(txnId).computeWait(attempt, units);
}
Command command = safeCommand.current();
if (command == null)
{
noSpamLogger.warn("{} invoked slowCoordinatorDelay for {} without knowing the command", safeStore.commandStore(), txnId, new RuntimeException());
return recover(txnId).computeWait(attempt, units);
}
Command command = command(safeStore, txnId, "slowCoordinatorDelay");
Ballot promised = promised(command);
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
long mostRecentStart = mostRecentStart(command, nowMicros);
long mostRecentStart = mostRecentStart(txnId, promised, nowMicros);
long waitMicros = recover(txnId).computeWait(attempt, MICROSECONDS);
long startTime = mostRecentStart + waitMicros;
if (startTime < nowMicros)
@ -355,7 +346,7 @@ public class AccordAgent implements Agent, OwnershipEventListener
}
RoutingKey homeKey = command.route().homeKey();
Shard shard = node.topology().active().forEpochIfKnown(homeKey, command.txnId().epoch());
Shard shard = node.topology().active().forEpochIfKnown(homeKey, txnId.epoch());
startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), ONE_SECOND, random);
long delayMicros = Math.max(1, startTime - nowMicros);
@ -363,15 +354,15 @@ public class AccordAgent implements Agent, OwnershipEventListener
return units.convert(delayMicros, MICROSECONDS);
}
private static long mostRecentStart(Command command, long nowMicros)
private static long mostRecentStart(TxnId txnId, Ballot promised, long nowMicros)
{
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long promisedHlc = command.promised().hlc();
long promisedHlc = promised.hlc();
if (promisedHlc > nowMicros + ONE_MINUTE)
promisedHlc = 0;
long result = Math.max(command.txnId().hlc(), promisedHlc);
long result = Math.max(txnId.hlc(), promisedHlc);
if (result > nowMicros + ONE_SECOND)
noSpamLogger.warn("max({},{})>{}", command.txnId(), command.promised(), nowMicros);
noSpamLogger.warn("max({},{})>{}", txnId, promised, nowMicros);
return result;
}
@ -405,25 +396,36 @@ public class AccordAgent implements Agent, OwnershipEventListener
return newStartTime;
}
@Override
public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil blockedUntil, TimeUnit units)
private static Ballot promised(@Nullable Command command)
{
return command == null ? Ballot.ZERO : command.promised();
}
private static Command command(SafeCommandStore safeStore, TxnId txnId, String source)
{
SafeCommand safeCommand = safeStore.unsafeGetNoCleanup(txnId);
if (safeCommand == null)
{
noSpamLogger.warn("{} invoked slowReplicaDelay for {} without having it in cache", safeStore.commandStore(), txnId, new RuntimeException());
return fetch(txnId).computeWait(attempt, units);
noSpamLogger.warn("{} invoked {} for {} without having it in cache", safeStore.commandStore(), source, txnId, new RuntimeException());
return null;
}
Command command = safeCommand.current();
if (command == null)
{
noSpamLogger.warn("{} invoked slowReplicaDelay for {} without knowing the command", safeStore.commandStore(), txnId, new RuntimeException());
return fetch(txnId).computeWait(attempt, units);
noSpamLogger.warn("{} invoked {} for {} without knowing the command", safeStore.commandStore(), source, txnId, new RuntimeException());
return null;
}
return command;
}
@Override
public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int attempt, BlockedUntil blockedUntil, TimeUnit units)
{
Ballot promised = promised(command(safeStore, txnId, "slowReplicaDelay"));
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
long mostRecentStart = mostRecentStart(command, nowMicros);
long mostRecentStart = mostRecentStart(txnId, promised, nowMicros);
long waitMicros = fetch(txnId).computeWait(attempt, units);
long startTime = mostRecentStart + waitMicros;
if (startTime < nowMicros)

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

@ -38,7 +38,7 @@ import static org.apache.cassandra.service.TimeoutStrategy.LatencySourceFactory.
public class AccordWaitStrategies
{
static TimeoutStrategy slowTxnPreaccept, slowSyncPointPreaccept, slowRead;
static TimeoutStrategy slowTxnPreaccept, slowSyncPointPreaccept, slowRead, slowDurability;
static TimeoutStrategy expireTxn, expireSyncPoint, expireDurability, expireEpochWait;
static TimeoutStrategy fetchTxn, fetchSyncPoint;
static RetryStrategy recoverTxn, recoverSyncPoint, retrySyncPoint, retryDurability, retryBootstrap, retryJoinBootstrap;
@ -88,6 +88,7 @@ public class AccordWaitStrategies
setSlowRead(config.slow_read);
setSlowTxnPreaccept(config.slow_txn_preaccept);
setSlowSyncPointPreaccept(config.slow_syncpoint_preaccept);
setSlowDurability(config.slow_durability);
setExpireTxn(config.expire_txn);
setExpireSyncPoint(config.expire_syncpoint);
setExpireDurability(config.expire_durability);
@ -119,6 +120,11 @@ public class AccordWaitStrategies
slowSyncPointPreaccept = TimeoutStrategy.parse(spec, none());
}
public static void setSlowDurability(String spec)
{
slowDurability = TimeoutStrategy.parse(spec, none());
}
public static void setExpireTxn(String spec)
{
expireTxn = TimeoutStrategy.parse(spec, rw(accordReadMetrics, accordWriteMetrics));

20
src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java
View File

@ -47,6 +47,7 @@ import org.apache.cassandra.service.accord.topology.AccordEndpointMapper;
import org.apache.cassandra.service.accord.txn.AccordUpdate;
import org.apache.cassandra.service.accord.txn.TxnRead;
import static accord.api.ProtocolModifiers.sendMinimal;
import static accord.messages.Apply.Kind.Maximal;
import static accord.messages.Apply.Kind.Minimal;
@ -55,12 +56,13 @@ public class AccordInteropAdapter extends TxnAdapter
private static final Logger logger = LoggerFactory.getLogger(AccordInteropAdapter.class);
public static final class AccordInteropFactory extends DefaultFactory
{
final AccordInteropAdapter standard, recovery;
final AccordInteropAdapter minimal, maximal, recovery;
public AccordInteropFactory(AccordEndpointMapper endpointMapper)
{
standard = new AccordInteropAdapter(endpointMapper, Minimal);
recovery = new AccordInteropAdapter(endpointMapper, Maximal);
minimal = new AccordInteropAdapter(endpointMapper, Minimal, true);
maximal = new AccordInteropAdapter(endpointMapper, Maximal, true);
recovery = new AccordInteropAdapter(endpointMapper, Maximal, false);
}
@Override
@ -68,18 +70,18 @@ public class AccordInteropAdapter extends TxnAdapter
{
if (txnId.isSyncPoint())
return super.get(txnId, step);
return (CoordinationAdapter<R>) (step == Kind.Recovery ? recovery : standard);
return (CoordinationAdapter<R>) (step == Kind.Recovery ? recovery : sendMinimal() ? minimal : maximal);
}
};
private final AccordEndpointMapper endpointMapper;
private final Apply.Kind applyKind;
private final boolean isUserFacing;
private AccordInteropAdapter(AccordEndpointMapper endpointMapper, Apply.Kind applyKind)
private AccordInteropAdapter(AccordEndpointMapper endpointMapper, Apply.Kind applyKind, boolean isUserFacing)
{
super(Minimal);
super(applyKind);
this.endpointMapper = endpointMapper;
this.applyKind = applyKind;
this.isUserFacing = isUserFacing;
}
@Override
@ -92,7 +94,7 @@ public class AccordInteropAdapter extends TxnAdapter
@Override
public void persist(Node node, ExclusiveAsyncExecutor executor, Topologies any, Route<?> require, Route<?> sendTo, FullRoute<?> route, Ballot ballot, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer<? super Result, Throwable> callback)
{
if (applyKind == Minimal && doInteropPersist(node, executor, any, require, sendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback))
if (isUserFacing && doInteropPersist(node, executor, any, require, sendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback))
return;
super.persist(node, executor, any, require, sendTo, route, ballot, flags, txnId, txn, executeAt, deps, writes, result, informDurableOnDone, callback);

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

@ -94,6 +94,7 @@ import org.apache.cassandra.service.reads.ReadCoordinator;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.Dispatcher;
import static accord.api.ProtocolModifiers.sendMinimal;
import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard;
import static accord.primitives.Txn.Kind.Write;
import static accord.topology.SelectShards.LIVE;
@ -144,6 +145,7 @@ public class AccordInteropExecution implements ReadCoordinator
ExclusiveAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) throws TopologyException
{
requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR);
// TODO (required): this does not support privileged coordinator optimisation, as must apply local stable before any other message is sent
this.node = node;
this.txnId = txnId;
this.txn = txn;
@ -217,8 +219,7 @@ public class AccordInteropExecution implements ReadCoordinator
// TODO (desired): It would be better to use the re-use the command from the transaction but it's fragile
// to try and figure out exactly what changed for things like read repair and short read protection
// Also this read scope doesn't reflect the contents of this particular read and is larger than it needs to be
// TODO (required): understand interop and whether StableFastPath is appropriate
AccordInteropStableThenRead commit = new AccordInteropStableThenRead(id, allTopologies, txnId, Kind.StableFastPath, executeAt, txn, deps, route, message.payload);
AccordInteropStableThenRead commit = new AccordInteropStableThenRead(id, allTopologies, txnId, commitKind(), executeAt, txn, deps, route, message.payload);
node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this), null);
}
@ -376,7 +377,7 @@ public class AccordInteropExecution implements ReadCoordinator
{
InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(to);
if (endpoint != null && !contacted.contains(endpoint))
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
node.send(to, new Commit(commitKind(), to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
}
}
@ -387,7 +388,7 @@ public class AccordInteropExecution implements ReadCoordinator
for (Node.Id to : allTopologies.nodes())
{
if (!executeTopology.contains(to))
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
node.send(to, new Commit(commitKind(), to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
}
}
AsyncChain<Data> result;
@ -415,6 +416,11 @@ public class AccordInteropExecution implements ReadCoordinator
});
}
private static Commit.Kind commitKind()
{
return sendMinimal() ? Kind.StableFastPath : Kind.StableWithTxnAndDeps;
}
private AsyncChain<Data> executeUnrecoverableRepairUpdate()
{
return AsyncChains.chain(Stage.ACCORD_MIGRATION.executor(), () -> {
@ -423,7 +429,7 @@ public class AccordInteropExecution implements ReadCoordinator
// and can be extended similar to MessageType which allows additional types not from Accord to be added
// This commit won't necessarily execute before the interop read repair message so there could be an insufficient which is fine
for (Node.Id to : executeTopology.nodes())
node.send(to, new Commit(Kind.StableFastPath, to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
node.send(to, new Commit(commitKind(), to, allTopologies, txnId, txn, route, Ballot.ZERO, executeAt, deps), null);
repairUpdate.runBRR(AccordInteropExecution.this);
return new TxnData();
});

10
src/java/org/apache/cassandra/service/accord/journal/AbstractSegmentCompactor.java
View File

@ -93,6 +93,7 @@ public abstract class AbstractSegmentCompactor<V> implements SegmentCompactor<Jo
// Only valid in the scope of a single `compact` call
private JournalKey prevKey;
private DecoratedKey prevDecoratedKey;
private volatile boolean stopped;
@Override
public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment<JournalKey, V>> segments)
@ -129,6 +130,9 @@ public abstract class AbstractSegmentCompactor<V> implements SegmentCompactor<Jo
KeyOrderReader<JournalKey> reader;
while ((reader = readers.poll()) != null)
{
if (stopped)
throw new Stopped();
if (key == null || !reader.key().equals(key))
{
maybeWritePartition(key, merger, serializer, firstDescriptor, firstOffset);
@ -232,6 +236,12 @@ public abstract class AbstractSegmentCompactor<V> implements SegmentCompactor<Jo
}
}
@Override
public void stop()
{
stopped = true;
}
private static int normalize(int cmp)
{
if (cmp == 0)

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

@ -39,6 +39,7 @@ import accord.local.DurableBefore;
import accord.local.MaxConflicts;
import accord.local.MaxDecidedRX;
import accord.local.RedundantBefore;
import accord.local.RedundantStatus;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.SaveStatus;
@ -274,7 +275,9 @@ public class CommandStoreSerializers
private short cast(long v)
{
if ((v & ~0xFFFF) != 0)
if (v < 0 || v >= (1 << RedundantStatus.Property.WAS_OWNED.ordinal()))
throw new IllegalStateException("Should not be serializing WAS_OWNED or NOT_OWNED statuses");
if ((v & ~0xFFFF) != 0) // retain this
throw new IllegalStateException("Cannot serialize RedundantStatus larger than 0xFFFF. Requires serialization changes.");
return (short)v;
}

20
src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
View File

@ -25,7 +25,11 @@ import accord.impl.AbstractFetchCoordinator.FetchRequest;
import accord.impl.AbstractFetchCoordinator.FetchResponse;
import accord.messages.ReadData.CommitOrReadNack;
import accord.messages.ReadData.ReadReply;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.VIntCoding;
import org.apache.cassandra.db.TypeSizes;
@ -51,18 +55,20 @@ public class FetchSerializers
out.writeUnsignedVInt(request.executeAtEpoch);
CommandSerializers.txnId.serialize(request.txnId, out);
KeySerializers.ranges.serialize((Ranges) request.scope, out);
DepsSerializers.partialDeps.serialize(request.partialDeps, out);
DepsSerializers.partialDeps.serialize(PartialDeps.NONE, out);
StreamingTxn.serializer.serialize(request.read, out, version);
}
@Override
public FetchRequest deserialize(DataInputPlus in, Version version) throws IOException
{
return new AccordFetchRequest(in.readUnsignedVInt(),
CommandSerializers.txnId.deserialize(in),
KeySerializers.ranges.deserialize(in),
DepsSerializers.partialDeps.deserialize(in),
StreamingTxn.serializer.deserialize(in, version));
long epoch = in.readUnsignedVInt();
TxnId txnId = CommandSerializers.txnId.deserialize(in);
Ranges ranges = KeySerializers.ranges.deserialize(in);
PartialDeps deps = DepsSerializers.partialDeps.deserialize(in);
Invariants.require(deps.isEmpty());
PartialTxn txn = StreamingTxn.serializer.deserialize(in, version);
return new AccordFetchRequest(epoch, txnId, ranges, txn);
}
@Override
@ -71,7 +77,7 @@ public class FetchSerializers
return TypeSizes.sizeofUnsignedVInt(request.executeAtEpoch)
+ CommandSerializers.txnId.serializedSize(request.txnId)
+ KeySerializers.ranges.serializedSize((Ranges) request.scope)
+ DepsSerializers.partialDeps.serializedSize(request.partialDeps)
+ DepsSerializers.partialDeps.serializedSize(PartialDeps.NONE)
+ StreamingTxn.serializer.serializedSize(request.read, version);
}
};

3
src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
View File

@ -74,6 +74,7 @@ import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import org.apache.cassandra.utils.vint.VIntCoding;
import static accord.local.BootstrapReason.GAIN_OWNERSHIP;
import static com.google.common.collect.ImmutableList.of;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.tcm.MultiStepOperation.Kind.JOIN;
@ -396,7 +397,7 @@ public class BootstrapAndJoin extends MultiStepOperation<Epoch>
Node node = service.node();
node.commandStores().forAllUnsafe(commandStore -> {
AsyncResult<EpochReady> ready = commandStore.resumeBootstrap(node);
AsyncResult<EpochReady> ready = commandStore.resumeBootstrap(node, GAIN_OWNERSHIP);
bootstraps.add(AccordService.toFuture(ready.flatMap(e -> e.reads)));
});
}

56
test/conf/logback-info.xml Normal file
View File

@ -0,0 +1,56 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<configuration debug="false" scan="true" scanPeriod="60 seconds">
<define name="cluster_id" class="org.apache.cassandra.distributed.impl.ClusterIDDefiner" />
<define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
<!-- Shutdown hook ensures that async appender flushes -->
<shutdownHook class="ch.qos.logback.core.hook.DefaultShutdownHook"/>
<appender name="INSTANCEFILE" class="ch.qos.logback.core.FileAppender">
<file>./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log</file>
<encoder>
<pattern>%-5level [%thread] ${instance_id} %date{"yyyy-MM-dd'T'HH:mm:ss,SSS", UTC} %F:%L - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
<immediateFlush>true</immediateFlush>
</appender>
<appender name="INSTANCESTDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%-5level %date{"HH:mm:ss,SSS"} %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
</appender>
<!-- Un-comment to enable TRACE logging for Accord transactions.
<logger name="accord" level="TRACE" />
<logger name="org.apache.cassandra.service.accord" level="TRACE" />
-->
<root level="INFO">
<appender-ref ref="INSTANCEFILE" /> <!-- use blocking to avoid race conditions with appending and searching -->
<appender-ref ref="INSTANCESTDERR" />
</root>
</configuration>

27
test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java
View File

@ -81,6 +81,7 @@ import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.QueryResults;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.distributed.shared.FutureUtils;
import org.apache.cassandra.distributed.test.sai.SAIUtil;
import org.apache.cassandra.distributed.util.QueryResultUtil;
import org.apache.cassandra.exceptions.InvalidRequestException;
@ -93,6 +94,7 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
import org.apache.cassandra.utils.AssertionUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FailingConsumer;
import org.apache.cassandra.utils.Pair;
@ -116,8 +118,10 @@ import static org.junit.Assert.fail;
public abstract class AccordCQLTestBase extends AccordTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordCQLTestBase.class);
private static final int CAS_SIMULATOR_LITE_CONCURRENCY = 20;
protected AccordCQLTestBase(TransactionalMode transactionalMode) {
protected AccordCQLTestBase(TransactionalMode transactionalMode)
{
super(transactionalMode);
}
@ -131,7 +135,9 @@ public abstract class AccordCQLTestBase extends AccordTestBase
public static void setupClass() throws IOException
{
AccordTestBase.setupCluster(builder -> builder.appendConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)
.set("paxos_variant", PaxosVariant.v2.name())), 2);
.set("paxos_variant", PaxosVariant.v2.name())
.set("accord.migration_concurrency", "" + (1 + CAS_SIMULATOR_LITE_CONCURRENCY))
), 2);
SHARED_CLUSTER.schemaChange("CREATE TYPE " + KEYSPACE + ".person (height int, age int)");
}
@ -3332,15 +3338,18 @@ public abstract class AccordCQLTestBase extends AccordTestBase
coordinator.execute("INSERT INTO " + qualifiedAccordTableName + " (pk, count, seq1, seq2) VALUES (1, 0, '', []) USING TIMESTAMP 0", ConsistencyLevel.ALL);
ListType<Integer> LIST_TYPE = ListType.getInstance(Int32Type.instance, true);
ExecutorService es = Executors.newCachedThreadPool();
List<Future<Object[][]>> futures = new ArrayList<>();
for (int ii = 0; ii < 10; ii++)
Future<Object[][]>[] futures = new Future[CAS_SIMULATOR_LITE_CONCURRENCY];
long[] starts = new long[CAS_SIMULATOR_LITE_CONCURRENCY];
for (int id = 0; id < CAS_SIMULATOR_LITE_CONCURRENCY; id++)
{
int id = ii;
futures.add(es.submit(() -> coordinator.execute("UPDATE " + qualifiedAccordTableName + " SET count = count + 1, seq1 = seq1 + ?, seq2 = seq2 + ? WHERE pk = ? IF EXISTS", ConsistencyLevel.ALL, id + ",", ByteBufferUtil.getArray(LIST_TYPE.decompose(singletonList(id))), 1)));
starts[id] = Clock.Global.nanoTime();
futures[id] = FutureUtils.map(coordinator.asyncExecuteWithResult("UPDATE " + qualifiedAccordTableName + " SET count = count + 1, seq1 = seq1 + ?, seq2 = seq2 + ? WHERE pk = ? IF EXISTS", ConsistencyLevel.ALL, id + ",", ByteBufferUtil.getArray(LIST_TYPE.decompose(singletonList(id))), 1), SimpleQueryResult::toObjectArrays);
}
for (int id = 0; id < CAS_SIMULATOR_LITE_CONCURRENCY; id++)
{
futures[id].get();
System.out.println(String.format("Completed in %.3fs", (Clock.Global.nanoTime() - starts[id]) * 0.000000001));
}
for (Future f : futures)
f.get();
Object[][] result = coordinator.execute("SELECT pk, count, seq1, seq2 FROM " + qualifiedAccordTableName + " WHERE pk = 1", ConsistencyLevel.SERIAL);

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
View File

@ -89,7 +89,7 @@ public class AccordIncrementalRepairTest extends TestBaseImpl
}
@Override
public AsyncResult<Void> sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection<Node.Id> include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
public AsyncResult<?> sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection<Node.Id> include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
{
return delegate.sync(requestedBy, onOrAfter, ranges, include, syncLocal, syncRemote, 10L, TimeUnit.MINUTES).map(v -> {
executedBarriers = true;
@ -98,7 +98,7 @@ public class AccordIncrementalRepairTest extends TestBaseImpl
}
@Override
public AsyncChain<Void> sync(@Nullable TxnId onOrAfter, Keys keys, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote)
public AsyncChain<?> sync(@Nullable TxnId onOrAfter, Keys keys, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote)
{
return delegate.sync(onOrAfter, keys, syncLocal, syncRemote).map(v -> {
executedBarriers = true;

7
test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
View File

@ -170,7 +170,12 @@ public abstract class AccordTestBase extends TestBaseImpl
{
SHARED_CLUSTER.filters().reset();
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> AccordService.instance().node().commandStores().forAllUnsafe(cs -> cs.unsafeProgressLog().start()));
{
instance.runOnInstance(() -> {
if (AccordService.isStarted())
AccordService.instance().node().commandStores().forAllUnsafe(cs -> cs.unsafeProgressLog().start());
});
}
truncateSystemTables();

976
test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordLoadTestBase.java Normal file
View File

@ -0,0 +1,976 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test.accord.load;
import java.io.IOException;
import java.security.SecureRandom;
import java.time.Duration;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.function.Consumer;
import java.util.function.Supplier;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Snapshot;
import com.codahale.metrics.Timer;
import com.google.common.util.concurrent.RateLimiter;
import org.agrona.collections.IntArrayList;
import org.junit.Before;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.Catchup;
import accord.local.CommandStore;
import accord.local.Node;
import accord.local.ExecutionContext;
import accord.local.SafeCommand;
import accord.primitives.PartialDeps;
import accord.primitives.TxnId;
import accord.utils.Functions;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.distributed.test.accord.AccordTestBase;
import org.apache.cassandra.distributed.test.accord.load.LoadSettings.ClusterChaos;
import org.apache.cassandra.metrics.AccordCoordinatorMetrics;
import org.apache.cassandra.metrics.AccordExecutorMetrics;
import org.apache.cassandra.metrics.ShardedDecayingHistograms.ShardedDecayingHistogram;
import org.apache.cassandra.metrics.ShardedHistogram;
import org.apache.cassandra.metrics.SnapshottingTimer;
import org.apache.cassandra.net.ArtificialLatency;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.debug.AccordTracing;
import org.apache.cassandra.service.accord.debug.AccordTracing.Message;
import org.apache.cassandra.service.accord.debug.CoordinationKinds;
import org.apache.cassandra.service.accord.debug.TxnKindsAndDomains;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.EstimatedHistogram;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static accord.coordinate.Coordination.CoordinationKind.Client;
import static accord.coordinate.Coordination.CoordinationKind.Execute;
import static accord.local.BootstrapReason.LOG_CORRUPTED;
import static accord.local.BootstrapReason.LOG_INCOMPLETE;
import static java.lang.System.currentTimeMillis;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
import static org.apache.cassandra.distributed.test.accord.load.LoadSettings.ClusterChaos.REBOOTSTRAP_RESET;
import static org.apache.cassandra.service.accord.debug.AccordTracing.BucketMode.LEAKY;
import static org.apache.cassandra.service.accord.debug.AccordTracing.BucketMode.RING;
import static org.apache.cassandra.service.accord.debug.AccordTracing.BucketMode.SLOWEST;
public class AccordLoadTestBase extends AccordTestBase
{
private static long CHAOS_TIMEOUT_NANOS = TimeUnit.MINUTES.toNanos(10L);
private static final Logger logger = LoggerFactory.getLogger(AccordLoadTestBase.class);
@Before
public void setup()
{
setupCluster();
super.setup();
}
public void setupCluster()
{
setupCluster(5);
}
public void setupCluster(int nodeCount)
{
setupCluster(nodeCount, config -> {
config.with(Feature.NETWORK, Feature.GOSSIP)
.set("accord.shard_durability_target_splits", "8")
.set("accord.shard_durability_max_splits", "16")
.set("accord.shard_durability_cycle", "1m")
.set("accord.queue_submission_model", "SIGNAL")
.set("accord.command_store_shard_count", "8")
.set("accord.queue_thread_count", "4")
.set("accord.queue_shard_count", "1")
.set("accord.replica_execution", "ALL")
.set("accord.send_stable", "TO_ALL_REPLICA_EXECUTABLE_ELSE_FOR_READS")
.set("accord.send_minimal", "false")
.set("accord.catchup_on_start_fail_latency", "2m");
});
}
public void setupCluster(int nodeCount, Consumer<IInstanceConfig> configure)
{
Invariants.require(SHARED_CLUSTER == null);
try { SHARED_CLUSTER = createCluster(nodeCount, builder -> builder.withDCs(nodeCount).withConfig(configure)); }
catch (IOException e) { throw new RuntimeException(e); }
CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis())));
}
public void testLoad(final LoadSettings settings) throws Exception
{
Cluster cluster = SHARED_CLUSTER;
cluster.schemaChange("CREATE TABLE " + qualifiedAccordTableName + " (k int, v int, PRIMARY KEY(k)) WITH transactional_mode = 'full'");
// long seed = new SecureRandom().nextLong();
long seed = 3705626102508196273L;
try
{
final ConcurrentHashMap<Verb, AtomicInteger> verbs = new ConcurrentHashMap<>();
cluster.filters().outbound().messagesMatching(new IMessageFilters.Matcher()
{
@Override
public boolean matches(int i, int i1, IMessage iMessage)
{
verbs.computeIfAbsent(Verb.fromId(iMessage.verb()), ignore -> new AtomicInteger()).incrementAndGet();
return false;
}
}).drop();
boolean waitForTransactions = settings.totalTransactions < Long.MAX_VALUE;
boolean waitForClusterChaos = settings.totalClusterChaos < Long.MAX_VALUE;
int clientCount = settings.clients < 0 ? cluster.size() : settings.clients;
long nextRepairAt = settings.repairInterval;
long nextCompactionAt = settings.compactionInterval;
long nextJournalFlushAt = settings.journalFlushInterval;
long nextDataFlushAt = settings.dataFlushInterval;
long nextCfkFlushAt = settings.cfkFlushInterval;
long nextChaosAt = settings.clusterChaosInterval;
final ExecutorService chaosExecutor = Executors.newFixedThreadPool(settings.clusterChaosConcurrency, new NamedThreadFactory("ClusterChaos"));
final ExecutorService clientExecutor = Executors.newFixedThreadPool(clientCount, new NamedThreadFactory("Client"));
final BitSet initialised = new BitSet();
final Supplier<ClusterChaos> clusterChaos = settings.clusterChaos.apply(seed);
List<ChaosActive> chaosActive = new ArrayList<>();
cluster.get(1).nodetoolResult("cms", "reconfigure", "datacenter1:1", "datacenter2:1", "datacenter3:1").asserts().success();
if (settings.cfkCompactionPeriodSeconds < Integer.MAX_VALUE && settings.cfkCompactionPeriodSeconds > 0)
{
cluster.forEach(i -> i.acceptOnInstance(period -> {
((AccordService) AccordService.instance()).journal().compactor().updateCompactionPeriod(period, SECONDS);
}, settings.cfkCompactionPeriodSeconds));
}
if (settings.artificialLatencies != null)
{
for (int i = 0 ; i < cluster.size() ; ++i)
{
StringBuilder str = new StringBuilder();
for (int j = 0 ; j < settings.artificialLatencies[i].length ; ++j)
{
if (j > 0)
str.append(",");
str.append("datacenter")
.append(j + 1)
.append(':')
.append(settings.artificialLatencies[i][j])
.append("ms");
}
cluster.get(i + 1).acceptOnInstance(latencies -> {
ArtificialLatency.setArtificialLatencies(latencies);
ArtificialLatency.setArtificialLatencyOnlyPermittedConsistencyLevels(false);
ArtificialLatency.setArtificialLatencyVerbs(ArtificialLatency.recommendedVerbs());
ArtificialLatency.setEnabled(true);
}, str.toString());
}
}
if (settings.traceSlowest > 0f)
{
float traceSlowest = settings.traceSlowest;
for (int i = 0 ; i < cluster.size() ; ++i)
{
cluster.get(i + 1).runOnInstance(() -> {
AccordTracing tracing = ((AccordAgent) AccordService.unsafeInstance().agent()).tracing();
tracing.setPattern(1, pattern -> pattern.withChance(traceSlowest)
.withKinds(TxnKindsAndDomains.parse("{K*}"))
.withTraceNew(CoordinationKinds.ALL),
SLOWEST, -1, 2, LEAKY, 10, 1, CoordinationKinds.ALL);
});
}
}
if (settings.traceLast > 0)
{
int traceLast = settings.traceLast;
for (int i = 0 ; i < cluster.size() ; ++i)
{
cluster.get(i + 1).runOnInstance(() -> {
AccordTracing tracing = ((AccordAgent) AccordService.unsafeInstance().agent()).tracing();
tracing.setPattern(2, pattern -> pattern.withKinds(TxnKindsAndDomains.parse("{KW}"))
.withTraceNew(CoordinationKinds.ALL),
RING, -1, traceLast, LEAKY, 10, 1, CoordinationKinds.ALL);
});
}
}
final AtomicBoolean stop = new AtomicBoolean();
final AtomicBoolean pauseOrStop = new AtomicBoolean();
final WaitQueue waitQueue = WaitQueue.newWaitQueue();
final Random random = new Random();
final Semaphore completed = new Semaphore(0);
final AtomicIntegerArray coordinatorIndexes = new AtomicIntegerArray(clientCount);
final Set<Integer> chaosCandidates = new ConcurrentSkipListSet<>();
for (int i = 1; i <= cluster.size() ; ++i)
chaosCandidates.add(i);
final List<java.util.concurrent.Future<?>> clients = new ArrayList<>();
final AtomicReferenceArray<RateLimiter> rateLimiters = new AtomicReferenceArray<>(clientCount);
final AtomicReference<EstimatedHistogram> readHistogram = new AtomicReference<>(new EstimatedHistogram(200));
final AtomicReference<EstimatedHistogram> writeHistogram = new AtomicReference<>(new EstimatedHistogram(200));
final List<String> chaosHistory = new ArrayList<>();
if (settings.clients >= cluster.size())
throw new IllegalArgumentException("Cannot have more clients than nodes");
if (settings.clusterChaosInterval < Integer.MAX_VALUE && settings.clients + 1 >= cluster.size())
throw new IllegalArgumentException("If restarting, cannot have as many clients as nodes, as must reroute client requests during restart");
int clientRatePerSecond = Math.min(settings.ratePerSecond, settings.minRatePerSecond) / clientCount;
for (int client = 0 ; client < clientCount ; ++client)
{
rateLimiters.set(client, RateLimiter.create(clientRatePerSecond));
final int clientIndex = client;
coordinatorIndexes.set(client, client + 1);
clients.add(clientExecutor.submit(() -> {
final Semaphore inFlight = new Semaphore(settings.clientConcurrency);
while (true)
{
while (pauseOrStop.get())
{
if (stop.get())
break;
WaitQueue.Signal signal = waitQueue.register();
if (pauseOrStop.get()) signal.awaitThrowUncheckedOnInterrupt();
else signal.cancel();
}
int coordinatorIdx = coordinatorIndexes.get(clientIndex);
ICoordinator coordinator = cluster.coordinator(coordinatorIdx);
try
{
rateLimiters.get(clientIndex).acquire();
inFlight.acquire();
long commandStart = System.nanoTime();
IntArrayList keys = new IntArrayList(settings.keysPerOperation, -1);
for (int i = 0 ; i < settings.keysPerOperation ; ++i)
{
int k = settings.keySelector.getAsInt();
if (!keys.containsInt(k))
keys.add(k);
}
if (!keys.intStream().allMatch(initialised::get))
{
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
completed.release();
if (fail == null)
{
long elapsed = System.nanoTime() - commandStart;
writeHistogram.get().add(NANOSECONDS.toMicros(elapsed));
synchronized (initialised)
{
keys.forEachInt(initialised::set);
}
}
else
{
logger.error("{}", fail.toString());
}
}, "UPDATE " + qualifiedAccordTableName + " SET v = 0 WHERE k IN ?", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, keys);
}
else if (random.nextFloat() < settings.readRatio)
{
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
completed.release();
if (fail == null)
readHistogram.get().add(NANOSECONDS.toMicros(System.nanoTime() - commandStart));
}, "BEGIN TRANSACTION\n" +
"SELECT * FROM " + qualifiedAccordTableName + " WHERE k IN ?;\n" +
"COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, keys
);
}
else
{
coordinator.executeWithResult((success, fail) -> {
inFlight.release();
completed.release();
if (fail == null)
{
long elapsed = System.nanoTime() - commandStart;
writeHistogram.get().add(NANOSECONDS.toMicros(elapsed));
}
else
logger.error("{}", fail.toString());
}, "BEGIN TRANSACTION\n" +
// "UPDATE " + qualifiedAccordTableName + " SET v = ? WHERE k = ?;\n" +
"UPDATE " + qualifiedAccordTableName + " SET v += ? WHERE k IN ?;\n" +
"COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(100), keys);
}
}
catch (RejectedExecutionException e)
{
inFlight.release();
}
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
}
}));
}
int targetClientRatePerSecond = settings.ratePerSecond / clientCount;
int nextRateLimitIncrease = settings.increaseRatePerSecondInterval;
long remainingTransactions = settings.totalTransactions;
int remainingClusterChaos = settings.totalClusterChaos;
while (true)
{
long batchStart = System.nanoTime();
int batchSize = 0;
if (completed.tryAcquire(settings.batchSize, settings.batchPeriodNanos, NANOSECONDS))
batchSize = settings.batchSize;
batchSize += completed.drainPermits();
if (clientRatePerSecond < targetClientRatePerSecond)
{
if ((nextRateLimitIncrease -= batchSize) <= 0)
{
clientRatePerSecond = Math.min(clientRatePerSecond * 2, targetClientRatePerSecond);
for (int i = 0 ; i < clientCount ; ++i)
rateLimiters.set(i, RateLimiter.create(clientRatePerSecond));
nextRateLimitIncrease = settings.increaseRatePerSecondInterval;
}
}
if ((nextRepairAt -= batchSize) <= 0)
{
nextRepairAt += settings.repairInterval;
System.out.println("repairing...");
cluster.coordinator(1).instance().nodetool("repair", qualifiedAccordTableName);
}
if ((nextCompactionAt -= batchSize) <= 0)
{
nextCompactionAt += settings.compactionInterval;
compactJournalCfs(cluster);
}
if ((nextJournalFlushAt -= batchSize) <= 0)
{
nextJournalFlushAt += settings.journalFlushInterval;
flushJournal(cluster);
}
if ((nextDataFlushAt -= batchSize) <= 0)
{
nextDataFlushAt += settings.dataFlushInterval;
flushData(cluster);
}
if ((nextCfkFlushAt -= batchSize) <= 0)
{
nextCfkFlushAt += settings.cfkFlushInterval;
flushCfk(cluster);
}
if ((nextChaosAt -= batchSize) <= 0)
{
Iterator<ChaosActive> iter = chaosActive.iterator();
while (iter.hasNext())
{
ChaosActive chaos = iter.next();
if (chaos.future.isDone())
{
chaos.future.get();
iter.remove();
Invariants.require(cluster.size() <= chaosActive.size() + chaosCandidates.size());
}
else
{
long elapsedNanos = Clock.Global.nanoTime() - chaos.startedAt;
if (elapsedNanos >= CHAOS_TIMEOUT_NANOS)
throw new AssertionError("Chaos " + chaos.kind + " has been running for " + NANOSECONDS.toSeconds(elapsedNanos) + "s with seed " + seed);
}
}
if (chaosActive.size() < settings.clusterChaosConcurrency)
{
if (remainingClusterChaos > 0)
{
--remainingClusterChaos;
nextChaosAt += settings.clusterChaosInterval;
ClusterChaos chaos = clusterChaos.get();
chaosActive.add(new ChaosActive(chaos, chaos(cluster, coordinatorIndexes, chaosCandidates, chaosExecutor, random, chaos, chaosHistory)));
}
else if (chaosActive.isEmpty() && (remainingTransactions <= 0 || !waitForTransactions))
{
break;
}
}
}
if ((remainingTransactions -= batchSize) <= 0 && (remainingClusterChaos <= 0 || !waitForClusterChaos))
break;
long nowMillis = System.currentTimeMillis();
EstimatedHistogram reads = readHistogram.getAndSet(new EstimatedHistogram(200));
EstimatedHistogram writes = writeHistogram.getAndSet(new EstimatedHistogram(200));
maybePrintSlowestTraces(cluster, settings);
maybePrintLastTraces(cluster, pauseOrStop, settings, waitQueue);
printInstanceMetrics(nowMillis, cluster);
printRates(nowMillis, batchStart, batchSize, reads, writes);
printVerbs(nowMillis, verbs);
}
}
catch (Throwable t)
{
t.printStackTrace();
System.exit(1);
}
logger.info("Workload completed successfully");
}
static void safeForEach(Cluster cluster, IIsolatedExecutor.SerializableRunnable run)
{
safeForEach(cluster, ignore -> run.run(), null);
}
static <P> void safeForEach(Cluster cluster, IIsolatedExecutor.SerializableConsumer<P> consumer, P param)
{
for (IInvokableInstance i : cluster)
{
try
{
if (!i.isShutdown())
{
i.acceptOnInstance(consumer, param);
}
}
catch (Throwable t)
{
logger.error("", t);
}
}
}
private void compactJournalCfs(Cluster cluster)
{
System.out.println("compacting journal cfs...");
for (IInvokableInstance i : cluster)
{
try { i.nodetool("compact", "system_accord.journal"); }
catch (Throwable t) { logger.error("", t); }
}
}
private void flushJournal(Cluster cluster)
{
System.out.println("flushing journal...");
safeForEach(cluster, () -> {
if (AccordService.started())
((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty();
});
}
private void flushData(Cluster cluster)
{
System.out.println("flushing data...");
safeForEach(cluster, name -> {
Schema.instance.getColumnFamilyStoreInstance(Schema.instance.getTableMetadata(KEYSPACE, name).id).forceFlush(UNIT_TESTS);
}, accordTableName);
}
private void flushCfk(Cluster cluster)
{
System.out.println("flushing cfk...");
safeForEach(cluster, () -> {
if (CommitLog.instance.isStarted())
AccordKeyspace.AccordColumnFamilyStores.commandsForKey.forceFlush(UNIT_TESTS);
});
}
private static class ChaosActive
{
final ClusterChaos kind;
final Future<?> future;
final long startedAt;
private ChaosActive(ClusterChaos kind, Future<?> future)
{
this.kind = kind;
this.future = future;
this.startedAt = Clock.Global.nanoTime();
}
}
private static Future<?> chaos(Cluster cluster, AtomicIntegerArray coordinatorIndexes, Set<Integer> candidates, ExecutorService chaosExecutor, Random random, ClusterChaos chaos, List<String> history)
{
List<Integer> snapshot = new ArrayList<>(candidates);
int nodeIdx;
{
int i = random.nextInt(snapshot.size());
Integer remove = snapshot.get(i);
candidates.remove(remove);
snapshot = new ArrayList<>(candidates);
Invariants.require(snapshot.size() > 0);
nodeIdx = remove;
}
for (int i = 0; i < coordinatorIndexes.length(); ++i)
{
if (nodeIdx == coordinatorIndexes.get(i))
{
int j = random.nextInt(snapshot.size());
int replaceIdx = snapshot.get(j);
coordinatorIndexes.set(j, replaceIdx);
}
}
String describe = String.format("%s node %d...", chaos, nodeIdx);
history.add(describe);
System.out.println("========= BEGIN CHAOS ==========");
System.out.println(describe);
System.out.println(candidates);
System.out.println("========= BEGIN CHAOS ==========");
switch (chaos)
{
default: throw UnhandledEnum.unknown(chaos);
case REBOOTSTRAP_INCOMPLETE:
case REBOOTSTRAP_RESET:
{
IInvokableInstance node = cluster.get(nodeIdx);
return node.asyncAcceptsOnInstance((Set<Integer> cnds) -> {
try
{
Node accordNode = AccordService.instance().node();
AccordService.getBlocking(accordNode.commandStores().rebootstrap(accordNode, chaos == REBOOTSTRAP_RESET ? LOG_CORRUPTED : LOG_INCOMPLETE));
}
finally
{
Invariants.require(cnds.add(nodeIdx));
System.out.println("========== END CHAOS ===========");
System.out.println(describe);
System.out.println("========== END CHAOS ===========");
}
}).apply(candidates);
}
case REBOOTSTRAP_IF_BEHIND:
{
IInvokableInstance node = cluster.get(nodeIdx);
return node.asyncAcceptsOnInstance((Set<Integer> cmds) -> {
try
{
AccordService.getBlocking(Catchup.rebootstrapIfBehind(AccordService.instance().node()));
}
finally
{
Invariants.require(cmds.add(nodeIdx));
System.out.println("========== END CHAOS ===========");
System.out.println(String.format("%s node %d...", chaos, nodeIdx));
System.out.println("========== END CHAOS ===========");
}
}).apply(candidates);
}
case RESTART:
case RESTART_AND_REBOOTSTRAP_INCOMPLETE:
case RESTART_AND_REBOOTSTRAP_RESET:
case RESTART_AND_REBOOTSTRAP_AFTER_TIMEOUT:
{
return chaosExecutor.submit(() -> {
IInvokableInstance node = cluster.get(nodeIdx);
try
{
node.shutdown().get();
switch (chaos)
{
case RESTART_AND_REBOOTSTRAP_AFTER_TIMEOUT:
node.config().set("accord.catchup_on_start_on_timeout", "REBOOTSTRAP");
node.config().set("accord.catchup_on_start_success_latency", "0s");
node.config().set("accord.catchup_on_start_fail_latency", "0s");
break;
case RESTART_AND_REBOOTSTRAP_INCOMPLETE:
node.config().set("accord.journal.replay", "REBOOTSTRAP_INCOMPLETE");
break;
case RESTART_AND_REBOOTSTRAP_RESET:
node.config().set("accord.journal.replay", "REBOOTSTRAP_RESET");
break;
}
node.startup();
return null;
}
catch (InterruptedException | ExecutionException e)
{
throw new RuntimeException(e);
}
finally
{
Invariants.require(candidates.add(nodeIdx));
node.config().set("accord.catchup_on_start_success_latency", "60s");
node.config().set("accord.catchup_on_start_fail_latency", "120s");
node.config().set("accord.catchup_on_start_on_timeout", "IGNORE");
node.config().set("accord.journal.replay", "PART_NON_DURABLE");
System.out.println("========== END CHAOS ===========");
System.out.println(String.format("%s node %d...", chaos, nodeIdx));
System.out.println("========== END CHAOS ===========");
}
});
}
}
}
private static void printRates(long nowMillis, long batchStartNanos, long batchSize, EstimatedHistogram reads, EstimatedHistogram writes)
{
System.out.println(String.format("%tT.%tL rate: %.2f/s (%d total)", nowMillis, nowMillis, (((float)batchSize * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStartNanos)), batchSize));
System.out.println(String.format("%tT.%tL reads : %d %d %d %d %d %d", nowMillis, nowMillis, reads.percentile(.25)/1000, reads.percentile(.5)/1000, reads.percentile(.95)/1000, reads.percentile(.99)/1000, reads.percentile(.999)/1000, reads.percentile(1)/1000));
System.out.println(String.format("%tT.%tL writes: %d %d %d %d %d %d", nowMillis, nowMillis, writes.percentile(.25)/1000, writes.percentile(.5)/1000, writes.percentile(.95)/1000, writes.percentile(.99)/1000, writes.percentile(.999)/1000, writes.percentile(1)/1000));
}
private static void printInstanceMetrics(long nowMillis, Cluster cluster)
{
safeForEach(cluster, () -> {
refresh(AccordExecutorMetrics.INSTANCE.elapsedRunning);
refresh(AccordExecutorMetrics.INSTANCE.elapsed);
System.out.println(String.format("%tT.%tL (%d %d %d %d %d %d)ms (%d %d %d %d %d %d)ms (%d %d %d %d %.0f, %d %d %d)us %d %d %d", nowMillis, nowMillis,
getLatency(AccordCoordinatorMetrics.readMetrics.preacceptLatency, 0.5),
getLatency(AccordCoordinatorMetrics.readMetrics.executeLatency, 0.5),
getLatency(AccordCoordinatorMetrics.readMetrics.applyLatency, 0.5),
getLatency(AccordCoordinatorMetrics.readMetrics.preacceptLatency, 0.999),
getLatency(AccordCoordinatorMetrics.readMetrics.executeLatency, 0.999),
getLatency(AccordCoordinatorMetrics.readMetrics.applyLatency, 0.999),
getLatency(AccordCoordinatorMetrics.writeMetrics.preacceptLatency, 0.95),
getLatency(AccordCoordinatorMetrics.writeMetrics.executeLatency, 0.5),
getLatency(AccordCoordinatorMetrics.writeMetrics.applyLatency, 0.5),
getLatency(AccordCoordinatorMetrics.writeMetrics.preacceptLatency, 0.999),
getLatency(AccordCoordinatorMetrics.writeMetrics.executeLatency, 0.999),
getLatency(AccordCoordinatorMetrics.writeMetrics.applyLatency, 0.999),
getLatency(AccordExecutorMetrics.INSTANCE.elapsedRunning, 0.5),
getLatency(AccordExecutorMetrics.INSTANCE.elapsedRunning, 0.9),
getLatency(AccordExecutorMetrics.INSTANCE.elapsedRunning, 1.0),
getCount(AccordExecutorMetrics.INSTANCE.elapsedRunning),
getTotal(AccordExecutorMetrics.INSTANCE.elapsedRunning),
getLatency(AccordExecutorMetrics.INSTANCE.elapsed, 0.5),
getLatency(AccordExecutorMetrics.INSTANCE.elapsed, 0.9),
getLatency(AccordExecutorMetrics.INSTANCE.elapsed, 0.999),
AccordExecutorMetrics.INSTANCE.running.getValue(),
AccordExecutorMetrics.INSTANCE.waitingToRun.getValue(),
AccordExecutorMetrics.INSTANCE.preparingToRun.getValue()
));
clear(AccordExecutorMetrics.INSTANCE.elapsedRunning);
clear(AccordExecutorMetrics.INSTANCE.elapsed);
});
}
private static void printVerbs(long nowMillis, Map<Verb, AtomicInteger> verbs)
{
class VerbCount
{
final Verb verb;
final int count;
VerbCount(Verb verb, int count)
{
this.verb = verb;
this.count = count;
}
}
List<VerbCount> verbCounts = new ArrayList<>();
for (Map.Entry<Verb, AtomicInteger> e : verbs.entrySet())
{
int count = e.getValue().getAndSet(0);
if (count != 0) verbCounts.add(new VerbCount(e.getKey(), count));
}
verbCounts.sort(Comparator.comparing(v -> -v.count));
StringBuilder verbSummary = new StringBuilder();
for (VerbCount vs : verbCounts)
{
{
if (verbSummary.length() > 0)
verbSummary.append(", ");
verbSummary.append(vs.verb);
verbSummary.append(": ");
verbSummary.append(vs.count);
}
}
System.out.println(String.format("%tT.%tL verbs: %s", nowMillis, nowMillis, verbSummary));
}
private static void maybePrintSlowestTraces(Cluster cluster, LoadSettings settings)
{
if (settings.traceSlowest > 0f)
{
safeForEach(cluster, () -> {
AccordTracing tracing = ((AccordAgent)AccordService.instance().agent()).tracing();
tracing.forEach(Functions.alwaysTrue(), (txnId, state) -> {
state.forEach(event -> {
if (event.elapsedNanos() < MILLISECONDS.toNanos(100))
return;
for (Message message : event.messages())
{
long multiplier = message.atNanos < event.doneAtNanos() ? 1 : -1;
System.out.println(String.format("%s %s %s %s %s %s", txnId, event.kind, multiplier * (message.atNanos - event.atNanos)/1000000, message.nodeId, message.commandStoreId, message.message));
}
});
});
tracing.eraseAll();
});
}
}
private static void maybePrintLastTraces(Cluster cluster, AtomicBoolean pauseOrStop, LoadSettings settings, WaitQueue waitQueue)
{
if (settings.traceLast > 0)
{
pauseOrStop.set(true);
Map<String, List<List<String>>> print = new HashMap<>();
for (int i = 1 ; i <= cluster.size() ; ++i)
{
cluster.get(i).acceptOnInstance(out -> {
AccordService service = (AccordService)AccordService.instance();
AccordTracing tracing = ((AccordAgent)AccordService.instance().agent()).tracing();
PriorityQueue<SortedByElapsed> candidates = new PriorityQueue<>(Comparator.comparingLong(c -> -c.elapsedMicros));
tracing.forEach(Functions.alwaysTrue(), (txnId, events) -> {
events.forEach(event -> {
if (event.kind == Client)
{
long doneAtMicros = event.doneAtMicros();
long elapsedMicros = doneAtMicros - event.txnId().hlc();
if (elapsedMicros > 350000 && elapsedMicros < 390000)
candidates.add(new SortedByElapsed(txnId, elapsedMicros));
}
});
});
AtomicInteger storeId = new AtomicInteger();
while (!candidates.isEmpty())
{
SortedByElapsed sortedCandidate = candidates.poll();
if (sortedCandidate.elapsedMicros < 300000)
return;
TxnId candidate = sortedCandidate.txnId;
storeId.lazySet(-1);
tracing.forEach(candidate, events -> {
events.forEach(event -> {
if (storeId.get() >= 0)
return;
for (Message message : event.messages())
{
if (message.nodeId < 0 && message.commandStoreId >= 0)
{
storeId.set(message.commandStoreId);
break;
}
}
});
});
if (storeId.get() >= 0)
{
CommandStore commandStore = service.node().commandStores().forId(storeId.get());
List<List<String>> result = AccordService.getBlocking(commandStore.submit(ExecutionContext.unsequenced(candidate, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(candidate);
PartialDeps deps = safeCommand.current().partialDeps();
if (deps == null)
return null;
List<List<String>> infos = new ArrayList<>();
for (TxnId txnId : deps.txnIds())
{
List<String> info = new ArrayList<>();
info.add(txnId.toString());
infos.add(info);
}
List<String> info = new ArrayList<>();
info.add(candidate.toString());
infos.add(info);
return infos;
}));
if (result != null)
{
for (List<String> info : result)
{
TxnId txnId = TxnId.parse(info.get(0));
AccordService.getBlocking(commandStore.execute(ExecutionContext.unsequenced(txnId, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
if (safeCommand.current().executeAt != null)
info.add(safeCommand.current().executeAt.toString());
}));
}
out.put(candidate.toString(), result);
return;
}
}
}
}, print);
}
for (int i = 1 ; i <= cluster.size() ; ++i)
{
cluster.get(i).acceptOnInstance(out -> {
AccordTracing tracing = ((AccordAgent)AccordService.instance().agent()).tracing();
for (Map.Entry<String, List<List<String>>> e : out.entrySet())
{
TxnId parentId = TxnId.parse(e.getKey());
for (List<String> infos : e.getValue())
{
TxnId depId = TxnId.parse(infos.get(0));
tracing.forEach(depId, events -> {
events.forEach(event -> {
infos.add(event.kind + ": [" + (event.idMicros - parentId.hlc()) + "..." + (event.doneAtMicros() - parentId.hlc()) + "][" + (event.idMicros - depId.hlc()) + "..." + (event.doneAtMicros() - depId.hlc()) + "]");
if (event.kind == Execute)
{
for (Message message : event.messages())
{
if (message.nodeId == parentId.node.id)
{
long atMicros = (event.idMicros + (message.atNanos - event.atNanos)/1000) - parentId.hlc();
infos.add(atMicros + ": " + message.message);
}
}
}
});
});
}
}
}, print);
}
for (Map.Entry<String, List<List<String>>> e : print.entrySet())
{
System.out.println("======" + e.getKey() + "======");
for (List<String> infos : e.getValue())
System.out.println(infos);
}
if (!print.isEmpty())
System.out.println();
pauseOrStop.set(false);
waitQueue.signalAll();
}
}
private static void refresh(Histogram histogram)
{
if (histogram instanceof ShardedHistogram)
((ShardedHistogram) histogram).refresh();
if (histogram instanceof ShardedDecayingHistogram)
((ShardedDecayingHistogram) histogram).refresh();
}
private static long getLatency(Histogram histogram, double percentile)
{
return (long)(histogram.getSnapshot().getValue(percentile) / 1000);
}
private static long getCount(Histogram histogram)
{
return histogram.getSnapshot().size();
}
private static double getTotal(Histogram histogram)
{
Snapshot snapshot = histogram.getSnapshot();
return (snapshot.getMean() * 0.0001d * snapshot.size());
}
private static void clear(Histogram histogram)
{
if (histogram instanceof ShardedHistogram)
((ShardedHistogram) histogram).clear();
if (histogram instanceof ShardedDecayingHistogram)
((ShardedDecayingHistogram) histogram).clear();
}
private static long getLatency(Timer timer, double percentile)
{
if (timer instanceof SnapshottingTimer)
return (long) (((SnapshottingTimer) timer).getPercentileSnapshot().getValue(percentile) / 1000);
return (long)(timer.getSnapshot().getValue(0.999) / 1000);
}
private static long getSize(Timer timer)
{
if (timer instanceof SnapshottingTimer)
return ((SnapshottingTimer) timer).getPercentileSnapshot().size();
return timer.getSnapshot().size();
}
@Override
protected Logger logger()
{
return logger;
}
static class SortedByElapsed
{
final TxnId txnId;
final long elapsedMicros;
SortedByElapsed(TxnId txnId, long elapsedMicros)
{
this.txnId = txnId;
this.elapsedMicros = elapsedMicros;
}
}
}

66
test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordRebootstrapLoadTest.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test.accord.load;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.api.Feature;
import static org.apache.cassandra.distributed.test.accord.load.LoadSettings.ycsbZipfian;
public class AccordRebootstrapLoadTest extends AccordLoadTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordRebootstrapLoadTest.class);
@Override
protected Logger logger()
{
return logger;
}
public void setupCluster(int nodeCount)
{
setupCluster(nodeCount, config -> {
config.with(Feature.NETWORK, Feature.GOSSIP)
.set("accord.shard_durability_target_splits", "8")
.set("accord.shard_durability_max_splits", "16")
.set("accord.shard_durability_cycle", "1m")
.set("accord.queue_submission_model", "SIGNAL")
.set("accord.command_store_shard_count", "8")
.set("accord.queue_thread_count", "4")
.set("accord.queue_shard_count", "1")
.set("accord.send_minimal", "false") // TODO (expected): only required because of misordering of preaccept, accept, commit if queued together
.set("accord.catchup_on_start_fail_latency", "2m");
});
}
@Test
public void testLoad() throws Exception
{
testLoad(new LoadSettings.Builder()
.setKeySelector(ycsbZipfian(100_000))
.setRatePerSecond(200)
.setClusterChaosInterval(10000)
.setClusterChaosConcurrency(2)
.setTotalClusterChaos(10)
.build());
}
}

137
test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordYcsbLoadTest.java Normal file
View File

@ -0,0 +1,137 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test.accord.load;
import java.util.Arrays;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.distributed.shared.DistributedTestBase;
import static org.apache.cassandra.distributed.test.accord.load.LoadSettings.ycsbZipfian;
public class AccordYcsbLoadTest extends AccordLoadTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordYcsbLoadTest.class);
private static final int[][] LATENCIES = new int[][] {
new int[] { 0, 44, 64, 43, 84 },
new int[] { 44, 0, 30, 3, 45 },
new int[] { 64, 30, 0, 28, 37 },
new int[] { 43, 3, 28, 0, 49 },
new int[] { 84, 45, 37, 49, 0 }
};
private static LoadSettings.Builder withArtificialLatencies(LoadSettings.Builder builder)
{
return builder.setArtificialLatencies(LATENCIES);
}
private static LoadSettings.Builder ycsbA(LoadSettings.Builder builder, int keyCount)
{
return builder.setKeySelector(ycsbZipfian(keyCount))
.setReadRatio(0.5f);
}
private static LoadSettings.Builder ycsbB(LoadSettings.Builder builder, int keyCount)
{
return builder.setKeySelector(ycsbZipfian(keyCount))
.setReadRatio(0.95f);
}
private static LoadSettings.Builder ycsbC(LoadSettings.Builder builder, int keyCount)
{
return builder.setKeySelector(ycsbZipfian(keyCount))
.setReadRatio(1.0f);
}
@Override
protected Logger logger()
{
return logger;
}
private static void computeWorstLatencies()
{
int[] qs = new int[LATENCIES.length];
for (int i = 0 ; i < qs.length ; ++i)
{
int[] copy = LATENCIES[i].clone();
Arrays.sort(copy);
qs[i] = copy[copy.length/2];
}
int[] ws = new int[qs.length];
for (int i = 0 ; i < qs.length ; ++i)
{
int iw = Integer.MIN_VALUE;
for (int j = 0; j < qs.length ; ++j)
iw = Math.max(iw, qs[i] + 3*qs[j] + LATENCIES[i][j]);
ws[i] = iw;
}
System.out.println(Arrays.toString(ws));
Arrays.fill(ws, 0);
for (int i = 0 ; i < qs.length ; ++i)
{
for (int j = 0 ; j < qs.length ; ++j)
{
if (j == i) continue;
if (qs[j] > 2*qs[i]) continue;
int w = qs[i] + 4*qs[j] + LATENCIES[i][j];
if (w > ws[i])
ws[i] = w;
}
}
System.out.println(Arrays.toString(ws));
}
@Ignore
@Test
public void testLoad() throws Exception
{
testLoad(ycsbA(new LoadSettings.Builder(), 100_000)
.setRatePerSecond(1600).setMinRatePerSecond(200)
.setIncreaseRatePerSecondInterval(5000)
.build());
}
public static void main(String[] args) throws Throwable
{
computeWorstLatencies();
DistributedTestBase.beforeClass();
AccordYcsbLoadTest test = new AccordYcsbLoadTest();
try
{
test.setupCluster();
test.setup();
test.testLoad(withArtificialLatencies(ycsbA(new LoadSettings.Builder(), 100_000)
.setRatePerSecond(1600).setMinRatePerSecond(200)
.setIncreaseRatePerSecondInterval(5000)
).build());
}
finally
{
test.tearDown();
}
}
}

338
test/distributed/org/apache/cassandra/distributed/test/accord/load/LoadSettings.java Normal file
View File

@ -0,0 +1,338 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test.accord.load;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import java.util.function.LongFunction;
import java.util.function.Supplier;
import org.apache.commons.math3.distribution.ZipfDistribution;
import org.apache.commons.math3.random.JDKRandomGenerator;
import accord.utils.DefaultRandom;
import static java.util.concurrent.TimeUnit.SECONDS;
class LoadSettings
{
enum ClusterChaos {
RESTART,
RESTART_AND_REBOOTSTRAP_INCOMPLETE, RESTART_AND_REBOOTSTRAP_RESET,
RESTART_AND_REBOOTSTRAP_AFTER_TIMEOUT,
REBOOTSTRAP_INCOMPLETE, REBOOTSTRAP_RESET,
REBOOTSTRAP_IF_BEHIND
}
final int repairInterval;
final int compactionInterval;
final int journalFlushInterval;
final int cfkFlushInterval;
final int cfkCompactionPeriodSeconds;
final int dataFlushInterval;
final int clusterChaosInterval;
final int clusterChaosDecay;
final int clusterChaosConcurrency;
final LongFunction<Supplier<ClusterChaos>> clusterChaos;
final int batchSize;
final long batchPeriodNanos;
final int clientConcurrency;
final int clients;
final int ratePerSecond;
final int minRatePerSecond;
final int increaseRatePerSecondInterval;
final int keysPerOperation;
final float readRatio;
final IntSupplier keySelector;
final boolean readBeforeWrite;
final float traceSlowest;
final int traceLast;
final long totalTransactions;
final int totalClusterChaos;
final int[][] artificialLatencies;
LoadSettings(Builder builder)
{
this.repairInterval = builder.repairInterval;
this.compactionInterval = builder.compactionInterval;
this.journalFlushInterval = builder.journalFlushInterval;
this.cfkFlushInterval = builder.cfkFlushInterval;
this.cfkCompactionPeriodSeconds = builder.cfkCompactionPeriodSeconds;
this.dataFlushInterval = builder.dataFlushInterval;
this.clusterChaosInterval = builder.clusterChaosInterval;
this.clusterChaosDecay = builder.clusterChaosDecay;
this.clusterChaosConcurrency = builder.clusterChaosConcurrency;
this.clusterChaos = builder.clusterChaos;
this.batchSize = builder.batchSize;
this.batchPeriodNanos = builder.batchPeriodNanos;
this.clientConcurrency = builder.clientConcurrency;
this.clients = builder.clients;
this.ratePerSecond = builder.ratePerSecond;
this.minRatePerSecond = builder.minRatePerSecond;
this.increaseRatePerSecondInterval = builder.increaseRatePerSecondInterval;
this.keysPerOperation = builder.keysPerOperation;
this.readRatio = builder.readRatio;
this.keySelector = builder.keySelector;
this.readBeforeWrite = builder.readBeforeWrite;
this.artificialLatencies = builder.artificialLatencies;
this.traceSlowest = builder.traceSlowest;
this.traceLast = builder.traceLast;
this.totalTransactions = builder.totalTransactions;
this.totalClusterChaos = builder.totalClusterChaos;
}
// interval is measured in terms of *operations* unless otherwise specified
public static class Builder
{
int repairInterval = Integer.MAX_VALUE;
int compactionInterval = Integer.MAX_VALUE;
int journalFlushInterval = Integer.MAX_VALUE;
int cfkFlushInterval = Integer.MAX_VALUE;
int cfkCompactionPeriodSeconds = 0;
int dataFlushInterval = Integer.MAX_VALUE;
int clusterChaosInterval = Integer.MAX_VALUE;
int clusterChaosDecay = 1;
int clusterChaosConcurrency = 1;
LongFunction<Supplier<ClusterChaos>> clusterChaos = seed -> new DefaultRandom(seed).randomWeightedPicker(LoadSettings.ClusterChaos.values());
int batchSize = 1000;
long batchPeriodNanos = SECONDS.toNanos(10);
int clientConcurrency = 50;
int clients = -1;
int ratePerSecond = 1000;
int minRatePerSecond = 50;
int increaseRatePerSecondInterval = 1000;
int keysPerOperation = 1;
float readRatio = 0.5f;
IntSupplier keySelector;
boolean readBeforeWrite;
float traceSlowest;
int traceLast;
int[][] artificialLatencies;
long totalTransactions = Long.MAX_VALUE;
int totalClusterChaos = Integer.MAX_VALUE;
public Builder setRepairInterval(int repairInterval)
{
this.repairInterval = repairInterval;
return this;
}
public Builder setCompactionInterval(int compactionInterval)
{
this.compactionInterval = compactionInterval;
return this;
}
public Builder setJournalFlushInterval(int journalFlushInterval)
{
this.journalFlushInterval = journalFlushInterval;
return this;
}
public Builder setCfkFlushInterval(int cfkFlushInterval)
{
this.cfkFlushInterval = cfkFlushInterval;
return this;
}
public Builder setCfkCompactionPeriodSeconds(int cfkCompactionPeriodSeconds)
{
this.cfkCompactionPeriodSeconds = cfkCompactionPeriodSeconds;
return this;
}
public Builder setDataFlushInterval(int dataFlushInterval)
{
this.dataFlushInterval = dataFlushInterval;
return this;
}
public Builder setClusterChaosInterval(int clusterChaosInterval)
{
this.clusterChaosInterval = clusterChaosInterval;
return this;
}
public Builder setClusterChaosDecay(int clusterChaosDecay)
{
this.clusterChaosDecay = clusterChaosDecay;
return this;
}
public Builder setClusterChaosConcurrency(int clusterChaosConcurrency)
{
this.clusterChaosConcurrency = clusterChaosConcurrency;
return this;
}
public Builder setClusterChaos(LongFunction<Supplier<ClusterChaos>> clusterChaos)
{
this.clusterChaos = clusterChaos;
return this;
}
public Builder setTotalTransactions(long totalTransactions)
{
this.totalTransactions = totalTransactions;
return this;
}
public Builder setTotalClusterChaos(int totalClusterChaos)
{
this.totalClusterChaos = totalClusterChaos;
return this;
}
public Builder setBatchSize(int batchSize)
{
this.batchSize = batchSize;
return this;
}
public Builder setBatchPeriodNanos(long batchPeriodNanos)
{
this.batchPeriodNanos = batchPeriodNanos;
return this;
}
public Builder setClientConcurrency(int clientConcurrency)
{
this.clientConcurrency = clientConcurrency;
return this;
}
public Builder setClients(int clients)
{
this.clients = clients;
return this;
}
public Builder setRatePerSecond(int ratePerSecond)
{
this.ratePerSecond = ratePerSecond;
return this;
}
public Builder setMinRatePerSecond(int minRatePerSecond)
{
this.minRatePerSecond = minRatePerSecond;
return this;
}
public Builder setIncreaseRatePerSecondInterval(int increaseRatePerSecondInterval)
{
this.increaseRatePerSecondInterval = increaseRatePerSecondInterval;
return this;
}
public Builder setKeysPerOperation(int keysPerOperation)
{
this.keysPerOperation = keysPerOperation;
return this;
}
public Builder setReadRatio(float readRatio)
{
this.readRatio = readRatio;
return this;
}
public Builder setReadBeforeWrite(boolean readBeforeWrite)
{
this.readBeforeWrite = readBeforeWrite;
return this;
}
public Builder setTraceSlowest(float traceSlowest)
{
this.traceSlowest = traceSlowest;
return this;
}
public Builder setTraceLast(int traceLast)
{
this.traceLast = traceLast;
return this;
}
public Builder setKeySelector(IntSupplier keySelector)
{
this.keySelector = keySelector;
return this;
}
public Builder setArtificialLatencies(int[][] artificialLatencies)
{
this.artificialLatencies = artificialLatencies;
return this;
}
public LoadSettings build()
{
return new LoadSettings(this);
}
}
static IntSupplier ycsbZipfian(int keyCount)
{
ZipfDistribution distribution = new ZipfDistribution(new JDKRandomGenerator(), keyCount, 0.99);
int count = distribution.inverseCumulativeProbability(0.65f);
float[] probs = new float[count];
for (int i = 0 ; i < probs.length ; ++i)
probs[i] = (float) distribution.cumulativeProbability(i);
// zipf is slow to compute, so we cache the first 65% of the distribution then use uniform probability; this is good enough for our purposes
float max = probs[probs.length - 1];
float inv_incr = probs.length >= keyCount ? 0f : 1f / ((1f-max)/(keyCount - probs.length));
Random random = new Random();
return () -> {
float v = random.nextFloat();
if (v < max)
{
int i = Arrays.binarySearch(probs, v);
if (i < 0) i = -1 - i;
return i;
}
else
{
return (int)((v - max)*inv_incr);
}
};
}
static IntSupplier roundrobin(int keyCount)
{
AtomicInteger next = new AtomicInteger();
return () -> {
int v = next.incrementAndGet();
if (v < keyCount)
return v;
return next.updateAndGet(i -> i > keyCount ? 0 : i + 1);
};
}
static IntSupplier uniform(int keyCount)
{
Random random = new Random();
return () -> random.nextInt(keyCount);
}
}

2
test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java
View File

@ -65,7 +65,7 @@ public class AccordRebootstrapTest extends FuzzTestBase
try (Cluster cluster = builder.withNodes(3)
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(100))
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(100, "dc0", "rack0"))
.withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP))
.withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP).set("accord.catchup_on_start_fail_latency", "60s"))
.start())
{
IInvokableInstance cmsInstance = cluster.get(1);

2
test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
View File

@ -84,7 +84,7 @@ public class AccordMessageSinkTest
TxnId id = nextTxnId(epoch, txn);
Ranges ranges = Ranges.of(IntKey.range(40, 50));
PartialTxn partialTxn = txn.slice(ranges, true);
Request request = new AccordFetchRequest(epoch, id, ranges, PartialDeps.NONE, partialTxn);
Request request = new AccordFetchRequest(epoch, id, ranges, partialTxn);
checkRequestReplies(request,
new AbstractFetchCoordinator.FetchResponse(null, null, id),

16
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -44,6 +44,7 @@ import accord.api.RemoteListeners.NoOpRemoteListeners;
import accord.api.Result;
import accord.api.Result.PersistableResult;
import accord.api.RoutingKey;
import accord.api.Scheduler;
import accord.api.Timeouts;
import accord.coordinate.Coordinations;
import accord.impl.DefaultLocalListeners;
@ -324,21 +325,11 @@ public class AccordTestUtils
{
NodeCommandStoreService time = new NodeCommandStoreService()
{
@Override
public AsyncExecutor someExecutor()
{
return null;
}
@Override
public ExclusiveAsyncExecutor someExclusiveExecutor()
{
return null;
}
private ToLongFunction<TimeUnit> elapsed = TimeService.elapsedWrapperFromNonMonotonicSource(TimeUnit.MICROSECONDS, this::now);
private long stamp = 0;
@Override public AsyncExecutor someExecutor() { return null; }
@Override public ExclusiveAsyncExecutor someExclusiveExecutor() { return null; }
@Override public Timeouts timeouts() { return null; }
@Override public DurableBefore durableBefore() { return DurableBefore.EMPTY; }
@Override public DurabilityService durability() { return null; }
@ -349,6 +340,7 @@ public class AccordTestUtils
@Override public long elapsed(TimeUnit timeUnit) { return elapsed.applyAsLong(timeUnit); }
@Override public TopologyManager topology() { throw new UnsupportedOperationException(); }
@Override public Coordinations coordinations() { return new Coordinations(); }
@Override public Scheduler scheduler() { return null; }
@Override public long currentStamp() { return stamp; }
@Override public void updateStamp() {++stamp;}
@Override public boolean isReplaying() { return false; }

7
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -42,6 +42,7 @@ import accord.api.ProgressLog;
import accord.api.RemoteListeners;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.api.Scheduler;
import accord.api.Timeouts;
import accord.coordinate.Coordinations;
import accord.impl.DefaultLocalListeners;
@ -268,6 +269,12 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
@Override public Coordinations coordinations() { return new Coordinations(); }
@Override
public Scheduler scheduler()
{
return null;
}
};
TestAgent.RethrowAgent agent = new TestAgent.RethrowAgent()

2
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -56,6 +56,7 @@ import accord.api.OwnershipEventListener;
import accord.api.ProgressLog;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.api.Scheduler;
import accord.api.Timeouts;
import accord.coordinate.Coordinations;
import accord.impl.AbstractReplayer;
@ -711,6 +712,7 @@ public class CommandsForKeySerializerTest
@Override public long now() { return 0; }
@Override public long elapsed(TimeUnit unit) { return 0; }
@Override public Coordinations coordinations() { return new Coordinations(); }
@Override public Scheduler scheduler() { return null; }
}; }
@Override public boolean visit(Unseekables<?> keysOrRanges, TxnId testTxnId, Kind.Kinds testKind, SupersedingCommandVisitor visit) { return false; }
@Override public <P1, P2> void visit(Unseekables<?> keysOrRanges, Timestamp startedBefore, Kind.Kinds testKind, ActiveCommandVisitor<P1, P2> visit, P1 p1, P2 p2) { }