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

Merge branch 'cassandra-6.0' into trunk 

* cassandra-6.0:
  SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity
This commit is contained in:
Dmitry Konstantinov 2026-06-08 23:06:33 +01:00
parent a84647c6c7 88aa5b6807
commit c9d468bf3c
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

@ -3,6 +3,7 @@
* Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
* Add a guardrail for misprepared statements (CASSANDRA-21139)
Merged from 6.0:
* SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429)
* Safely regain ranges and delete retired command stores (CASSANDRA-21212)
* 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)

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
// this permits us to schedule threads non-spuriously; it also means work is serviced fairly
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
// 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)
{
long current = permits.get();
int workPermits = workPermits(current);
int taskPermits = taskPermits(current);
if (workPermits <= 0 || taskPermits == 0)
int newWorkPermits = workPermits(current) - 1;
int newTaskPermits = taskPermits(current) - taskDelta;
if (newWorkPermits < 0 || newTaskPermits < 0)
return false;
if (permits.compareAndSet(current, combine(taskPermits - taskDelta, workPermits - 1)))
if (permits.compareAndSet(current, combine(newTaskPermits, newWorkPermits)))
{
return true;
}
@ -207,13 +198,7 @@ public class SEPExecutor implements LocalAwareExecutorPlus, SEPExecutorMBean
// gives up a work permit
void returnWorkPermit()
{
while (true)
{
long current = permits.get();
int workPermits = workPermits(current);
if (permits.compareAndSet(current, updateWorkPermits(current, workPermits + 1)))
return;
}
permits.addAndGet(updateWorkPermits(0, 1));
}
@Override