diff --git a/CHANGES.txt b/CHANGES.txt index 1549165f6c..4f223e469a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,6 +18,7 @@ Merged from 4.1: * Memoize Cassandra verion and add a backoff interval for failed schema pulls (CASSANDRA-18902) * Fix StackOverflowError on ALTER after many previous schema changes (CASSANDRA-19166) Merged from 4.0: + * Add new concurrent_merkle_tree_requests config property to prevent OOM during multi-range and/or multi-table repairs (CASSANDRA-19336) * Skip version check if an endpoint is dead state in Gossiper#upgradeFromVersionSupplier (CASSANDRA-19187) * Fix Gossiper::hasMajorVersion3Nodes to return false during minor upgrade (CASSANDRA-18999) * Revert unnecessary read lock acquisition when reading ring version in TokenMetadata introduced in CASSANDRA-16286 (CASSANDRA-19107) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 0d2c711608..cc3f404d85 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -738,18 +738,42 @@ concurrent_materialized_view_writes: 32 # off heap objects memtable_allocation_type: heap_buffers -# Limit memory usage for Merkle tree calculations during repairs. The default -# is 1/16th of the available heap. The main tradeoff is that smaller trees -# have less resolution, which can lead to over-streaming data. If you see heap -# pressure during repairs, consider lowering this, but you cannot go below -# one mebibyte. If you see lots of over-streaming, consider raising -# this or using subrange repair. +# Limit memory usage for Merkle tree calculations during repairs of a certain +# table and common token range. Repair commands targetting multiple tables or +# virtual nodes can exceed this limit depending on concurrent_merkle_tree_requests. +# +# The default is 1/16th of the available heap. The main tradeoff is that +# smaller trees have less resolution, which can lead to over-streaming data. +# If you see heap pressure during repairs, consider lowering this, but you +# cannot go below one mebibyte. If you see lots of over-streaming, consider +# raising this or using subrange repair. # # For more details see https://issues.apache.org/jira/browse/CASSANDRA-14096. # # Min unit: MiB # repair_session_space: +# The number of simultaneous Merkle tree requests during repairs that can +# be performed by a repair command. The size of each validation request is +# limited by the repair_session_space property, so setting this to 1 will make +# sure that a repair command doesn't exceed that limit, even if the repair +# command is repairing multiple tables or multiple virtual nodes. +# +# There isn't a limit by default for backwards compatibility, but this can +# produce OOM for commands repairing multiple tables or multiple virtual nodes. +# A limit of just 1 simultaneous Merkle tree request is generally recommended +# with no virtual nodes so repair_session_space, and thereof the Merkle tree +# resolution, can be high. For virtual nodes a value of 1 with the default +# repair_session_space value will produce higher resolution Merkle trees +# at the expense of speed. Alternatively, when working with virtual nodes it +# can make sense to reduce the repair_session_space and increase the value of +# concurrent_merkle_tree_requests because each range will contain fewer data. +# +# For more details see https://issues.apache.org/jira/browse/CASSANDRA-19336. +# +# A zero value means no limit. +# concurrent_merkle_tree_requests: 0 + # repair: # # Configure the retries for each of the repair messages that support it. As of this moment retries use an exponential algorithm where each attempt sleeps longer based off the base_sleep_time and attempt. # retries: diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 57b52e7208..b027085b40 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -205,6 +205,8 @@ public class Config @Replaces(oldName = "repair_session_space_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true) public volatile DataStorageSpec.IntMebibytesBound repair_session_space = null; + public volatile int concurrent_merkle_tree_requests = 0; + public volatile boolean use_offheap_merkle_trees = true; public int storage_port = 7000; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index df917c888c..b50de6da69 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -3859,6 +3859,16 @@ public class DatabaseDescriptor conf.repair_session_space = new DataStorageSpec.IntMebibytesBound(sizeInMiB); } + public static int getConcurrentMerkleTreeRequests() + { + return conf.concurrent_merkle_tree_requests; + } + + public static void setConcurrentMerkleTreeRequests(int value) + { + conf.concurrent_merkle_tree_requests = value; + } + public static int getPaxosRepairParallelism() { return conf.paxos_repair_parallelism; diff --git a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java index c525597f5c..94cc3545c2 100644 --- a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java +++ b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java @@ -57,6 +57,7 @@ public abstract class AbstractRepairTask implements RepairTask private List submitRepairSessions(TimeUUID parentSession, boolean isIncremental, ExecutorPlus executor, + Scheduler validationScheduler, List commonRanges, String... cfnames) { @@ -76,6 +77,7 @@ public abstract class AbstractRepairTask implements RepairTask options.repairPaxos(), options.paxosOnly(), executor, + validationScheduler, cfnames); if (session == null) continue; @@ -88,10 +90,11 @@ public abstract class AbstractRepairTask implements RepairTask protected Future runRepair(TimeUUID parentSession, boolean isIncremental, ExecutorPlus executor, + Scheduler validationScheduler, List commonRanges, String... cfnames) { - List allSessions = submitRepairSessions(parentSession, isIncremental, executor, commonRanges, cfnames); + List allSessions = submitRepairSessions(parentSession, isIncremental, executor, validationScheduler, commonRanges, cfnames); List>> ranges = Lists.transform(allSessions, RepairSession::ranges); Future> f = FutureCombiner.successfulOf(allSessions); return f.map(results -> { diff --git a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java index 7f5e5630cc..347846cf6b 100644 --- a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java +++ b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java @@ -52,7 +52,7 @@ public class IncrementalRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) throws Exception + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) throws Exception { // the local node also needs to be included in the set of participants, since coordinator sessions aren't persisted Set allParticipants = ImmutableSet.builder() @@ -64,7 +64,7 @@ public class IncrementalRepairTask extends AbstractRepairTask CoordinatorSession coordinatorSession = coordinator.ctx.repair().consistent.coordinated.registerSession(parentSession, allParticipants, neighborsAndRanges.shouldExcludeDeadParticipants); - return coordinatorSession.execute(() -> runRepair(parentSession, true, executor, allRanges, cfnames)); + return coordinatorSession.execute(() -> runRepair(parentSession, true, executor, validationScheduler, allRanges, cfnames)); } } diff --git a/src/java/org/apache/cassandra/repair/NormalRepairTask.java b/src/java/org/apache/cassandra/repair/NormalRepairTask.java index bdc15cce53..e304280c58 100644 --- a/src/java/org/apache/cassandra/repair/NormalRepairTask.java +++ b/src/java/org/apache/cassandra/repair/NormalRepairTask.java @@ -47,8 +47,8 @@ public class NormalRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) { - return runRepair(parentSession, false, executor, commonRanges, cfnames); + return runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames); } } diff --git a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java index 5b861f71b3..edee11cf20 100644 --- a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java +++ b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java @@ -65,9 +65,9 @@ public class PreviewRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) { - Future f = runRepair(parentSession, false, executor, commonRanges, cfnames); + Future f = runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames); return f.map(result -> { if (result.hasFailed()) return result; diff --git a/src/java/org/apache/cassandra/repair/RepairCoordinator.java b/src/java/org/apache/cassandra/repair/RepairCoordinator.java index 3d1eab3e9e..27dd3a73b5 100644 --- a/src/java/org/apache/cassandra/repair/RepairCoordinator.java +++ b/src/java/org/apache/cassandra/repair/RepairCoordinator.java @@ -99,6 +99,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai private final List listeners = new ArrayList<>(); private final AtomicReference firstError = new AtomicReference<>(null); final SharedContext ctx; + final Scheduler validationScheduler; private TraceState traceState; @@ -116,6 +117,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai int cmd, RepairOption options, String keyspace) { this.ctx = ctx; + this.validationScheduler = Scheduler.build(DatabaseDescriptor.getConcurrentMerkleTreeRequests()); this.state = new CoordinatorState(ctx.clock(), cmd, keyspace, options); this.tag = "repair:" + cmd; this.validColumnFamilies = validColumnFamilies; @@ -453,7 +455,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai ExecutorPlus executor = createExecutor(); state.phase.repairSubmitted(); - return task.perform(executor) + return task.perform(executor, validationScheduler) // after adding the callback java could no longer infer the type... .>>map(r -> Pair.create(r, task::successMessage)) .addCallback((s, f) -> executor.shutdown()); diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java index 8fec3e9d54..c9966c5a05 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -20,11 +20,9 @@ package org.apache.cassandra.repair; import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; -import java.util.function.Predicate; import java.util.function.Function; +import java.util.function.Predicate; import java.util.stream.Collectors; - - import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; @@ -42,16 +40,16 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.asymmetric.DifferenceHolder; import org.apache.cassandra.repair.asymmetric.HostDifferences; import org.apache.cassandra.repair.asymmetric.PreferedNodeFilter; import org.apache.cassandra.repair.asymmetric.ReduceHelper; -import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.schema.SystemDistributedKeyspace; -import org.apache.cassandra.streaming.PreviewKind; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.paxos.cleanup.PaxosCleanup; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTrees; @@ -113,7 +111,7 @@ public class RepairJob extends AsyncFuture implements Runnable /** * Runs repair job. - * + *

