diff --git a/.gitmodules b/.gitmodules index 4b1eea6020..c00814bf07 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,4 +1,4 @@ [submodule "modules/accord"] path = modules/accord url = https://github.com/belliottsmith/cassandra-accord.git - branch = executor-fair + branch = rerebootstrap diff --git a/doc/modules/cassandra/pages/managing/configuration/configuration.adoc b/doc/modules/cassandra/pages/managing/configuration/configuration.adoc index 9b4df248b2..0933da73fd 100644 --- a/doc/modules/cassandra/pages/managing/configuration/configuration.adoc +++ b/doc/modules/cassandra/pages/managing/configuration/configuration.adoc @@ -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. diff --git a/modules/accord b/modules/accord index 138e96dd0f..673cf2ff93 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 138e96dd0f0b9e315be21a8659f1aa81ec48637a +Subproject commit 673cf2ff93a4b32a392c91feacda09bb48ac16e7 diff --git a/src/java/org/apache/cassandra/concurrent/Stage.java b/src/java/org/apache/cassandra/concurrent/Stage.java index 557c1ca0a4..4e99729e0f 100644 --- a/src/java/org/apache/cassandra/concurrent/Stage.java +++ b/src/java/org/apache/cassandra/concurrent/Stage.java @@ -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), diff --git a/src/java/org/apache/cassandra/config/AccordConfig.java b/src/java/org/apache/cassandra/config/AccordConfig.java index 9813f3ffa0..50d00d2b6f 100644 --- a/src/java/org/apache/cassandra/config/AccordConfig.java +++ b/src/java/org/apache/cassandra/config/AccordConfig.java @@ -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 diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 5bc6de7ac1..765ee10d45 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 5b3e9f79e5..9b149de9f2 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -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) diff --git a/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java b/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java index 834abca49d..50a5393d10 100644 --- a/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java +++ b/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java @@ -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 stackTraceElementSerializer = new IVersionedSerializer() diff --git a/src/java/org/apache/cassandra/journal/Compactor.java b/src/java/org/apache/cassandra/journal/Compactor.java index ffc48ff5f2..428866098c 100644 --- a/src/java/org/apache/cassandra/journal/Compactor.java +++ b/src/java/org/apache/cassandra/journal/Compactor.java @@ -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 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 implements Runnable, Shutdownable logger.debug("Shutting down " + executor); if (scheduled != null) scheduled.cancel(false); + segmentCompactor.stop(); executor.shutdown(); } diff --git a/src/java/org/apache/cassandra/journal/SegmentCompactor.java b/src/java/org/apache/cassandra/journal/SegmentCompactor.java index 7b84ea82e1..01897a52c1 100644 --- a/src/java/org/apache/cassandra/journal/SegmentCompactor.java +++ b/src/java/org/apache/cassandra/journal/SegmentCompactor.java @@ -17,18 +17,20 @@ */ package org.apache.cassandra.journal; -import java.io.IOException; import java.util.Collection; public interface SegmentCompactor { SegmentCompactor NOOP = (SegmentCompactor) (segments) -> segments; + class Stopped extends RuntimeException {} + static SegmentCompactor noop() { //noinspection unchecked return (SegmentCompactor) NOOP; } - Collection> compact(Collection> segments) throws IOException; + Collection> compact(Collection> segments) throws Stopped; + default void stop() {} } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 2dbbc07e00..3fe7550439 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java index 0e757ba418..9e227e5f5a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java @@ -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 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 rangesById, FetchRanges callback) + @Override + public FetchResult sync(Node node, SafeCommandStore safeStore, Ranges ranges, TxnId atLeast, SortedArrayList 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> syncs = new ArrayList<>(); - for (Map.Entry 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) + { + @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; diff --git a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java index a14d9b3cd4..630f81ccbf 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordFetchCoordinator.java @@ -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 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 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)); + } + } diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index b0efd33b1b..a7bdedefdb 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -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); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java index 18dafb1908..2ddac1b52c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java +++ b/src/java/org/apache/cassandra/service/accord/AccordMessageSink.java @@ -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 message = Message.out(verb, request, expiresAtNanos); diff --git a/src/java/org/apache/cassandra/service/accord/AccordResult.java b/src/java/org/apache/cassandra/service/accord/AccordResult.java index 033ccdd88a..59440ee174 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordResult.java +++ b/src/java/org/apache/cassandra/service/accord/AccordResult.java @@ -201,7 +201,9 @@ public class AccordResult extends AsyncFuture implements BiConsumer 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 f; + Future f; { - AsyncChain submit; - if (mode == HARD) - { - if (!node.durability().isStarted()) - node.durability().start(); - submit = CatchupHard.catchup(node); - } - else submit = Catchup.catchup(node); + AsyncChain 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 sync(Object requestedBy, TxnId minBound, Ranges ranges, @Nullable Collection include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits) + public AsyncResult sync(Object requestedBy, TxnId minBound, Ranges ranges, @Nullable Collection include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits) { return node.durability().sync(requestedBy, ExclusiveSyncPoint, minBound, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits); } diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index abf7336b22..79bde24b57 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -88,8 +88,8 @@ public interface IAccordService IVerbHandler requestHandler(); IVerbHandler responseHandler(); - AsyncResult sync(Object requestedBy, @Nullable TxnId minBound, Ranges ranges, @Nullable Collection include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits); - AsyncChain sync(@Nullable TxnId minBound, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote); + AsyncResult sync(Object requestedBy, @Nullable TxnId minBound, Ranges ranges, @Nullable Collection include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits); + AsyncChain sync(@Nullable TxnId minBound, Keys keys, SyncLocal syncLocal, SyncRemote syncRemote); AsyncChain maxConflict(Ranges ranges); @Nonnull @@ -402,13 +402,13 @@ public interface IAccordService } @Override - public AsyncResult sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits) + public AsyncResult sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection include, SyncLocal syncLocal, SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits) { return delegate.sync(requestedBy, onOrAfter, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits); } @Override - public AsyncChain 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); } diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index fc3ffa975b..cabb3de079 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -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) diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordWaitStrategies.java b/src/java/org/apache/cassandra/service/accord/api/AccordWaitStrategies.java index cf7d8ff5ae..eb1fcbfbfa 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordWaitStrategies.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordWaitStrategies.java @@ -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)); diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java index 697a094e60..adaf531a5b 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java @@ -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) (step == Kind.Recovery ? recovery : standard); + return (CoordinationAdapter) (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 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); diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index 262e98af4e..874c2e305a 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -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 result; @@ -415,6 +416,11 @@ public class AccordInteropExecution implements ReadCoordinator }); } + private static Commit.Kind commitKind() + { + return sendMinimal() ? Kind.StableFastPath : Kind.StableWithTxnAndDeps; + } + private AsyncChain 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(); }); diff --git a/src/java/org/apache/cassandra/service/accord/journal/AbstractSegmentCompactor.java b/src/java/org/apache/cassandra/service/accord/journal/AbstractSegmentCompactor.java index f2caf35537..32f17625fd 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AbstractSegmentCompactor.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AbstractSegmentCompactor.java @@ -93,6 +93,7 @@ public abstract class AbstractSegmentCompactor implements SegmentCompactor> compact(Collection> segments) @@ -129,6 +130,9 @@ public abstract class AbstractSegmentCompactor implements SegmentCompactor 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 implements SegmentCompactor= (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; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java index 2acd8e2464..c413780546 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java @@ -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); } }; diff --git a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java index 07147fac17..1629b94fa3 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java +++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java @@ -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 Node node = service.node(); node.commandStores().forAllUnsafe(commandStore -> { - AsyncResult ready = commandStore.resumeBootstrap(node); + AsyncResult ready = commandStore.resumeBootstrap(node, GAIN_OWNERSHIP); bootstraps.add(AccordService.toFuture(ready.flatMap(e -> e.reads))); }); } diff --git a/test/conf/logback-info.xml b/test/conf/logback-info.xml new file mode 100644 index 0000000000..bc121ca649 --- /dev/null +++ b/test/conf/logback-info.xml @@ -0,0 +1,56 @@ + + + + + + + + + + + ./build/test/logs/${cassandra.testtag}/${suitename}/${cluster_id}/${instance_id}/system.log + + %-5level [%thread] ${instance_id} %date{"yyyy-MM-dd'T'HH:mm:ss,SSS", UTC} %F:%L - %msg%n + + + INFO + + true + + + + + %-5level %date{"HH:mm:ss,SSS"} %msg%n + + + INFO + + + + + + + + + + diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index f38b84ede2..03c16edd2a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -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 LIST_TYPE = ListType.getInstance(Int32Type.instance, true); - ExecutorService es = Executors.newCachedThreadPool(); - List> futures = new ArrayList<>(); - for (int ii = 0; ii < 10; ii++) + Future[] 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); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java index d3c1fadfc4..1802efa82a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java @@ -89,7 +89,7 @@ public class AccordIncrementalRepairTest extends TestBaseImpl } @Override - public AsyncResult sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits) + public AsyncResult sync(Object requestedBy, @Nullable TxnId onOrAfter, Ranges ranges, @Nullable Collection 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 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; diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java index 763cbd4d5a..c3ec06b6a7 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java @@ -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(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordLoadTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordLoadTestBase.java new file mode 100644 index 0000000000..fc1763a273 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordLoadTestBase.java @@ -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 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 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 = settings.clusterChaos.apply(seed); + + List 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 chaosCandidates = new ConcurrentSkipListSet<>(); + for (int i = 1; i <= cluster.size() ; ++i) + chaosCandidates.add(i); + final List> clients = new ArrayList<>(); + final AtomicReferenceArray rateLimiters = new AtomicReferenceArray<>(clientCount); + final AtomicReference readHistogram = new AtomicReference<>(new EstimatedHistogram(200)); + final AtomicReference writeHistogram = new AtomicReference<>(new EstimatedHistogram(200)); + final List 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 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

