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

Use an ExecutorService for repair commands instead of new Thread(..).start() 

Patch by marcuse; reviewed by Ariel Weisberg for CASSANDRA-13594
This commit is contained in:
Marcus Eriksson 2017-06-08 13:34:18 +02:00
parent e9cc805db1
commit 62d39f6544
6 changed files with 76 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

14
src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java
View File

@ -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<Runnable> 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());

9
src/java/org/apache/cassandra/config/Config.java
View File

@ -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<String> SENSITIVE_KEYS = new ArrayList<String>() {{
add("client_encryption_options");
add("server_encryption_options");

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

@ -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;
}
}

24
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -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<UUID, ParentRepairSession> parentRepairSessions = new ConcurrentHashMap<>();
public final static ExecutorService repairCommandExecutor;
static
{
Config.RepairCommandPoolFullStrategy strategy = DatabaseDescriptor.getRepairCommandPoolFullStrategy();
BlockingQueue<Runnable> 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<Integer, Pair<ParentRepairStatus, List<String>>> repairStatusByCmd;

21
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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);
}