diff --git a/CHANGES.txt b/CHANGES.txt index 7c9d79ab98..a6428d3c25 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Use an ExecutorService for repair commands instead of new Thread(..).start() (CASSANDRA-13594) * Fix race / ref leak in anticompaction (CASSANDRA-13688) * Expose tasks queue length via JMX (CASSANDRA-12758) * Fix race / ref leak in PendingRepairManager (CASSANDRA-13751) diff --git a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java index a7a54f273b..2dafb4f16c 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java @@ -21,6 +21,7 @@ import java.lang.management.ManagementFactory; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.TimeUnit; import javax.management.MBeanServer; import javax.management.ObjectName; @@ -93,6 +94,19 @@ public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor i } } + public JMXEnabledThreadPoolExecutor(int corePoolSize, + int maxPoolSize, + long keepAliveTime, + TimeUnit unit, + BlockingQueue workQueue, + NamedThreadFactory threadFactory, + String jmxPath, + RejectedExecutionHandler rejectedExecutionHandler) + { + this(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, threadFactory, jmxPath); + setRejectedExecutionHandler(rejectedExecutionHandler); + } + public JMXEnabledThreadPoolExecutor(Stage stage) { this(stage.getJmxName(), stage.getJmxType()); diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 22f3551102..5a45282e1e 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -348,6 +348,9 @@ public class Config public volatile boolean back_pressure_enabled = false; public volatile ParameterizedClass back_pressure_strategy; + public RepairCommandPoolFullStrategy repair_command_pool_full_strategy = RepairCommandPoolFullStrategy.queue; + public int repair_command_pool_size = concurrent_validations; + /** * @deprecated migrate to {@link DatabaseDescriptor#isClientInitialized()} */ @@ -425,6 +428,12 @@ public class Config spinning } + public enum RepairCommandPoolFullStrategy + { + queue, + reject + } + private static final List SENSITIVE_KEYS = new ArrayList() {{ add("client_encryption_options"); add("server_encryption_options"); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 89c9432abf..fb50826029 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2328,4 +2328,14 @@ public class DatabaseDescriptor { conf.ideal_consistency_level = cl; } + + public static int getRepairCommandPoolSize() + { + return conf.repair_command_pool_size; + } + + public static Config.RepairCommandPoolFullStrategy getRepairCommandPoolFullStrategy() + { + return conf.repair_command_pool_full_strategy; + } } diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index d50dc3faf0..2e02f0ca7c 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -42,7 +42,10 @@ import com.google.common.util.concurrent.MoreExecutors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.dht.Bounds; @@ -77,6 +80,8 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.UUIDGen; +import static org.apache.cassandra.config.Config.RepairCommandPoolFullStrategy.queue; + /** * ActiveRepairService is the starting point for manual "active" repairs. * @@ -125,6 +130,25 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai private final ConcurrentMap parentRepairSessions = new ConcurrentHashMap<>(); + public final static ExecutorService repairCommandExecutor; + static + { + Config.RepairCommandPoolFullStrategy strategy = DatabaseDescriptor.getRepairCommandPoolFullStrategy(); + BlockingQueue queue; + if (strategy == Config.RepairCommandPoolFullStrategy.reject) + queue = new SynchronousQueue<>(); + else + queue = new LinkedBlockingQueue<>(); + + repairCommandExecutor = new JMXEnabledThreadPoolExecutor(1, + DatabaseDescriptor.getRepairCommandPoolSize(), + 1, TimeUnit.HOURS, + queue, + new NamedThreadFactory("Repair-Task"), + "internal", + new ThreadPoolExecutor.AbortPolicy()); + } + private final IFailureDetector failureDetector; private final Gossiper gossiper; private final Cache>> repairStatusByCmd; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 9070e8914c..5f691b4880 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -56,6 +56,7 @@ import org.apache.cassandra.auth.AuthSchemaChangeListener; import org.apache.cassandra.batchlog.BatchRemoveVerbHandler; import org.apache.cassandra.batchlog.BatchStoreVerbHandler; import org.apache.cassandra.batchlog.BatchlogManager; +import org.apache.cassandra.concurrent.ExecutorLocals; import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; @@ -95,6 +96,7 @@ import org.apache.cassandra.service.paxos.PrepareVerbHandler; import org.apache.cassandra.service.paxos.ProposeVerbHandler; import org.apache.cassandra.streaming.*; import org.apache.cassandra.tracing.TraceKeyspace; +import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.*; import org.apache.cassandra.utils.progress.ProgressEvent; @@ -120,8 +122,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE private final JMXProgressSupport progressSupport = new JMXProgressSupport(this); - private static final AtomicInteger threadCounter = new AtomicInteger(1); - private static int getRingDelay() { String newdelay = System.getProperty("cassandra.ring_delay_ms"); @@ -3304,7 +3304,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return 0; int cmd = nextRepairCommand.incrementAndGet(); - NamedThreadFactory.createThread(createRepairTask(cmd, keyspace, option), "Repair-Task-" + threadCounter.incrementAndGet()).start(); + ActiveRepairService.repairCommandExecutor.execute(createRepairTask(cmd, keyspace, option)); return cmd; } @@ -3360,6 +3360,21 @@ public class StorageService extends NotificationBroadcasterSupport implements IE RepairRunnable task = new RepairRunnable(this, cmd, options, keyspace); task.addProgressListener(progressSupport); + if (options.isTraced()) + { + Runnable r = () -> + { + try + { + task.run(); + } + finally + { + ExecutorLocals.set(null); + } + }; + return new FutureTask<>(r, null); + } return new FutureTask<>(task, null); }