* This sets up necessary task and runs them on given {@code taskExecutor}. * After submitting all tasks, waits until validation with replica completes. */ @@ -126,7 +124,6 @@ public class RepairJob extends AsyncFuture implements Runnable List allEndpoints = new ArrayList<>(session.state.commonRange.endpoints); allEndpoints.add(ctx.broadcastAddressAndPort()); - Future> treeResponses; Future paxosRepair; if (paxosRepairEnabled() && ((useV2() && session.repairPaxos) || session.paxosOnly)) { @@ -142,7 +139,7 @@ public class RepairJob extends AsyncFuture implements Runnable if (session.paxosOnly) { - paxosRepair.addCallback(new FutureCallback() + paxosRepair.addCallback(new FutureCallback<>() { public void onSuccess(Void v) { @@ -163,9 +160,9 @@ public class RepairJob extends AsyncFuture implements Runnable } // Create a snapshot at all nodes unless we're using pure parallel repairs + final Future allSnapshotTasks; if (parallelismDegree != RepairParallelism.PARALLEL) { - Future allSnapshotTasks; if (session.isIncremental) { // consistent repair does it's own "snapshotting" @@ -189,30 +186,19 @@ public class RepairJob extends AsyncFuture implements Runnable }); }); } - - // When all snapshot complete, send validation requests - treeResponses = allSnapshotTasks.flatMap(endpoints -> { - if (parallelismDegree == RepairParallelism.SEQUENTIAL) - return sendSequentialValidationRequest(allEndpoints); - else - return sendDCAwareValidationRequest(allEndpoints); - }, taskExecutor); } else { - // If not sequential, just send validation request to all replica - treeResponses = paxosRepair.flatMap(input -> sendValidationRequest(allEndpoints)); + allSnapshotTasks = null; } - treeResponses = treeResponses.map(a -> { - state.phase.validationCompleted(); - return a; - }); - // When all validations complete, submit sync tasks - Future> syncResults = treeResponses.flatMap(session.optimiseStreams && !session.pullRepair ? this::optimisedSyncing : this::standardSyncing, taskExecutor); + // Run validations and the creation of sync tasks in the scheduler, so it can limit the number of Merkle trees + // that there are in memory at once. When all validations complete, submit sync tasks out of the scheduler. + Future> syncResults = session.validationScheduler.schedule(() -> createSyncTasks(paxosRepair, allSnapshotTasks, allEndpoints), taskExecutor) + .flatMap(this::executeTasks, taskExecutor); // When all sync complete, set the final result - syncResults.addCallback(new FutureCallback>() + syncResults.addCallback(new FutureCallback<>() { @Override public void onSuccess(List stats) @@ -249,6 +235,35 @@ public class RepairJob extends AsyncFuture implements Runnable }, taskExecutor); } + private Future> createSyncTasks(Future paxosRepair, Future allSnapshotTasks, List allEndpoints) + { + Future> treeResponses; + if (allSnapshotTasks != null) + { + // When all snapshot complete, send validation requests + treeResponses = allSnapshotTasks.flatMap(endpoints -> { + if (parallelismDegree == RepairParallelism.SEQUENTIAL) + return sendSequentialValidationRequest(allEndpoints); + else + return sendDCAwareValidationRequest(allEndpoints); + }, taskExecutor); + } + else + { + // If not sequential, just send validation request to all replica + treeResponses = paxosRepair.flatMap(input -> sendValidationRequest(allEndpoints)); + } + + treeResponses = treeResponses.map(a -> { + state.phase.validationCompleted(); + return a; + }); + + return treeResponses.map(session.optimiseStreams && !session.pullRepair + ? this::createOptimisedSyncingSyncTasks + : this::createStandardSyncTasks, taskExecutor); + } + public synchronized void abort(@Nullable Throwable reason) { if (reason == null) @@ -265,18 +280,18 @@ public class RepairJob extends AsyncFuture implements Runnable return session.state.commonRange.transEndpoints.contains(ep); } - private Future> standardSyncing(List trees) + private List createStandardSyncTasks(List trees) { - List syncTasks = createStandardSyncTasks(ctx, desc, - trees, - ctx.broadcastAddressAndPort(), - this::isTransient, - session.isIncremental, - session.pullRepair, - session.previewKind); - return executeTasks(syncTasks); + return createStandardSyncTasks(ctx, desc, + trees, + ctx.broadcastAddressAndPort(), + this::isTransient, + session.isIncremental, + session.pullRepair, + session.previewKind); } + @VisibleForTesting static List createStandardSyncTasks(SharedContext ctx, RepairJobDesc desc, List trees, @@ -345,20 +360,6 @@ public class RepairJob extends AsyncFuture implements Runnable return syncTasks; } - private Future> optimisedSyncing(List trees) - { - List syncTasks = createOptimisedSyncingSyncTasks(ctx, - desc, - trees, - FBUtilities.getLocalAddressAndPort(), - this::isTransient, - this::getDC, - session.isIncremental, - session.previewKind); - - return executeTasks(syncTasks); - } - @VisibleForTesting Future> executeTasks(List tasks) { @@ -400,6 +401,19 @@ public class RepairJob extends AsyncFuture implements Runnable } } + private List createOptimisedSyncingSyncTasks(List trees) + { + return createOptimisedSyncingSyncTasks(ctx, + desc, + trees, + FBUtilities.getLocalAddressAndPort(), + this::isTransient, + this::getDC, + session.isIncremental, + session.previewKind); + } + + @VisibleForTesting static List createOptimisedSyncingSyncTasks(SharedContext ctx, RepairJobDesc desc, List trees, @@ -483,7 +497,7 @@ public class RepairJob extends AsyncFuture implements Runnable logger.info("{} {}", session.previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); long nowInSec = getNowInSeconds(); - List> tasks = new ArrayList<>(endpoints.size()); + List tasks = new ArrayList<>(endpoints.size()); for (InetAddressAndPort endpoint : endpoints) { ValidationTask task = newValidationTask(endpoint, nowInSec); @@ -518,7 +532,7 @@ public class RepairJob extends AsyncFuture implements Runnable final InetAddressAndPort nextAddress = requests.poll(); final ValidationTask nextTask = newValidationTask(nextAddress, nowInSec); tasks.add(nextTask); - currentTask.addCallback(new FutureCallback() + currentTask.addCallback(new FutureCallback<>() { public void onSuccess(TreeResponse result) { @@ -553,12 +567,7 @@ public class RepairJob extends AsyncFuture implements Runnable for (InetAddressAndPort endpoint : endpoints) { String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint); - Queue queue = requestsByDatacenter.get(dc); - if (queue == null) - { - queue = new LinkedList<>(); - requestsByDatacenter.put(dc, queue); - } + Queue queue = requestsByDatacenter.computeIfAbsent(dc, k -> new LinkedList<>()); queue.add(endpoint); } @@ -576,7 +585,7 @@ public class RepairJob extends AsyncFuture implements Runnable final InetAddressAndPort nextAddress = requests.poll(); final ValidationTask nextTask = newValidationTask(nextAddress, nowInSec); tasks.add(nextTask); - currentTask.addCallback(new FutureCallback() + currentTask.addCallback(new FutureCallback<>() { public void onSuccess(TreeResponse result) { diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index e0e104ff8e..7ec64502eb 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -133,6 +133,7 @@ public class RepairSession extends AsyncFuture implements I public final SafeExecutor taskExecutor; public final boolean optimiseStreams; public final SharedContext ctx; + public final Scheduler validationScheduler; private volatile List jobs = Collections.emptyList(); private volatile boolean terminated = false; @@ -149,6 +150,7 @@ public class RepairSession extends AsyncFuture implements I * @param cfnames names of columnfamilies */ public RepairSession(SharedContext ctx, + Scheduler validationScheduler, TimeUUID parentRepairSession, CommonRange commonRange, String keyspace, @@ -162,6 +164,7 @@ public class RepairSession extends AsyncFuture implements I String... cfnames) { this.ctx = ctx; + this.validationScheduler = validationScheduler; this.repairPaxos = repairPaxos; this.paxosOnly = paxosOnly; assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it"; diff --git a/src/java/org/apache/cassandra/repair/RepairTask.java b/src/java/org/apache/cassandra/repair/RepairTask.java index dc71d6e1b5..8e3c0bfb64 100644 --- a/src/java/org/apache/cassandra/repair/RepairTask.java +++ b/src/java/org/apache/cassandra/repair/RepairTask.java @@ -31,13 +31,13 @@ public interface RepairTask return name() + " completed successfully"; } - Future performUnsafe(ExecutorPlus executor) throws Exception; + Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) throws Exception; - default Future perform(ExecutorPlus executor) + default Future perform(ExecutorPlus executor, Scheduler validationScheduler) { try { - return performUnsafe(executor); + return performUnsafe(executor, validationScheduler); } catch (Exception | Error e) { diff --git a/src/java/org/apache/cassandra/repair/Scheduler.java b/src/java/org/apache/cassandra/repair/Scheduler.java new file mode 100644 index 0000000000..cc08a0666c --- /dev/null +++ b/src/java/org/apache/cassandra/repair/Scheduler.java @@ -0,0 +1,118 @@ +/* + * 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.repair; + +import java.util.LinkedList; +import java.util.Queue; +import java.util.concurrent.Executor; +import java.util.function.Supplier; +import javax.annotation.concurrent.GuardedBy; + +import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.concurrent.AsyncFuture; +import org.apache.cassandra.utils.concurrent.Future; + +/** + * Task scheduler that limits the number of concurrent tasks across multiple executors. + */ +public interface Scheduler +{ + default Future schedule(Supplier> task, Executor executor) + { + return schedule(new Task<>(task), executor); + } + + Task schedule(Task task, Executor executor); + + static Scheduler build(int concurrentValidations) + { + return concurrentValidations <= 0 + ? new NoopScheduler() + : new LimitedConcurrentScheduler(concurrentValidations); + } + + final class NoopScheduler implements Scheduler + { + @Override + public Task schedule(Task task, Executor executor) + { + executor.execute(task); + return task; + } + } + + final class LimitedConcurrentScheduler implements Scheduler + { + private final int concurrentValidations; + @GuardedBy("this") + private int inflight = 0; + @GuardedBy("this") + private final Queue, Executor>> tasks = new LinkedList<>(); + + LimitedConcurrentScheduler(int concurrentValidations) + { + this.concurrentValidations = concurrentValidations; + } + + @Override + public synchronized Task schedule(Task task, Executor executor) + { + tasks.offer(Pair.create(task, executor)); + maybeSchedule(); + return task; + } + + private synchronized void onDone() + { + inflight--; + maybeSchedule(); + } + + private void maybeSchedule() + { + if (inflight == concurrentValidations || tasks.isEmpty()) + return; + inflight++; + Pair, Executor> pair = tasks.poll(); + pair.left.addCallback((s, f) -> onDone()); + pair.right.execute(pair.left); + } + } + + class Task extends AsyncFuture implements Runnable + { + private final Supplier> supplier; + + public Task(Supplier> supplier) + { + this.supplier = supplier; + } + + @Override + public void run() + { + supplier.get().addCallback((s, f) -> { + if (f != null) + tryFailure(f); + else + trySuccess(s); + }); + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index 3c709950f0..7324b968a4 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -44,6 +44,7 @@ import com.google.common.collect.Multimap; import org.apache.cassandra.concurrent.ExecutorPlus; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DurationSpec; +import org.apache.cassandra.repair.Scheduler; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.locator.AbstractReplicationStrategy; @@ -351,6 +352,19 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai return DatabaseDescriptor.getRepairSessionSpaceInMiB(); } + @Override + public int getConcurrentMerkleTreeRequests() + { + return DatabaseDescriptor.getConcurrentMerkleTreeRequests(); + } + + @Override + public void setConcurrentMerkleTreeRequests(int value) + { + logger.info("Setting concurrent_merkle_tree_requests to {}", value); + DatabaseDescriptor.setConcurrentMerkleTreeRequests(value); + } + public List getRepairStats(List schemaArgs, String rangeString) { List stats = new ArrayList<>(); @@ -432,6 +446,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai boolean repairPaxos, boolean paxosOnly, ExecutorPlus executor, + Scheduler validationScheduler, String... cfnames) { if (repairPaxos && previewKind != PreviewKind.NONE) @@ -443,7 +458,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai if (cfnames.length == 0) return null; - final RepairSession session = new RepairSession(ctx, parentRepairSession, range, keyspace, + final RepairSession session = new RepairSession(ctx, validationScheduler, parentRepairSession, range, keyspace, parallelismDegree, isIncremental, pullRepair, previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames); repairs.getIfPresent(parentRepairSession).register(session.state); diff --git a/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java b/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java index 0a1444689e..851dc6c802 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java @@ -50,6 +50,10 @@ public interface ActiveRepairServiceMBean public void setRepairSessionSpaceInMiB(int sizeInMebibytes); public int getRepairSessionSpaceInMiB(); + int getConcurrentMerkleTreeRequests(); + + void setConcurrentMerkleTreeRequests(int value); + public boolean getUseOffheapMerkleTrees(); public void setUseOffheapMerkleTrees(boolean value); diff --git a/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java index 1febbdbff7..63a2f95f16 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java @@ -29,8 +29,6 @@ import java.util.Map; import java.util.Random; import java.util.Set; import java.util.concurrent.Callable; -import java.util.function.Function; -import java.util.function.Predicate; import java.util.concurrent.TimeoutException; import org.junit.Test; @@ -39,7 +37,6 @@ import net.bytebuddy.ByteBuddy; import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; import net.bytebuddy.implementation.MethodDelegation; import net.bytebuddy.implementation.bind.annotation.SuperCall; -import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.Cluster; @@ -49,12 +46,11 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.AsymmetricRemoteSyncTask; import org.apache.cassandra.repair.LocalSyncTask; import org.apache.cassandra.repair.RepairJob; -import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.SyncTask; import org.apache.cassandra.repair.TreeResponse; -import org.apache.cassandra.streaming.PreviewKind; import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; import static org.junit.Assert.assertEquals; @@ -109,20 +105,13 @@ public class OptimiseStreamsRepairTest extends TestBaseImpl public static void install(ClassLoader cl, int id) { new ByteBuddy().rebase(RepairJob.class) - .method(named("createOptimisedSyncingSyncTasks")) + .method(named("createOptimisedSyncingSyncTasks").and(takesArguments(1))) .intercept(MethodDelegation.to(BBHelper.class)) .make() .load(cl, ClassLoadingStrategy.Default.INJECTION); } - public static List createOptimisedSyncingSyncTasks(SharedContext ctx, - RepairJobDesc desc, - List trees, - InetAddressAndPort local, - Predicate isTransient, - Function getDC, - boolean isIncremental, - PreviewKind previewKind, + public static List createOptimisedSyncingSyncTasks(List trees, @SuperCall Callable> zuperCall) { List tasks = null; diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java new file mode 100644 index 0000000000..82a0b73371 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java @@ -0,0 +1,136 @@ +/* + * 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.repair; + +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.SuperCall; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.repair.ValidationTask; +import org.apache.cassandra.utils.MerkleTrees; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +/** + * Verifies that the config property {@code concurrent_merkle_tree_requests} limits the number of concurrent validation + * requests. + */ +public class ConcurrentValidationRequestsTest extends TestBaseImpl +{ + private static final int NODES = 4; + private static final int RF = 3; + private static final int TABLES = 5; + private static final int ROWS = 100; + private static final int MAX_REQUESTS = 1; + + @Test + public void testConcurrentValidations() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(NODES) + .withInstanceInitializer(ConcurrentValidationRequestsTest.BBHelper::install) + .withConfig(c -> c.set("concurrent_merkle_tree_requests", MAX_REQUESTS) + .with(NETWORK, GOSSIP)) + .start(), RF)) + { + for (int t = 1; t <= TABLES; t++) + cluster.schemaChange(withKeyspace("CREATE TABLE %s.t" + t + " (k int PRIMARY KEY, v int)")); + + int v = 0; + for (int k = 0; k < ROWS; k++) + { + v++; + for (int t = 1; t <= TABLES; t++) + { + String insert = withKeyspace("INSERT INTO %s.t" + t + " (k, v) VALUES (?, ?)"); + cluster.coordinator(1).execute(insert, ConsistencyLevel.ALL, k, v); + } + } + cluster.forEach(x -> x.flush(KEYSPACE)); + + NodeToolResult res = cluster.get(1).nodetoolResult("repair", "-j=4", KEYSPACE); + res.asserts().success(); + } + } + + public static class BBHelper + { + /** + * Keeps track of the number of concurrent validation requests, which should never be greater than RF times + * the value of the {@code concurrent_merkle_tree_requests} config property. + */ + public static volatile AtomicInteger requests = new AtomicInteger(0); + + public static void install(ClassLoader cl, int node) + { + if (node != 1) + return; + + new ByteBuddy().rebase(ValidationTask.class) + .method(named("run")) + .intercept(MethodDelegation.to(ConcurrentValidationRequestsTest.BBHelper.class)) + .method(named("treesReceived")) + .intercept(MethodDelegation.to(ConcurrentValidationRequestsTest.BBHelper.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings("unused") + public static void run(@SuperCall Callable zuper) + { + try + { + zuper.call(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + + int requests = BBHelper.requests.incrementAndGet(); + if (requests > MAX_REQUESTS * RF) + throw new AssertionError("Too many concurrent validation requests: " + requests); + } + + @SuppressWarnings("unused") + public static void treesReceived(MerkleTrees trees, @SuperCall Callable zuperCall) + { + requests.decrementAndGet(); + + try + { + zuperCall.call(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/repair/RepairJobTest.java b/test/unit/org/apache/cassandra/repair/RepairJobTest.java index 641926ed8a..165df1bbc9 100644 --- a/test/unit/org/apache/cassandra/repair/RepairJobTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairJobTest.java @@ -127,7 +127,8 @@ public class RepairJobTest PreviewKind previewKind, boolean optimiseStreams, boolean repairPaxos, boolean paxosOnly, String... cfnames) { - super(SharedContext.Global.instance, parentRepairSession, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair, + super(SharedContext.Global.instance, new Scheduler.NoopScheduler(), + parentRepairSession, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair, previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames); } @@ -333,7 +334,7 @@ public class RepairJobTest } catch (ExecutionException e) { - Assertions.assertThat(e.getCause()).isInstanceOf(RepairException.class); + Assertions.assertThat(e).hasRootCauseInstanceOf(RepairException.class); } // When the job fails, all three outstanding validation tasks should be aborted. diff --git a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java index 98baf46c57..f4d177870e 100644 --- a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java @@ -63,7 +63,7 @@ public class RepairSessionTest IPartitioner p = Murmur3Partitioner.instance; Range repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100))); Set endpoints = Sets.newHashSet(remote); - RepairSession session = new RepairSession(SharedContext.Global.instance, parentSessionId, + RepairSession session = new RepairSession(SharedContext.Global.instance, new Scheduler.NoopScheduler(), parentSessionId, new CommonRange(endpoints, Collections.emptySet(), Arrays.asList(repairRange)), "Keyspace1", RepairParallelism.SEQUENTIAL, false, false,