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

SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity 

Additional improvement: use a wait-free logic to return a task or work permit

patch by Dmitry Konstantinov; reviewed by Benedict Elliott Smith for CASSANDRA-21429
This commit is contained in:
Dmitry Konstantinov 2026-06-06 18:56:14 +01:00
parent 22eb0f9fb7
commit 88aa5b6807
2 changed files with 8 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
6.0-alpha2 6.0-alpha2
* SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429)
* Safely regain ranges and delete retired command stores (CASSANDRA-21212) * Safely regain ranges and delete retired command stores (CASSANDRA-21212)
* Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360) * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360)
* Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362) * Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362)

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

@ -123,18 +123,9 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
// we add to the queue first, so that when a worker takes a task permit it can be certain there is a task available // we add to the queue first, so that when a worker takes a task permit it can be certain there is a task available
// this permits us to schedule threads non-spuriously; it also means work is serviced fairly // this permits us to schedule threads non-spuriously; it also means work is serviced fairly
tasks.add(task); tasks.add(task);
int taskPermits;
while (true)
{
long current = permits.get();
taskPermits = taskPermits(current);
// because there is no difference in practical terms between the work permit being added or not (the work is already in existence)
// we always add our permit, but block after the fact if we breached the queue limit
if (permits.compareAndSet(current, updateTaskPermits(current, taskPermits + 1)))
break;
}
if (taskPermits == 0) long prev = permits.getAndAdd(updateTaskPermits(0, 1));
if (taskPermits(prev) == 0)
{ {
// we only need to schedule a thread if there are no tasks already waiting to be processed, as // we only need to schedule a thread if there are no tasks already waiting to be processed, as
// the original enqueue will have started a thread to service its work which will have itself // the original enqueue will have started a thread to service its work which will have itself
@ -193,11 +184,11 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
while (true) while (true)
{ {
long current = permits.get(); long current = permits.get();
int workPermits = workPermits(current); int newWorkPermits = workPermits(current) - 1;
int taskPermits = taskPermits(current); int newTaskPermits = taskPermits(current) - taskDelta;
if (workPermits <= 0 || taskPermits == 0) if (newWorkPermits < 0 || newTaskPermits < 0)
return false; return false;
if (permits.compareAndSet(current, combine(taskPermits - taskDelta, workPermits - 1))) if (permits.compareAndSet(current, combine(newTaskPermits, newWorkPermits)))
{ {
return true; return true;
} }
@ -207,13 +198,7 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
// gives up a work permit // gives up a work permit
void returnWorkPermit() void returnWorkPermit()
{ {
while (true) permits.addAndGet(updateWorkPermits(0, 1));
{
long current = permits.get();
int workPermits = workPermits(current);
if (permits.compareAndSet(current, updateWorkPermits(current, workPermits + 1)))
return;
}
} }
@Override @Override