diff --git a/CHANGES.txt b/CHANGES.txt index 81efa76f79..69e58a2232 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0-alpha5 + * Prevent validation request submission from blocking ANTI_ENTROPY stage (CASSANDRA-15812) * Add fqltool and auditlogviewer to rpm and deb packages (CASSANDRA-14712) * Include DROPPED_COLUMNS in schema digest computation (CASSANDRA-15843) * Fix Cassandra restart from rpm install (CASSANDRA-15830) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 3a44a094da..79f643483a 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -814,8 +814,13 @@ column_index_cache_size_in_kb: 2 # to the number of cores. #concurrent_compactors: 1 -# Number of simultaneous repair validations to allow. Default is unbounded -# Values less than one are interpreted as unbounded (the default) +# Number of simultaneous repair validations to allow. If not set or set to +# a value less than 1, it defaults to the value of concurrent_compactors. +# To set a value greeater than concurrent_compactors at startup, the system +# property cassandra.allow_unlimited_concurrent_validations must be set to +# true. To dynamically resize to a value > concurrent_compactors on a running +# node, first call the bypassConcurrentValidatorsLimit method on the +# org.apache.cassandra.db:type=StorageService mbean # concurrent_validations: 0 # Number of simultaneous materialized view builder tasks to allow. diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 6d426563dd..34ac04919f 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -216,7 +216,6 @@ public class Config public volatile int compaction_large_partition_warning_threshold_mb = 100; public int min_free_space_per_drive_in_mb = 50; - public volatile int concurrent_validations = Integer.MAX_VALUE; public volatile int concurrent_materialized_view_builders = 1; /** @@ -416,6 +415,7 @@ public class Config public volatile boolean back_pressure_enabled = false; public volatile ParameterizedClass back_pressure_strategy; + public volatile int concurrent_validations; public RepairCommandPoolFullStrategy repair_command_pool_full_strategy = RepairCommandPoolFullStrategy.queue; public int repair_command_pool_size = concurrent_validations; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 5c753e0131..a36ac1eb62 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -152,6 +152,8 @@ public class DatabaseDescriptor // turns some warnings into exceptions for testing private static final boolean strictRuntimeChecks = Boolean.getBoolean("cassandra.strict.runtime.checks"); + public static volatile boolean allowUnlimitedConcurrentValidations = Boolean.getBoolean("cassandra.allow_unlimited_concurrent_validations"); + private static Function commitLogSegmentMgrProvider = c -> DatabaseDescriptor.isCDCEnabled() ? new CommitLogSegmentManagerCDC(c, DatabaseDescriptor.getCommitLogLocation()) : new CommitLogSegmentManagerStandard(c, DatabaseDescriptor.getCommitLogLocation()); @@ -668,12 +670,12 @@ public class DatabaseDescriptor if (conf.concurrent_compactors == null) conf.concurrent_compactors = Math.min(8, Math.max(2, Math.min(FBUtilities.getAvailableProcessors(), conf.data_file_directories.length))); - if (conf.concurrent_validations < 1) - conf.concurrent_validations = Integer.MAX_VALUE; - if (conf.concurrent_compactors <= 0) throw new ConfigurationException("concurrent_compactors should be strictly greater than 0, but was " + conf.concurrent_compactors, false); + applyConcurrentValidations(conf); + applyRepairCommandPoolSize(conf); + if (conf.concurrent_materialized_view_builders <= 0) throw new ConfigurationException("concurrent_materialized_view_builders should be strictly greater than 0, but was " + conf.concurrent_materialized_view_builders, false); @@ -841,6 +843,27 @@ public class DatabaseDescriptor validateMaxConcurrentAutoUpgradeTasksConf(conf.max_concurrent_automatic_sstable_upgrades); } + @VisibleForTesting + static void applyConcurrentValidations(Config config) + { + if (config.concurrent_validations < 1) + { + config.concurrent_validations = config.concurrent_compactors; + } + else if (config.concurrent_validations > config.concurrent_compactors && !allowUnlimitedConcurrentValidations) + { + throw new ConfigurationException("To set concurrent_validations > concurrent_compactors, " + + "set the system property cassandra.allow_unlimited_concurrent_validations=true"); + } + } + + @VisibleForTesting + static void applyRepairCommandPoolSize(Config config) + { + if (config.repair_command_pool_size < 1) + config.repair_command_pool_size = config.concurrent_validations; + } + private static String storagedirFor(String type) { return storagedir(type + "_directory") + File.separator + type; diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 2b5ffe6a69..7df56fc7d5 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -124,7 +124,7 @@ public class CompactionManager implements CompactionManagerMBean } private final CompactionExecutor executor = new CompactionExecutor(); - private final CompactionExecutor validationExecutor = new ValidationExecutor(); + private final ValidationExecutor validationExecutor = new ValidationExecutor(); private final CompactionExecutor cacheCleanupExecutor = new CacheCleanupExecutor(); private final CompactionExecutor viewBuildExecutor = new ViewBuildExecutor(); @@ -1897,9 +1897,32 @@ public class CompactionManager implements CompactionManagerMBean // TODO: pull out relevant parts of CompactionExecutor and move to ValidationManager public static class ValidationExecutor extends CompactionExecutor { + // CompactionExecutor, and by extension ValidationExecutor, use DebuggableThreadPoolExecutor's + // default RejectedExecutionHandler which blocks the submitting thread when the work queue is + // full. The calling thread in this case is AntiEntropyStage, so in most cases we don't actually + // want to block when the ValidationExecutor is saturated as this prevents progress on all + // repair tasks and may cause repair sessions to time out. Also, it can lead to references to + // heavyweight validation responses containing merkle trees being held for extended periods which + // increases GC pressure. Using LinkedBlockingQueue instead of the default SynchronousQueue allows + // tasks to be submitted without blocking the caller, but will always prefer queueing to creating + // new threads if the pool already has at least `corePoolSize` threads already running. For this + // reason we set corePoolSize to the maximum desired concurrency, but allow idle core threads to + // be terminated. + public ValidationExecutor() { - super(1, DatabaseDescriptor.getConcurrentValidations(), "ValidationExecutor", new SynchronousQueue()); + super(DatabaseDescriptor.getConcurrentValidations(), + DatabaseDescriptor.getConcurrentValidations(), + "ValidationExecutor", + new LinkedBlockingQueue()); + + allowCoreThreadTimeOut(true); + } + + public void adjustPoolSize() + { + setMaximumPoolSize(DatabaseDescriptor.getConcurrentValidations()); + setCorePoolSize(DatabaseDescriptor.getConcurrentValidations()); } } @@ -2021,10 +2044,9 @@ public class CompactionManager implements CompactionManagerMBean } } - public void setConcurrentValidations(int value) + public void setConcurrentValidations() { - value = value > 0 ? value : Integer.MAX_VALUE; - validationExecutor.setMaximumPoolSize(value); + validationExecutor.adjustPoolSize(); } public void setConcurrentViewBuilders(int value) diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index c679987532..d386d377a8 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -23,6 +23,7 @@ import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; @@ -126,24 +127,55 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai private final ConcurrentMap parentRepairSessions = new ConcurrentHashMap<>(); - public final static ExecutorService repairCommandExecutor; static { - Config.RepairCommandPoolFullStrategy strategy = DatabaseDescriptor.getRepairCommandPoolFullStrategy(); + RepairMetrics.init(); + } + + public static class RepairCommandExecutorHandle + { + private static final ThreadPoolExecutor repairCommandExecutor = + initializeExecutor(DatabaseDescriptor.getRepairCommandPoolSize(), + DatabaseDescriptor.getRepairCommandPoolFullStrategy()); + } + + @VisibleForTesting + static ThreadPoolExecutor initializeExecutor(int maxPoolSize, Config.RepairCommandPoolFullStrategy strategy) + { + int corePoolSize = 1; BlockingQueue queue; if (strategy == Config.RepairCommandPoolFullStrategy.reject) + { + // new threads will be created on demand up to max pool + // size so we can leave corePoolSize at 1 to start with queue = new SynchronousQueue<>(); + } else + { + // new threads are only created if > corePoolSize threads are running + // and the queue is full, so set corePoolSize to the desired max as the + // queue will _never_ be full. Idle core threads will eventually time + // out and may be re-created if/when subsequent tasks are submitted. + corePoolSize = maxPoolSize; queue = new LinkedBlockingQueue<>(); + } - repairCommandExecutor = new JMXEnabledThreadPoolExecutor(1, - DatabaseDescriptor.getRepairCommandPoolSize(), - 1, TimeUnit.HOURS, - queue, - new NamedThreadFactory("Repair-Task"), - "internal", - new ThreadPoolExecutor.AbortPolicy()); - RepairMetrics.init(); + ThreadPoolExecutor executor = new JMXEnabledThreadPoolExecutor(corePoolSize, + maxPoolSize, + 1, + TimeUnit.HOURS, + queue, + new NamedThreadFactory("Repair-Task"), + "internal", + new ThreadPoolExecutor.AbortPolicy()); + // allow idle core threads to be terminated + executor.allowCoreThreadTimeOut(true); + return executor; + } + + public static ThreadPoolExecutor repairCommandExecutor() + { + return RepairCommandExecutorHandle.repairCommandExecutor; } private final IFailureDetector failureDetector; @@ -461,10 +493,8 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai } try { - // Failed repair is expensive so we wait for longer time. - if (!prepareLatch.await(1, TimeUnit.HOURS)) { + if (!prepareLatch.await(DatabaseDescriptor.getRpcTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)) failRepair(parentRepairSession, "Did not get replies from all endpoints."); - } } catch (InterruptedException e) { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 0e4290448c..428dac1230 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1440,6 +1440,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE CompactionManager.instance.setConcurrentCompactors(value); } + public void bypassConcurrentValidatorsLimit() + { + logger.info("Enabling the ability to set concurrent validations to an unlimited value"); + DatabaseDescriptor.allowUnlimitedConcurrentValidations = true ; + } + + public void enforceConcurrentValidatorsLimit() + { + logger.info("Disabling the ability to set concurrent validations to an unlimited value"); + DatabaseDescriptor.allowUnlimitedConcurrentValidations = false ; + } + + public boolean isConcurrentValidatorsLimitEnforced() + { + return DatabaseDescriptor.allowUnlimitedConcurrentValidations; + } + public int getConcurrentValidators() { return DatabaseDescriptor.getConcurrentValidations(); @@ -1447,8 +1464,24 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public void setConcurrentValidators(int value) { + int concurrentCompactors = DatabaseDescriptor.getConcurrentCompactors(); + if (value > concurrentCompactors && !DatabaseDescriptor.allowUnlimitedConcurrentValidations) + throw new IllegalArgumentException( + String.format("Cannot set concurrent_validations greater than concurrent_compactors (%d)", + concurrentCompactors)); + + if (value <= 0) + { + logger.info("Using default value of concurrent_compactors ({}) for concurrent_validations", concurrentCompactors); + value = concurrentCompactors; + } + else + { + logger.info("Setting concurrent_validations to {}", value); + } + DatabaseDescriptor.setConcurrentValidations(value); - CompactionManager.instance.setConcurrentValidations(DatabaseDescriptor.getConcurrentValidations()); + CompactionManager.instance.setConcurrentValidations(); } public int getConcurrentViewBuilders() @@ -3733,7 +3766,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return Pair.create(0, Futures.immediateFuture(null)); int cmd = nextRepairCommand.incrementAndGet(); - return Pair.create(cmd, ActiveRepairService.repairCommandExecutor.submit(createRepairTask(cmd, keyspace, option, listeners))); + return Pair.create(cmd, ActiveRepairService.repairCommandExecutor().submit(createRepairTask(cmd, keyspace, option, listeners))); } /** diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index f8488c74a0..9664769167 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -581,6 +581,10 @@ public interface StorageServiceMBean extends NotificationEmitter public int getConcurrentCompactors(); public void setConcurrentCompactors(int value); + public void bypassConcurrentValidatorsLimit(); + public void enforceConcurrentValidatorsLimit(); + public boolean isConcurrentValidatorsLimitEnforced(); + public int getConcurrentValidators(); public void setConcurrentValidators(int value); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 908ccb23ff..c3e9982e3e 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -553,7 +553,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance () -> DiagnosticSnapshotService.instance.shutdownAndWait(1L, MINUTES), () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES), () -> SSTableReader.shutdownBlocking(1L, MINUTES), - () -> shutdownAndWait(Collections.singletonList(ActiveRepairService.repairCommandExecutor)), + () -> shutdownAndWait(Collections.singletonList(ActiveRepairService.repairCommandExecutor())), () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES) ); diff --git a/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorTimeout.java b/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorTimeout.java index 7a08187011..f523396adb 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorTimeout.java +++ b/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorTimeout.java @@ -42,19 +42,19 @@ public abstract class RepairCoordinatorTimeout extends RepairCoordinatorBase NodeToolResult result = repair(1, KEYSPACE, table); result.asserts() .failure() - .errorContains("Got negative replies from endpoints [127.0.0.2:7012]"); + .errorContains("Did not get replies from all endpoints."); if (withNotifications) { result.asserts() .notificationContains(NodeToolResult.ProgressEventType.START, "Starting repair command") .notificationContains(NodeToolResult.ProgressEventType.START, "repairing keyspace " + KEYSPACE + " with repair options") - .notificationContains(NodeToolResult.ProgressEventType.ERROR, "Got negative replies from endpoints [127.0.0.2:7012]") + .notificationContains(NodeToolResult.ProgressEventType.ERROR, "Did not get replies from all endpoints.") .notificationContains(NodeToolResult.ProgressEventType.COMPLETE, "finished with error"); } if (repairType != RepairType.PREVIEW) { - assertParentRepairFailedWithMessageContains(CLUSTER, KEYSPACE, table, "Got negative replies from endpoints [127.0.0.2:7012]"); + assertParentRepairFailedWithMessageContains(CLUSTER, KEYSPACE, table, "Did not get replies from all endpoints."); } else { diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java index 4a6a4526ac..2992a60ee5 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java @@ -26,9 +26,9 @@ import java.util.Arrays; import java.util.Collection; import java.util.Enumeration; + import com.google.common.base.Throwables; import org.junit.Assert; -import org.junit.Assume; import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; @@ -37,6 +37,7 @@ import org.apache.cassandra.OrderedJUnit4ClassRunner; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.exceptions.ConfigurationException; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -455,4 +456,50 @@ public class DatabaseDescriptorTest assertEquals(100, // total size is more than preferred so keep the configured limit DatabaseDescriptor.calculateDefaultSpaceInMB("type", "/path", "setting_name", preferredInMB, spaceInBytes, numerator, denominator)); } + + @Test + public void testConcurrentValidations() + { + Config conf = new Config(); + conf.concurrent_compactors = 8; + // if concurrent_validations is < 1 (including being unset) it should default to concurrent_compactors + assertThat(conf.concurrent_validations).isLessThan(1); + DatabaseDescriptor.applyConcurrentValidations(conf); + assertThat(conf.concurrent_validations).isEqualTo(conf.concurrent_compactors); + + // otherwise, it must be <= concurrent_compactors + conf.concurrent_validations = conf.concurrent_compactors + 1; + try + { + DatabaseDescriptor.applyConcurrentValidations(conf); + fail("Expected exception"); + } + catch (ConfigurationException e) + { + assertThat(e.getMessage()).isEqualTo("To set concurrent_validations > concurrent_compactors, " + + "set the system property cassandra.allow_unlimited_concurrent_validations=true"); + } + + // unless we disable that check (done with a system property at startup or via JMX) + DatabaseDescriptor.allowUnlimitedConcurrentValidations = true; + conf.concurrent_validations = conf.concurrent_compactors + 1; + DatabaseDescriptor.applyConcurrentValidations(conf); + assertThat(conf.concurrent_validations).isEqualTo(conf.concurrent_compactors + 1); + } + + @Test + public void testRepairCommandPoolSize() + { + Config conf = new Config(); + conf.concurrent_validations = 3; + // if repair_command_pool_size is < 1 (including being unset) it should default to concurrent_validations + assertThat(conf.repair_command_pool_size).isLessThan(1); + DatabaseDescriptor.applyRepairCommandPoolSize(conf); + assertThat(conf.repair_command_pool_size).isEqualTo(conf.concurrent_validations); + + // but it can be overridden + conf.repair_command_pool_size = conf.concurrent_validations + 1; + DatabaseDescriptor.applyRepairCommandPoolSize(conf); + assertThat(conf.repair_command_pool_size).isEqualTo(conf.concurrent_validations + 1); + } } diff --git a/test/unit/org/apache/cassandra/db/compaction/ValidationExecutorTest.java b/test/unit/org/apache/cassandra/db/compaction/ValidationExecutorTest.java new file mode 100644 index 0000000000..a175bdd5f5 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/compaction/ValidationExecutorTest.java @@ -0,0 +1,125 @@ +/* + * 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.db.compaction; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; + +import com.google.common.util.concurrent.Uninterruptibles; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.utils.concurrent.SimpleCondition; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; + +public class ValidationExecutorTest +{ + + CompactionManager.ValidationExecutor validationExecutor; + + @Before + public void setup() + { + DatabaseDescriptor.clientInitialization(); + // required for static initialization of CompactionManager + DatabaseDescriptor.setConcurrentCompactors(2); + DatabaseDescriptor.setConcurrentValidations(2); + + // shutdown the singleton CompactionManager to ensure MBeans are unregistered + CompactionManager.instance.forceShutdown(); + } + + @After + public void tearDown() + { + if (null != validationExecutor) + validationExecutor.shutdownNow(); + } + + @Test + public void testQueueOnValidationSubmission() throws InterruptedException + { + Condition taskBlocked = new SimpleCondition(); + AtomicInteger threadsAvailable = new AtomicInteger(DatabaseDescriptor.getConcurrentValidations()); + CountDownLatch taskComplete = new CountDownLatch(5); + validationExecutor = new CompactionManager.ValidationExecutor(); + + ExecutorService testExecutor = Executors.newSingleThreadExecutor(); + for (int i=0; i< 5; i++) + testExecutor.submit(() -> { + threadsAvailable.decrementAndGet(); + validationExecutor.submit(new Task(taskBlocked, taskComplete)); + }); + + // wait for all tasks to be submitted & check that the excess ones were queued + while (threadsAvailable.get() > 0) + TimeUnit.MILLISECONDS.sleep(10); + + assertEquals(2, validationExecutor.getActiveTaskCount()); + assertEquals(3, validationExecutor.getPendingTaskCount()); + + taskBlocked.signalAll(); + taskComplete.await(10, TimeUnit.SECONDS); + validationExecutor.shutdownNow(); + } + + @Test + public void testAdjustPoolSize() + { + // adjusting the pool size should dynamically set core and max pool + // size to DatabaseDescriptor::getConcurrentValidations + + validationExecutor = new CompactionManager.ValidationExecutor(); + + int corePoolSize = validationExecutor.getCorePoolSize(); + int maxPoolSize = validationExecutor.getMaximumPoolSize(); + + DatabaseDescriptor.setConcurrentValidations(corePoolSize * 2); + validationExecutor.adjustPoolSize(); + assertThat(validationExecutor.getCorePoolSize()).isEqualTo(corePoolSize * 2); + assertThat(validationExecutor.getMaximumPoolSize()).isEqualTo(maxPoolSize * 2); + validationExecutor.shutdownNow(); + } + + private static class Task implements Runnable + { + private final Condition blocked; + private final CountDownLatch complete; + + Task(Condition blocked, CountDownLatch complete) + { + this.blocked = blocked; + this.complete = complete; + } + + public void run() + { + Uninterruptibles.awaitUninterruptibly(blocked, 10, TimeUnit.SECONDS); + complete.countDown(); + } + } +} diff --git a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java index 4f7cde035a..8883f21502 100644 --- a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java +++ b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java @@ -19,15 +19,26 @@ package org.apache.cassandra.service; import java.util.*; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; import com.google.common.collect.ImmutableList; import com.google.common.collect.Sets; +import com.google.common.util.concurrent.Uninterruptibles; import org.junit.Assert; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; +import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; @@ -47,6 +58,7 @@ import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.Refs; +import org.apache.cassandra.utils.concurrent.SimpleCondition; import static org.apache.cassandra.repair.messages.RepairOption.DATACENTERS_KEY; import static org.apache.cassandra.repair.messages.RepairOption.FORCE_REPAIR_KEY; @@ -352,4 +364,98 @@ public class ActiveRepairServiceTest // full repair Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(false)), false)); } + + @Test + public void testRejectWhenPoolFullStrategy() throws InterruptedException + { + // Using RepairCommandPoolFullStrategy.reject, new threads are spawned up to + // repair_command_pool_size, at which point futher submissions are rejected + ExecutorService validationExecutor = ActiveRepairService.initializeExecutor(2, Config.RepairCommandPoolFullStrategy.reject); + try + { + Condition blocked = new SimpleCondition(); + CountDownLatch completed = new CountDownLatch(2); + validationExecutor.submit(new Task(blocked, completed)); + validationExecutor.submit(new Task(blocked, completed)); + try + { + validationExecutor.submit(new Task(blocked, completed)); + Assert.fail("Expected task submission to be rejected"); + } + catch (RejectedExecutionException e) + { + // expected + } + // allow executing tests to complete + blocked.signalAll(); + completed.await(10, TimeUnit.SECONDS); + // Submission is unblocked + validationExecutor.submit(() -> {}); + } + finally + { + // necessary to unregister mbean + validationExecutor.shutdownNow(); + } + } + + @Test + public void testQueueWhenPoolFullStrategy() throws InterruptedException + { + // Using RepairCommandPoolFullStrategy.queue, the pool is initialized to + // repair_command_pool_size and any tasks which cannot immediately be + // serviced are queued + ExecutorService validationExecutor = ActiveRepairService.initializeExecutor(2, Config.RepairCommandPoolFullStrategy.queue); + try + { + Condition allSubmitted = new SimpleCondition(); + Condition blocked = new SimpleCondition(); + CountDownLatch completed = new CountDownLatch(5); + ExecutorService testExecutor = Executors.newSingleThreadExecutor(); + for (int i = 0; i < 5; i++) + { + if (i < 4) + testExecutor.submit(() -> validationExecutor.submit(new Task(blocked, completed))); + else + testExecutor.submit(() -> { + validationExecutor.submit(new Task(blocked, completed)); + allSubmitted.signalAll(); + }); + } + + // Make sure all tasks have been submitted to the validation executor + allSubmitted.await(10, TimeUnit.SECONDS); + + // 2 threads actively processing tasks + Assert.assertEquals(2, ((DebuggableThreadPoolExecutor) validationExecutor).getActiveTaskCount()); + // 3 tasks queued + Assert.assertEquals(3, ((DebuggableThreadPoolExecutor) validationExecutor).getPendingTaskCount()); + // allow executing tests to complete + blocked.signalAll(); + completed.await(10, TimeUnit.SECONDS); + } + finally + { + // necessary to unregister mbean + validationExecutor.shutdownNow(); + } + } + + private static class Task implements Runnable + { + private final Condition blocked; + private final CountDownLatch complete; + + Task(Condition blocked, CountDownLatch complete) + { + this.blocked = blocked; + this.complete = complete; + } + + public void run() + { + Uninterruptibles.awaitUninterruptibly(blocked, 10, TimeUnit.SECONDS); + complete.countDown(); + } + } }