void safeForEach(Cluster cluster, IIsolatedExecutor.SerializableConsumer

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 candidates, ExecutorService chaosExecutor, Random random, ClusterChaos chaos, List history) + { + List 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 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 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 verbs) + { + class VerbCount + { + final Verb verb; + final int count; + + VerbCount(Verb verb, int count) + { + this.verb = verb; + this.count = count; + } + } + List verbCounts = new ArrayList<>(); + for (Map.Entry 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>> 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 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> 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> infos = new ArrayList<>(); + for (TxnId txnId : deps.txnIds()) + { + List info = new ArrayList<>(); + info.add(txnId.toString()); + infos.add(info); + } + List info = new ArrayList<>(); + info.add(candidate.toString()); + infos.add(info); + return infos; + })); + + if (result != null) + { + for (List 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>> e : out.entrySet()) + { + TxnId parentId = TxnId.parse(e.getKey()); + for (List 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>> e : print.entrySet()) + { + System.out.println("======" + e.getKey() + "======"); + for (List 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; + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordRebootstrapLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordRebootstrapLoadTest.java new file mode 100644 index 0000000000..4ddbf72fe0 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordRebootstrapLoadTest.java @@ -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()); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordYcsbLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordYcsbLoadTest.java new file mode 100644 index 0000000000..e72cd30aad --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/load/AccordYcsbLoadTest.java @@ -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(); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/load/LoadSettings.java b/test/distributed/org/apache/cassandra/distributed/test/accord/load/LoadSettings.java new file mode 100644 index 0000000000..004df82941 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/load/LoadSettings.java @@ -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> 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> 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> 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); + } +} diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java index 640ce44740..04121c55d4 100644 --- a/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java +++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java index b4af888878..8ab7e56bff 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java @@ -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), diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 1f8f963ccc..c5689e813a 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -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 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; } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 165115d8fd..ae5b6468e3 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -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() diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 448583ac00..3b975207b7 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -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 void visit(Unseekables keysOrRanges, Timestamp startedBefore, Kind.Kinds testKind, ActiveCommandVisitor visit, P1 p1, P2 p2) { }