From 88aa5b6807dbd97446d34864e87a34493880358b Mon Sep 17 00:00:00 2001
From: Dmitry Konstantinov <netudima@gmail.com>
Date: Sat, 6 Jun 2026 18:56:14 +0100
Subject: [PATCH] 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
---
 CHANGES.txt                                   |  1 +
 .../cassandra/concurrent/SEPExecutor.java     | 29 +++++--------------
 2 files changed, 8 insertions(+), 22 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index 8e4c39b56f..86f663dfdf 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 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)
  * 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)
diff --git a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java
index 32ac77afb4..ccce797be7 100644
--- a/src/java/org/apache/cassandra/concurrent/SEPExecutor.java
+++ b/src/java/org/apache/cassandra/concurrent/SEPExecutor.java
@@ -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