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

Remove the SEPExecutor blocking behavior 

patch by Benjamin Lerer; reviewed by Andrés de la Peña for CASSANDRA-16186

If the number of tasks added to a SEPExecutor exceed the max queue size,
the threads adding those task will be block until enough space become
available for all the blocked tasks. At this point all the blocked threads
will be released at once.

As the maxTasksQueued was always set to INTEGER.MAX_VALUE, the code was in
practice never using. By consequence, removing the code was a better
option than trying to fix it.
This commit is contained in:
Benjamin Lerer 2020-11-10 09:55:20 +01:00
parent 0845008061
commit 3b970ddbd6
7 changed files with 10 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.24:
* Remove the SEPExecutor blocking behavior (CASSANDRA-16186)
* Wait for schema agreement when bootstrapping (CASSANDRA-15158)
* Fix invalid cell value skipping when reading from disk (CASSANDRA-16223)
* Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146)

34
src/java/org/apache/cassandra/concurrent/SEPExecutor.java
View File

@ -25,7 +25,6 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.metrics.SEPMetrics;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
@ -34,7 +33,6 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
private final SharedExecutorPool pool;
public final int maxWorkers;
private final int maxTasksQueued;
private final SEPMetrics metrics;
// stores both a set of work permits and task permits:
@ -42,8 +40,6 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
// top 32 bits are number of work permits available in the range [0..maxWorkers] (initially maxWorkers)
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;
@ -52,11 +48,10 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
// TODO: see if other queue implementations might improve throughput
protected final ConcurrentLinkedQueue<FutureTask<?>> tasks = new ConcurrentLinkedQueue<>();
SEPExecutor(SharedExecutorPool pool, int maxWorkers, int maxTasksQueued, String jmxPath, String name)
SEPExecutor(SharedExecutorPool pool, int maxWorkers, String jmxPath, String name)
{
this.pool = pool;
this.maxWorkers = maxWorkers;
this.maxTasksQueued = maxTasksQueued;
this.permits.set(combine(0, maxWorkers));
this.metrics = new SEPMetrics(this, jmxPath, name);
}
@ -102,29 +97,6 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
// 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();
}
}
// takes permission to perform a task, if any are available; once taken it is guaranteed
@ -139,8 +111,6 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
return false;
if (permits.compareAndSet(current, updateTaskPermits(current, taskPermits - 1)))
{
if (taskPermits == maxTasksQueued && hasRoom.hasWaiters())
hasRoom.signalAll();
return true;
}
}
@ -159,8 +129,6 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService
return false;
if (permits.compareAndSet(current, combine(taskPermits - taskDelta, workPermits - 1)))
{
if (takeTaskPermit && taskPermits == maxTasksQueued && hasRoom.hasWaiters())
hasRoom.signalAll();
return true;
}
}

4
src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java
View File

@ -108,9 +108,9 @@ public class SharedExecutorPool
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)
{
SEPExecutor executor = new SEPExecutor(this, maxConcurrency, maxQueuedTasks, jmxPath, name);
SEPExecutor executor = new SEPExecutor(this, maxConcurrency, jmxPath, name);
executors.add(executor);
return executor;
}

2
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -91,7 +91,7 @@ public class StageManager
private static LocalAwareExecutorService multiThreadedLowSignalStage(Stage stage, int numThreads)
{
return SharedExecutorPool.SHARED.newExecutor(numThreads, Integer.MAX_VALUE, stage.getJmxType(), stage.getJmxName());
return SharedExecutorPool.SHARED.newExecutor(numThreads, stage.getJmxType(), stage.getJmxName());
}
/**

1
src/java/org/apache/cassandra/transport/Message.java
View File

@ -421,7 +421,6 @@ public abstract class Message
public static class Dispatcher extends SimpleChannelInboundHandler<Request>
{
private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(),
Integer.MAX_VALUE,
"transport",
"Native-Transport-Requests");

8
test/burn/org/apache/cassandra/concurrent/LongSharedExecutorPoolTest.java
View File

@ -106,7 +106,7 @@ public class LongSharedExecutorPoolTest
{
final int executorCount = 4;
int threadCount = 8;
int maxQueued = 1024;
int scale = 1024;
final WeibullDistribution workTime = new WeibullDistribution(3, 200000);
final long minWorkTime = TimeUnit.MICROSECONDS.toNanos(1);
final long maxWorkTime = TimeUnit.MILLISECONDS.toNanos(1);
@ -116,11 +116,11 @@ public class LongSharedExecutorPoolTest
final ExecutorService[] executors = new ExecutorService[executorCount];
for (int i = 0 ; i < executors.length ; i++)
{
executors[i] = SharedExecutorPool.SHARED.newExecutor(threadCount, maxQueued, "test" + i, "test" + i);
executors[i] = SharedExecutorPool.SHARED.newExecutor(threadCount, "test" + i, "test" + i);
threadCounts[i] = threadCount;
workCount[i] = new WeibullDistribution(2, maxQueued);
workCount[i] = new WeibullDistribution(2, scale);
threadCount *= 2;
maxQueued *= 2;
scale *= 2;
}
long runs = 0;

2
test/unit/org/apache/cassandra/concurrent/SEPExecutorTest.java
View File

@ -51,7 +51,7 @@ public class SEPExecutorTest
for (int idx = 0; idx < 20; idx++)
{
ExecutorService es = sharedPool.newExecutor(FBUtilities.getAvailableProcessors(), Integer.MAX_VALUE, "STAGE", run + MAGIC + idx);
ExecutorService es = sharedPool.newExecutor(FBUtilities.getAvailableProcessors(), "STAGE", run + MAGIC + idx);
// Write to black hole
es.execute(() -> nullPrintSteam.println("TEST" + es));
}