You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bl...@apache.org on 2020/11/10 09:31:07 UTC
[cassandra] branch trunk updated (4dc1b5c -> 43929a5)
This is an automated email from the ASF dual-hosted git repository.
blerer pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from 4dc1b5c Merge branch 'cassandra-3.11' into trunk
add 3b970dd Remove the SEPExecutor blocking behavior
add 4e24afa Merge branch cassandra-3.0 into cassandra-3.11
new 43929a5 Merge branch cassandra-3.11 into trunk
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../apache/cassandra/concurrent/SEPExecutor.java | 41 +--
.../cassandra/concurrent/SharedExecutorPool.java | 8 +-
.../org/apache/cassandra/concurrent/Stage.java | 2 +-
.../org/apache/cassandra/transport/Message.java | 1 -
.../concurrent/LongSharedExecutorPoolTest.java | 8 +-
.../cassandra/concurrent/SEPExecutorTest.java | 4 +-
.../cassandra/metrics/ThreadPoolMetricsTest.java | 311 +++++++++++++++++++++
8 files changed, 329 insertions(+), 47 deletions(-)
create mode 100644 test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch cassandra-3.11 into trunk
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blerer pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 43929a5db51d766155185d63b318641ef640b397
Merge: 4dc1b5c 4e24afa
Author: Benjamin Lerer <b....@gmail.com>
AuthorDate: Tue Nov 10 10:08:35 2020 +0100
Merge branch cassandra-3.11 into trunk
CHANGES.txt | 1 +
.../apache/cassandra/concurrent/SEPExecutor.java | 41 +--
.../cassandra/concurrent/SharedExecutorPool.java | 8 +-
.../org/apache/cassandra/concurrent/Stage.java | 2 +-
.../org/apache/cassandra/transport/Message.java | 1 -
.../concurrent/LongSharedExecutorPoolTest.java | 8 +-
.../cassandra/concurrent/SEPExecutorTest.java | 4 +-
.../cassandra/metrics/ThreadPoolMetricsTest.java | 311 +++++++++++++++++++++
8 files changed, 329 insertions(+), 47 deletions(-)
diff --cc CHANGES.txt
index 1b08837,0cef27a..13bb0bf
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,42 -1,11 +1,43 @@@
-3.11.10
+4.0-beta4
+ * Add a ratelimiter to snapshot creation and deletion (CASSANDRA-13019)
+ * Produce consistent tombstone for reads to avoid digest mistmatch (CASSANDRA-15369)
+ * Fix SSTableloader issue when restoring a table named backups (CASSANDRA-16235)
+ * Invalid serialized size for responses caused by increasing message time by 1ms which caused extra bytes in size calculation (CASSANDRA-16103)
+ * Throw BufferOverflowException from DataOutputBuffer for better visibility (CASSANDRA-16214)
+ * TLS connections to the storage port on a node without server encryption configured causes java.io.IOException accessing missing keystore (CASSANDRA-16144)
+Merged from 3.11:
Merged from 3.0:
+ * Remove the SEPExecutor blocking behavior (CASSANDRA-16186)
- * Fix invalid cell value skipping when reading from disk (CASSANDRA-16223)
+ * Wait for schema agreement when bootstrapping (CASSANDRA-15158)
* Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146)
-3.11.9
- * Synchronize Keyspace instance store/clear (CASSANDRA-16210)
+4.0-beta3
+ * Segregate Network and Chunk Cache BufferPools and Recirculate Partially Freed Chunks (CASSANDRA-15229)
+ * Fail truncation requests when they fail on a replica (CASSANDRA-16208)
+ * Move compact storage validation earlier in startup process (CASSANDRA-16063)
+ * Fix ByteBufferAccessor cast exceptions are thrown when trying to query a virtual table (CASSANDRA-16155)
+ * Consolidate node liveness check for forced repair (CASSANDRA-16113)
+ * Use unsigned short in ValueAccessor.sliceWithShortLength (CASSANDRA-16147)
+ * Abort repairs when getting a truncation request (CASSANDRA-15854)
+ * Remove bad assert when getting active compactions for an sstable (CASSANDRA-15457)
+ * Avoid failing compactions with very large partitions (CASSANDRA-15164)
+ * Prevent NPE in StreamMessage in type lookup (CASSANDRA-16131)
+ * Avoid invalid state transition exception during incremental repair (CASSANDRA-16067)
+ * Allow zero padding in timestamp serialization (CASSANDRA-16105)
+ * Add byte array backed cells (CASSANDRA-15393)
+ * Correctly handle pending ranges with adjacent range movements (CASSANDRA-14801)
+ * Avoid adding locahost when streaming trivial ranges (CASSANDRA-16099)
+ * Add nodetool getfullquerylog (CASSANDRA-15988)
+ * Fix yaml format and alignment in tpstats (CASSANDRA-11402)
+ * Avoid trying to keep track of RTs for endpoints we won't write to during read repair (CASSANDRA-16084)
+ * When compaction gets interrupted, the exception should include the compactionId (CASSANDRA-15954)
+ * Make Table/Keyspace Metric Names Consistent With Each Other (CASSANDRA-15909)
+ * Mutating sstable component may race with entire-sstable-streaming(ZCS) causing checksum validation failure (CASSANDRA-15861)
+ * NPE thrown while updating speculative execution time if keyspace is removed during task execution (CASSANDRA-15949)
+ * Show the progress of data streaming and index build (CASSANDRA-15406)
+ * Add flag to disable chunk cache and disable by default (CASSANDRA-16036)
+ * Upgrade to snakeyaml >= 1.26 version for CVE-2017-18640 fix (CASSANDRA-16150)
+Merged from 3.11:
* Fix ColumnFilter to avoid querying cells of unselected complex columns (CASSANDRA-15977)
* Fix memory leak in CompressedChunkReader (CASSANDRA-15880)
* Don't attempt value skipping with mixed version cluster (CASSANDRA-15833)
diff --cc src/java/org/apache/cassandra/concurrent/SEPExecutor.java
index ee939a2,4641a19..58878b2
--- a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java
+++ b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java
@@@ -21,38 -21,26 +21,37 @@@ import java.util.ArrayList
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.cassandra.metrics.SEPMetrics;
++import com.google.common.annotations.VisibleForTesting;
++
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.metrics.ThreadPoolMetrics;
+import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
- import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
-public class SEPExecutor extends AbstractLocalAwareExecutorService
+public class SEPExecutor extends AbstractLocalAwareExecutorService implements SEPExecutorMBean
{
+ private static final Logger logger = LoggerFactory.getLogger(SEPExecutor.class);
private final SharedExecutorPool pool;
- public final int maxWorkers;
+ private final AtomicInteger maximumPoolSize;
+ private final MaximumPoolSizeListener maximumPoolSizeListener;
public final String name;
- private final SEPMetrics metrics;
+ private final String mbeanName;
- public final int maxTasksQueued;
- private final ThreadPoolMetrics metrics;
++ @VisibleForTesting
++ public final ThreadPoolMetrics metrics;
// stores both a set of work permits and task permits:
// bottom 32 bits are number of queued tasks, in the range [0..maxTasksQueued] (initially 0)
- // top 32 bits are number of work permits available in the range [0..maxWorkers] (initially maxWorkers)
+ // top 32 bits are number of work permits available in the range [-resizeDelta..maximumPoolSize] (initially maximumPoolSize)
private final AtomicLong permits = new AtomicLong();
- // producers wait on this when there is no room on the queue
- private final WaitQueue hasRoom = new WaitQueue();
private final AtomicLong completedTasks = new AtomicLong();
volatile boolean shuttingDown = false;
@@@ -61,17 -49,13 +60,16 @@@
// TODO: see if other queue implementations might improve throughput
protected final ConcurrentLinkedQueue<FutureTask<?>> tasks = new ConcurrentLinkedQueue<>();
- SEPExecutor(SharedExecutorPool pool, int maximumPoolSize, MaximumPoolSizeListener maximumPoolSizeListener, int maxTasksQueued, String jmxPath, String name)
- SEPExecutor(SharedExecutorPool pool, int maxWorkers, String jmxPath, String name)
++ SEPExecutor(SharedExecutorPool pool, int maximumPoolSize, MaximumPoolSizeListener maximumPoolSizeListener, String jmxPath, String name)
{
this.pool = pool;
this.name = name;
- this.maxWorkers = maxWorkers;
- this.permits.set(combine(0, maxWorkers));
- this.metrics = new SEPMetrics(this, jmxPath, name);
+ this.mbeanName = "org.apache.cassandra." + jmxPath + ":type=" + name;
+ this.maximumPoolSize = new AtomicInteger(maximumPoolSize);
+ this.maximumPoolSizeListener = maximumPoolSizeListener;
- this.maxTasksQueued = maxTasksQueued;
+ this.permits.set(combine(0, maximumPoolSize));
+ this.metrics = new ThreadPoolMetrics(this, jmxPath, name).register();
+ MBeanWrapper.instance.registerMBean(this, mbeanName);
}
protected void onCompletion()
@@@ -79,12 -63,6 +77,12 @@@
completedTasks.incrementAndGet();
}
+ @Override
+ public int getMaxTasksQueued()
+ {
- return maxTasksQueued;
++ return Integer.MAX_VALUE;
+ }
+
// schedules another worker for this pool if there is work outstanding and there are no spinning threads that
// will self-assign to it in the immediate future
boolean maybeSchedule()
@@@ -121,69 -99,21 +119,44 @@@
// worker, we simply start a worker in a spinning state
pool.maybeStartSpinningWorker();
}
- else if (taskPermits >= maxTasksQueued)
- {
- // register to receive a signal once a task is processed bringing the queue below its threshold
- WaitQueue.Signal s = hasRoom.register();
-
- // we will only be signalled once the queue drops below full, so this creates equivalent external behaviour
- // however the advantage is that we never wake-up spuriously;
- // we choose to always sleep, even if in the intervening time the queue has dropped below limit,
- // so long as we _will_ eventually receive a signal
- if (taskPermits(permits.get()) > maxTasksQueued)
- {
- // if we're blocking, we might as well directly schedule a worker if we aren't already at max
- if (takeWorkPermit(true))
- pool.schedule(new Work(this));
-
- metrics.totalBlocked.inc();
- metrics.currentBlocked.inc();
- s.awaitUninterruptibly();
- metrics.currentBlocked.dec();
- }
- else // don't propagate our signal when we cancel, just cancel
- s.cancel();
- }
}
+ public enum TakeTaskPermitResult
+ {
+ NONE_AVAILABLE, // No task permits available
+ TOOK_PERMIT, // Took a permit and reduced task permits
+ RETURNED_WORK_PERMIT // Detected pool shrinking and returned work permit ahead of SEPWorker exit.
+ };
+
// takes permission to perform a task, if any are available; once taken it is guaranteed
// that a proceeding call to tasks.poll() will return some work
- boolean takeTaskPermit()
+ TakeTaskPermitResult takeTaskPermit(boolean checkForWorkPermitOvercommit)
{
+ TakeTaskPermitResult result;
while (true)
{
long current = permits.get();
+ long updated;
+ int workPermits = workPermits(current);
int taskPermits = taskPermits(current);
- if (taskPermits == 0)
- return false;
- if (permits.compareAndSet(current, updateTaskPermits(current, taskPermits - 1)))
+ if (workPermits < 0 && checkForWorkPermitOvercommit)
{
- return true;
+ // Work permits are negative when the pool is reducing in size. Atomically
+ // adjust the number of work permits so there is no race of multiple SEPWorkers
+ // exiting. On conflicting update, recheck.
+ result = TakeTaskPermitResult.RETURNED_WORK_PERMIT;
+ updated = updateWorkPermits(current, workPermits + 1);
+ }
+ else
+ {
+ if (taskPermits == 0)
+ return TakeTaskPermitResult.NONE_AVAILABLE;
+ result = TakeTaskPermitResult.TOOK_PERMIT;
+ updated = updateTaskPermits(current, taskPermits - 1);
+ }
+ if (permits.compareAndSet(current, updated))
+ {
- if (taskPermits == maxTasksQueued && hasRoom.hasWaiters())
- hasRoom.signalAll();
+ return result;
}
}
}
diff --cc src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
index 28a994c,b94090b..7a07cf4
--- a/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
+++ b/src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
@@@ -108,14 -108,9 +108,14 @@@ public class SharedExecutorPoo
schedule(Work.SPINNING);
}
- public synchronized LocalAwareExecutorService newExecutor(int maxConcurrency, int maxQueuedTasks, String jmxPath, String name)
+ public synchronized LocalAwareExecutorService newExecutor(int maxConcurrency, String jmxPath, String name)
{
- return newExecutor(maxConcurrency, i -> {}, maxQueuedTasks, jmxPath, name);
- SEPExecutor executor = new SEPExecutor(this, maxConcurrency, jmxPath, name);
++ return newExecutor(maxConcurrency, i -> {}, jmxPath, name);
+ }
+
- public LocalAwareExecutorService newExecutor(int maxConcurrency, LocalAwareExecutorService.MaximumPoolSizeListener maximumPoolSizeListener, int maxQueuedTasks, String jmxPath, String name)
++ public LocalAwareExecutorService newExecutor(int maxConcurrency, LocalAwareExecutorService.MaximumPoolSizeListener maximumPoolSizeListener, String jmxPath, String name)
+ {
- SEPExecutor executor = new SEPExecutor(this, maxConcurrency, maximumPoolSizeListener, maxQueuedTasks, jmxPath, name);
++ SEPExecutor executor = new SEPExecutor(this, maxConcurrency, maximumPoolSizeListener, jmxPath, name);
executors.add(executor);
return executor;
}
diff --cc src/java/org/apache/cassandra/concurrent/Stage.java
index d2ebe8d,ccb1565..d4b214c
--- a/src/java/org/apache/cassandra/concurrent/Stage.java
+++ b/src/java/org/apache/cassandra/concurrent/Stage.java
@@@ -49,230 -24,60 +49,230 @@@ import static java.util.stream.Collecto
public enum Stage
{
- READ,
- MUTATION,
- COUNTER_MUTATION,
- VIEW_MUTATION,
- GOSSIP,
- REQUEST_RESPONSE,
- ANTI_ENTROPY,
- MIGRATION,
- MISC,
- TRACING,
- INTERNAL_RESPONSE,
- READ_REPAIR;
-
- public static Iterable<Stage> jmxEnabledStages()
- {
- return Iterables.filter(Arrays.asList(values()), new Predicate<Stage>()
+ READ ("ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage),
+ MUTATION ("MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage),
+ COUNTER_MUTATION ("CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage),
+ VIEW_MUTATION ("ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage),
+ GOSSIP ("GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage),
+ REQUEST_RESPONSE ("RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage),
+ ANTI_ENTROPY ("AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage),
+ MIGRATION ("MigrationStage", "internal", () -> 1, null, Stage::singleThreadedStage),
+ MISC ("MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage),
+ TRACING ("TracingStage", "internal", () -> 1, null, Stage::tracingExecutor),
+ INTERNAL_RESPONSE ("InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage),
+ IMMEDIATE ("ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor);
+
+
+ public static final long KEEP_ALIVE_SECONDS = 60; // seconds to keep "extra" threads alive for when idle
+ public final String jmxName;
+ private final Supplier<LocalAwareExecutorService> initialiser;
+ private volatile LocalAwareExecutorService executor = null;
+
+ Stage(String jmxName, String jmxType, IntSupplier numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize, ExecutorServiceInitialiser initialiser)
+ {
+ this.jmxName = jmxName;
+ this.initialiser = () -> initialiser.init(jmxName, jmxType, numThreads.getAsInt(), onSetMaximumPoolSize);
+ }
+
+ private static String normalizeName(String stageName)
+ {
+ // Handle discrepancy between JMX names and actual pool names
+ String upperStageName = stageName.toUpperCase();
+ if (upperStageName.endsWith("STAGE"))
+ {
+ upperStageName = upperStageName.substring(0, stageName.length() - 5);
+ }
+ return upperStageName;
+ }
+
+ private static final Map<String,Stage> nameMap = Arrays.stream(values())
+ .collect(toMap(s -> Stage.normalizeName(s.jmxName),
+ s -> s));
+
+ public static Stage fromPoolName(String stageName)
+ {
+ String upperStageName = normalizeName(stageName);
+
+ Stage result = nameMap.get(upperStageName);
+ if (result != null)
+ return result;
+
+ try
+ {
+ return valueOf(upperStageName);
+ }
+ catch (IllegalArgumentException e)
{
- public boolean apply(Stage stage)
+ switch(upperStageName) // Handle discrepancy between configuration file and stage names
{
- return stage != TRACING;
+ case "CONCURRENT_READS":
+ return READ;
+ case "CONCURRENT_WRITERS":
+ return MUTATION;
+ case "CONCURRENT_COUNTER_WRITES":
+ return COUNTER_MUTATION;
+ case "CONCURRENT_MATERIALIZED_VIEW_WRITES":
+ return VIEW_MUTATION;
+ default:
+ throw new IllegalStateException("Must be one of " + Arrays.stream(values())
+ .map(Enum::toString)
+ .collect(Collectors.joining(",")));
}
- });
+ }
}
- public String getJmxType()
+ // Convenience functions to execute on this stage
+ public void execute(Runnable command) { executor().execute(command); }
+ public void execute(Runnable command, ExecutorLocals locals) { executor().execute(command, locals); }
+ public void maybeExecuteImmediately(Runnable command) { executor().maybeExecuteImmediately(command); };
+ public <T> Future<T> submit(Callable<T> task) { return executor().submit(task); }
+ public Future<?> submit(Runnable task) { return executor().submit(task); }
+ public <T> Future<T> submit(Runnable task, T result) { return executor().submit(task, result); }
+
+ public LocalAwareExecutorService executor()
{
- switch (this)
+ if (executor == null)
{
- case ANTI_ENTROPY:
- case GOSSIP:
- case MIGRATION:
- case MISC:
- case TRACING:
- case INTERNAL_RESPONSE:
- return "internal";
- case MUTATION:
- case COUNTER_MUTATION:
- case VIEW_MUTATION:
- case READ:
- case REQUEST_RESPONSE:
- case READ_REPAIR:
- return "request";
- default:
- throw new AssertionError("Unknown stage " + this);
+ synchronized (this)
+ {
+ if (executor == null)
+ {
+ executor = initialiser.get();
+ }
+ }
}
+ return executor;
}
- public String getJmxName()
+ private static List<ExecutorService> executors()
{
- String name = "";
- for (String word : toString().split("_"))
+ return Stream.of(Stage.values())
+ .map(Stage::executor)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * This method shuts down all registered stages.
+ */
+ public static void shutdownNow()
+ {
+ ExecutorUtils.shutdownNow(executors());
+ }
+
+ @VisibleForTesting
+ public static void shutdownAndWait(long timeout, TimeUnit units) throws InterruptedException, TimeoutException
+ {
+ List<ExecutorService> executors = executors();
+ ExecutorUtils.shutdownNow(executors);
+ ExecutorUtils.awaitTermination(timeout, units, executors);
+ }
+
+ static LocalAwareExecutorService tracingExecutor(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize)
+ {
+ RejectedExecutionHandler reh = (r, executor) -> MessagingService.instance().metrics.recordSelfDroppedMessage(Verb._TRACE);
+ return new TracingExecutor(1,
+ 1,
+ KEEP_ALIVE_SECONDS,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(1000),
+ new NamedThreadFactory(jmxName),
+ reh);
+ }
+
+ static LocalAwareExecutorService multiThreadedStage(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize)
+ {
+ return new JMXEnabledThreadPoolExecutor(numThreads,
+ KEEP_ALIVE_SECONDS,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new NamedThreadFactory(jmxName),
+ jmxType);
+ }
+
+ static LocalAwareExecutorService multiThreadedLowSignalStage(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize)
+ {
- return SharedExecutorPool.SHARED.newExecutor(numThreads, onSetMaximumPoolSize, Integer.MAX_VALUE, jmxType, jmxName);
++ return SharedExecutorPool.SHARED.newExecutor(numThreads, onSetMaximumPoolSize, jmxType, jmxName);
+ }
+
+ static LocalAwareExecutorService singleThreadedStage(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize)
+ {
+ return new JMXEnabledSingleThreadExecutor(jmxName, jmxType);
+ }
+
+ static LocalAwareExecutorService immediateExecutor(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize)
+ {
+ return ImmediateExecutor.INSTANCE;
+ }
+
+ @FunctionalInterface
+ public interface ExecutorServiceInitialiser
+ {
+ public LocalAwareExecutorService init(String jmxName, String jmxType, int numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize);
+ }
+
+ /**
+ * Returns core thread pool size
+ */
+ public int getCorePoolSize()
+ {
+ return executor().getCorePoolSize();
+ }
+
+ /**
+ * Allows user to resize core thread pool size
+ */
+ public void setCorePoolSize(int newCorePoolSize)
+ {
+ executor().setCorePoolSize(newCorePoolSize);
+ }
+
+ /**
+ * Returns maximum pool size of thread pool.
+ */
+ public int getMaximumPoolSize()
+ {
+ return executor().getMaximumPoolSize();
+ }
+
+ /**
+ * Allows user to resize maximum size of the thread pool.
+ */
+ public void setMaximumPoolSize(int newMaximumPoolSize)
+ {
+ executor().setMaximumPoolSize(newMaximumPoolSize);
+ }
+
+ /**
+ * The executor used for tracing.
+ */
+ private static class TracingExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService
+ {
+ TracingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
+ {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
+ }
+
+ public void execute(Runnable command, ExecutorLocals locals)
+ {
+ assert locals == null;
+ super.execute(command);
+ }
+
+ public void maybeExecuteImmediately(Runnable command)
+ {
+ execute(command);
+ }
+
+ @Override
+ public int getActiveTaskCount()
+ {
+ return getActiveCount();
+ }
+
+ @Override
+ public int getPendingTaskCount()
{
- name += word.substring(0, 1) + word.substring(1).toLowerCase();
+ return getQueue().size();
}
- return name + "Stage";
}
}
diff --cc src/java/org/apache/cassandra/transport/Message.java
index 17e67a2,99a8d3a..09ea2e5
--- a/src/java/org/apache/cassandra/transport/Message.java
+++ b/src/java/org/apache/cassandra/transport/Message.java
@@@ -462,8 -425,6 +462,7 @@@ public abstract class Messag
public static class Dispatcher extends SimpleChannelInboundHandler<Request>
{
private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
+ DatabaseDescriptor::setNativeTransportMaxThreads,
- Integer.MAX_VALUE,
"transport",
"Native-Transport-Requests");
diff --cc test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java
index 9a2d52d,f223e57..a2fc238
--- a/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java
+++ b/test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java
@@@ -76,126 -77,4 +76,126 @@@ public class SEPExecutorTes
}
}
}
+
+ @Test
+ public void changingMaxWorkersMeetsConcurrencyGoalsTest() throws InterruptedException, TimeoutException
+ {
+ // Number of busy worker threads to run and gum things up. Chosen to be
+ // between the low and high max pool size so the test exercises resizing
+ // under a number of different conditions.
+ final int numBusyWorkers = 2;
+ SharedExecutorPool sharedPool = new SharedExecutorPool("ChangingMaxWorkersMeetsConcurrencyGoalsTest");
+ final AtomicInteger notifiedMaxPoolSize = new AtomicInteger();
+
- LocalAwareExecutorService executor = sharedPool.newExecutor(0, notifiedMaxPoolSize::set, 4, "internal", "resizetest");
++ LocalAwareExecutorService executor = sharedPool.newExecutor(0, notifiedMaxPoolSize::set, "internal", "resizetest");
+
+ // Keep feeding the executor work while resizing
+ // so it stays under load.
+ AtomicBoolean stayBusy = new AtomicBoolean(true);
+ Semaphore busyWorkerPermits = new Semaphore(numBusyWorkers);
+ Thread makeBusy = new Thread(() -> {
+ while (stayBusy.get() == true)
+ {
+ try
+ {
+ if (busyWorkerPermits.tryAcquire(1, MILLISECONDS)) {
+ executor.execute(new BusyWork(busyWorkerPermits));
+ }
+ }
+ catch (InterruptedException e)
+ {
+ // ignore, will either stop looping if done or retry the lock
+ }
+ }
+ });
+
+ makeBusy.start();
+ try
+ {
+ for (int repeat = 0; repeat < 1000; repeat++)
+ {
+ assertMaxTaskConcurrency(executor, 1);
+ Assert.assertEquals(1, notifiedMaxPoolSize.get());
+
+ assertMaxTaskConcurrency(executor, 2);
+ Assert.assertEquals(2, notifiedMaxPoolSize.get());
+
+ assertMaxTaskConcurrency(executor, 1);
+ Assert.assertEquals(1, notifiedMaxPoolSize.get());
+
+ assertMaxTaskConcurrency(executor, 3);
+ Assert.assertEquals(3, notifiedMaxPoolSize.get());
+
+ executor.setMaximumPoolSize(0);
+ Assert.assertEquals(0, notifiedMaxPoolSize.get());
+
+ assertMaxTaskConcurrency(executor, 4);
+ Assert.assertEquals(4, notifiedMaxPoolSize.get());
+ }
+ }
+ finally
+ {
+ stayBusy.set(false);
+ makeBusy.join(TimeUnit.SECONDS.toMillis(5));
+ Assert.assertFalse("makeBusy thread should have checked stayBusy and exited",
+ makeBusy.isAlive());
+ sharedPool.shutdownAndWait(1L, MINUTES);
+ }
+ }
+
+ static class LatchWaiter implements Runnable
+ {
+ CountDownLatch latch;
+ long timeout;
+ TimeUnit unit;
+
+ public LatchWaiter(CountDownLatch latch, long timeout, TimeUnit unit)
+ {
+ this.latch = latch;
+ this.timeout = timeout;
+ this.unit = unit;
+ }
+
+ public void run()
+ {
+ latch.countDown();
+ try
+ {
+ latch.await(timeout, unit); // block until all the latch waiters have run, now at desired concurrency
+ }
+ catch (InterruptedException e)
+ {
+ Assert.fail("interrupted: " + e);
+ }
+ }
+ }
+
+ static class BusyWork implements Runnable
+ {
+ private Semaphore busyWorkers;
+
+ public BusyWork(Semaphore busyWorkers)
+ {
+ this.busyWorkers = busyWorkers;
+ }
+
+ public void run()
+ {
+ busyWorkers.release();
+ }
+ }
+
+ void assertMaxTaskConcurrency(LocalAwareExecutorService executor, int concurrency) throws InterruptedException
+ {
+ executor.setMaximumPoolSize(concurrency);
+
+ CountDownLatch concurrencyGoal = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++)
+ {
+ executor.execute(new LatchWaiter(concurrencyGoal, 5L, TimeUnit.SECONDS));
+ }
+ // Will return true if all of the LatchWaiters count down before the timeout
+ Assert.assertEquals("Test tasks did not hit max concurrency goal",
+ true, concurrencyGoal.await(3L, TimeUnit.SECONDS));
+ }
}
diff --cc test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java
index 0000000,0000000..03c1acf
new file mode 100644
--- /dev/null
+++ b/test/unit/org/apache/cassandra/metrics/ThreadPoolMetricsTest.java
@@@ -1,0 -1,0 +1,311 @@@
++/*
++ * 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.metrics;
++
++import java.util.concurrent.ArrayBlockingQueue;
++import java.util.concurrent.CountDownLatch;
++import java.util.concurrent.TimeUnit;
++import java.util.concurrent.atomic.AtomicInteger;
++import java.util.function.Supplier;
++
++import org.junit.Test;
++
++import org.apache.cassandra.Util;
++import org.apache.cassandra.concurrent.*;
++
++import static org.junit.Assert.*;
++
++public class ThreadPoolMetricsTest
++{
++ @Test
++ public void testJMXEnabledThreadPoolMetricsWithNoBlockedThread()
++ {
++ JMXEnabledThreadPoolExecutor executor = new JMXEnabledThreadPoolExecutor(2,
++ Integer.MAX_VALUE,
++ TimeUnit.SECONDS,
++ new ArrayBlockingQueue<>(2),
++ new NamedThreadFactory("ThreadPoolMetricsTest-1"),
++ "internal");
++ testMetricsWithNoBlockedThreads(executor, executor.metrics);
++ }
++
++ @Test
++ public void testJMXEnabledThreadPoolMetricsWithBlockedThread()
++ {
++ JMXEnabledThreadPoolExecutor executor = new JMXEnabledThreadPoolExecutor(2,
++ Integer.MAX_VALUE,
++ TimeUnit.SECONDS,
++ new ArrayBlockingQueue<>(2),
++ new NamedThreadFactory("ThreadPoolMetricsTest-2"),
++ "internal");
++ testMetricsWithBlockedThreads(executor, executor.metrics);
++ }
++
++ @Test
++ public void testSEPExecutorMetrics()
++ {
++ SEPExecutor executor = (SEPExecutor) new SharedExecutorPool("ThreadPoolMetricsTest-2").newExecutor(2,
++ "ThreadPoolMetricsTest-3",
++ "internal");
++
++ testMetricsWithNoBlockedThreads(executor, executor.metrics);
++ }
++
++ private static void testMetricsWithBlockedThreads(LocalAwareExecutorService threadPool, ThreadPoolMetrics metrics)
++ {
++ assertEquals(2, metrics.maxPoolSize.getValue().intValue());
++
++ spinAssertEquals(0, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ BlockingTask task1 = new BlockingTask();
++ BlockingTask task2 = new BlockingTask();
++ BlockingTask task3 = new BlockingTask();
++ BlockingTask task4 = new BlockingTask();
++
++ // The ThreadPool has a size of 2 so the 2 first tasks should go into active straight away
++ threadPool.execute(task1);
++ threadPool.execute(task2);
++
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // There are no threads available any more the 2 next tasks should go into the queue
++ threadPool.execute(task3);
++ threadPool.execute(task4);
++
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // The queue is full the 2 next task should go into blocked and block the thread
++ BlockingTask task5 = new BlockingTask();
++ BlockingTask task6 = new BlockingTask();
++
++ AtomicInteger blockedThreads = new AtomicInteger(0);
++ new Thread(() ->
++ {
++ blockedThreads.incrementAndGet();
++ threadPool.execute(task5);
++ blockedThreads.decrementAndGet();
++ }).start();
++
++ spinAssertEquals(1, blockedThreads::get);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(1L, metrics.currentBlocked::getCount);
++ spinAssertEquals(1L, metrics.totalBlocked::getCount);
++
++ new Thread(() ->
++ {
++ blockedThreads.incrementAndGet();
++ threadPool.execute(task6);
++ blockedThreads.decrementAndGet();
++ }).start();
++
++ spinAssertEquals(2, blockedThreads::get);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(2L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++
++ // Allowing first task to complete
++ task1.allowToComplete();
++
++ spinAssertEquals(true, task3::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(1L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(1L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++ spinAssertEquals(1, blockedThreads::get);
++
++ // Allowing second task to complete
++ task2.allowToComplete();
++
++ spinAssertEquals(true, task4::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(2L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++ spinAssertEquals(0, blockedThreads::get);
++
++ // Allowing third task to complete
++ task3.allowToComplete();
++
++ spinAssertEquals(true, task5::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(3L, metrics.completedTasks::getValue);
++ spinAssertEquals(1, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++
++ // Allowing fourth task to complete
++ task4.allowToComplete();
++
++ spinAssertEquals(true, task6::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(4L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++
++ // Allowing last tasks to complete
++ task5.allowToComplete();
++ task6.allowToComplete();
++
++ spinAssertEquals(0, metrics.activeTasks::getValue);
++ spinAssertEquals(6L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(2L, metrics.totalBlocked::getCount);
++ }
++
++ private static void testMetricsWithNoBlockedThreads(LocalAwareExecutorService threadPool, ThreadPoolMetrics metrics)
++ {
++ spinAssertEquals(0, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ BlockingTask task1 = new BlockingTask();
++ BlockingTask task2 = new BlockingTask();
++ BlockingTask task3 = new BlockingTask();
++ BlockingTask task4 = new BlockingTask();
++
++ // The ThreadPool has a size of 2 so the 2 first tasks should go into active straight away
++ threadPool.execute(task1);
++
++ spinAssertEquals(1, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ threadPool.execute(task2);
++
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // There are no threads available any more the 2 next tasks should go into the queue
++ threadPool.execute(task3);
++
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(1, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ threadPool.execute(task4);
++
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(0L, metrics.completedTasks::getValue);
++ spinAssertEquals(2, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // Allowing first task to complete
++ task1.allowToComplete();
++
++ spinAssertEquals(true, task3::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(1L, metrics.completedTasks::getValue);
++ spinAssertEquals(1, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // Allowing second task to complete
++ task2.allowToComplete();
++
++ spinAssertEquals(true, task4::isStarted);
++ spinAssertEquals(2, metrics.activeTasks::getValue);
++ spinAssertEquals(2L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // Allowing third task to complete
++ task3.allowToComplete();
++
++ spinAssertEquals(1, metrics.activeTasks::getValue);
++ spinAssertEquals(3L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++
++ // Allowing fourth task to complete
++ task4.allowToComplete();
++
++ spinAssertEquals(0, metrics.activeTasks::getValue);
++ spinAssertEquals(4L, metrics.completedTasks::getValue);
++ spinAssertEquals(0, metrics.pendingTasks::getValue);
++ spinAssertEquals(0L, metrics.currentBlocked::getCount);
++ spinAssertEquals(0L, metrics.totalBlocked::getCount);
++ }
++
++ private static void spinAssertEquals(Object expected, Supplier<Object> actualSupplier)
++ {
++ Util.spinAssertEquals(expected, actualSupplier, 1);
++ }
++
++ private static final class BlockingTask implements Runnable
++ {
++ private final CountDownLatch latch = new CountDownLatch(1);
++
++ private volatile boolean started;
++
++ public boolean isStarted()
++ {
++ return started;
++ }
++
++ public void allowToComplete()
++ {
++ latch.countDown();
++ }
++
++ @Override
++ public void run()
++ {
++ started = true;
++ try
++ {
++ latch.await(30, TimeUnit.SECONDS);
++ }
++ catch (InterruptedException e)
++ {
++ Thread.currentThread().interrupt();
++ }
++ }
++ }
++}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org