From bfcdf927f812a8b2cbfb01183df57cbc632db92d Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Mon, 6 Jul 2026 10:29:06 +0100 Subject: [PATCH] Fix SignalLock.incrementAsyncWork (wrong guard, can lead to ISE) --- .../service/accord/AccordExecutor.java | 18 +++---- .../AccordExecutorAbstractLockLoop.java | 4 +- .../accord/AccordExecutorSignalLoop.java | 4 +- .../service/accord/AccordExecutorSimple.java | 2 +- .../cassandra/service/accord/AccordTask.java | 2 +- .../utils/concurrent/SignalLock.java | 8 +++- .../simulator/test/AccordExecutorTest.java | 48 +++++++++++++++---- 7 files changed, 60 insertions(+), 26 deletions(-) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index b1aff08f13..cff519396a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -1176,7 +1176,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor) task).preLoadContext())); else - task.runInternal(); + task.run(); // NOTE: cannot safely release owner here, in case an immediate-execution runs before we can release our references and store their changes to the cache } @@ -1631,7 +1631,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor name, Agent agent) { super(lock, executorId, agent); - Invariants.require(threads < SignalLock.MAX_THREADS); + Invariants.require(threads <= SignalLock.MAX_THREADS); this.lock = lock; this.loops = new AccordExecutorLoops(mode, threads, name, this::task); this.readyToRunLimit = Math.min(threads * 4, SignalLock.MAX_SIGNAL_COUNT); @@ -397,7 +397,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop try { setRunning(task); - task.runInternal(); + task.run(); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java index 4e89f068ab..ab6c8c25e7 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java @@ -95,7 +95,7 @@ class AccordExecutorSimple extends AccordExecutor return; } - try { task.preRunExclusive(); task.runInternal(); } + try { task.preRunExclusive(); task.run(); } catch (Throwable t) { task.fail(t); } finally { diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index e3fe9656e0..a766cc2097 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -672,7 +672,7 @@ public abstract class AccordTask extends Task implements Function= 1 && increment <= MAX_SIGNAL_COUNT); while (true) { long cur = state; int count = asyncSignalCount(cur); - if (count == MAX_THREADS) + if (count == MAX_SIGNAL_COUNT) return false; if (signalIfWaiting) diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java index 44888830cc..103f45b2db 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java @@ -18,7 +18,9 @@ package org.apache.cassandra.simulator.test; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; @@ -36,6 +38,7 @@ import accord.api.AsyncExecutor; import accord.local.SequentialAsyncExecutor; import org.apache.cassandra.concurrent.ExecutorFactory; +import org.apache.cassandra.concurrent.ExecutorPlus; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableSupplier; import org.apache.cassandra.service.accord.AccordExecutor; @@ -43,47 +46,71 @@ import org.apache.cassandra.service.accord.AccordExecutorAsyncSubmit; import org.apache.cassandra.service.accord.AccordExecutorSignalLoop; import org.apache.cassandra.service.accord.api.AccordAgent; +import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITHOUT_LOCK; import static org.apache.cassandra.service.accord.AccordService.toFuture; // TODO (required): have simulator intercept ReentantLock so we can test SyncSubmit and SemiSyncSubmit public class AccordExecutorTest extends SimulationTestBase { - static final int THREAD_COUNT = 8; + static final int EXECUTOR_THREAD_COUNT = 44; @Test public void signalLoopTest() { - executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, THREAD_COUNT, -1, -1, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent())); + executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, -1, -1, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent()), + 16); } @Test public void signalSpinLoopTest() { - executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, THREAD_COUNT, 10, 100, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent())); + executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, 10, 100, TimeUnit.MICROSECONDS, i ->"Loop" + i, new AccordAgent()), + 16); } @Test public void ayncSubmitTest() { - executorTest(() -> new AccordExecutorAsyncSubmit(1, RUN_WITHOUT_LOCK, THREAD_COUNT, i -> "Loop" + i, new AccordAgent())); + executorTest(() -> new AccordExecutorAsyncSubmit(1, RUN_WITHOUT_LOCK, EXECUTOR_THREAD_COUNT, i -> "Loop" + i, new AccordAgent()), + 16); } - public void executorTest(SerializableSupplier supplier) + public void executorTest(SerializableSupplier supplier, int submissionThreads) { simulate(arr(() -> { try { DatabaseDescriptor.daemonInitialization(); + ExecutorPlus submit = executorFactory().pooled("submit-test", submissionThreads); AccordExecutor executor = supplier.get(); Lock lock = executor.unsafeLock(); SequentialAsyncExecutor sequentialExecutor = executor.newSequentialExecutor(); - Executor lockExecutor = ExecutorFactory.Global.executorFactory().sequential("lock"); - ConcurrentLinkedQueue> await = new ConcurrentLinkedQueue<>(); + Executor lockExecutor = executorFactory().sequential("lock"); for (float sleepChance : new float[] { 0f, 0.01f, 0.1f }) + { for (float lockChance : new float[] { 0f, 0.01f, 0.1f }) - submitLoop(lock, executor, sequentialExecutor, lockExecutor, 200, 10, await, sleepChance, lockChance); + { + List> done = new ArrayList<>(); + for (int i = 0 ; i < submissionThreads ; ++i) + { + int id = i; + done.add(submit.submit(() -> { + try + { + submitLoop(id, lock, executor, sequentialExecutor, lockExecutor, 20, 10, sleepChance, lockChance); + } + catch (ExecutionException | InterruptedException e) + { + throw new RuntimeException(e); + } + })); + } + for (Future f : done) + f.get(); + } + } } catch (Throwable t) { @@ -93,8 +120,9 @@ public class AccordExecutorTest extends SimulationTestBase () -> {}, 1L); } - private static void submitLoop(Lock lock, AccordExecutor executor, SequentialAsyncExecutor sequentialExecutor, Executor lockExecutor, int outerLoop, int innerLoop, ConcurrentLinkedQueue> await, float sleepChance, float lockChance) throws ExecutionException, InterruptedException + private static void submitLoop(int id, Lock lock, AccordExecutor executor, SequentialAsyncExecutor sequentialExecutor, Executor lockExecutor, int outerLoop, int innerLoop, float sleepChance, float lockChance) throws ExecutionException, InterruptedException { + ConcurrentLinkedQueue> await = new ConcurrentLinkedQueue<>(); while (outerLoop-- > 0) { for (int i = 0; i < innerLoop; ++i) @@ -105,7 +133,7 @@ public class AccordExecutorTest extends SimulationTestBase while (!await.isEmpty()) await.poll().get(); done.set(true); - System.out.println("Loop " + (1 + outerLoop)); + System.out.println("Loop " + id + '.' + (1 + outerLoop)); } }