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

Fix SignalLock.incrementAsyncWork (wrong guard, can lead to ISE)

This commit is contained in:
Benedict Elliott Smith 2026-07-06 10:29:06 +01:00
parent 57dabaea8c
commit bfcdf927f8
7 changed files with 60 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

@ -1176,7 +1176,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
/**
* Run the command; the state cache lock may or may not be held depending on the executor implementation
*/
protected abstract void runInternal();
protected abstract void run();
/**
* Fail the command; the state cache lock may or may not be held depending on the executor implementation
@ -1235,7 +1235,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
protected void runInternal()
protected void run()
{
queue.runTask();
}
@ -1322,7 +1322,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
if (stopped && reject(task))
task.fail(new RejectedExecutionException(commandStoreId + " is terminated. Cannot execute " + ((AccordTask<?>) 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<AccordTa
private CancelTask(Task cancel) { this.cancel = cancel; }
@Override void submitExclusive(AccordExecutor owner) { cancel.cancelExclusive(owner); }
@Override protected void preRunExclusive() { throw new UnsupportedOperationException(); }
@Override protected void runInternal() { throw new UnsupportedOperationException(); }
@Override protected void run() { throw new UnsupportedOperationException(); }
@Override protected void fail(Throwable fail) { throw new UnsupportedOperationException(); }
@Override protected void addToQueue(TaskQueue queue) { throw new UnsupportedOperationException(); }
}
@ -1706,7 +1706,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
protected void runInternal()
protected void run()
{
onRunning();
try (Closeable close = resources.get())
@ -1809,7 +1809,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
public void runInternal()
public void run()
{
onRunning();
try (Closeable close = resources.get())
@ -1856,7 +1856,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
protected void runInternal()
protected void run()
{
onRunning();
try (Closeable close = resources.get())
@ -1907,7 +1907,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
public void runInternal()
public void run()
{
onRunning();
try (Closeable close = resources.get())
@ -1955,7 +1955,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
@Override
protected void runInternal()
protected void run()
{
onRunning();
try (Closeable close = resources.get())

4
src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
View File

@ -148,7 +148,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
try
{
task.preRunExclusive();
task.runInternal();
task.run();
}
catch (Throwable t)
{
@ -282,7 +282,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
try
{
++count;
task.runInternal();
task.run();
}
catch (Throwable t)
{

4
src/java/org/apache/cassandra/service/accord/AccordExecutorSignalLoop.java
View File

@ -61,7 +61,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
public AccordExecutorSignalLoop(SignalLock lock, int executorId, Mode mode, int threads, IntFunction<String> 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)
{

2
src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java
View File

@ -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
{

2
src/java/org/apache/cassandra/service/accord/AccordTask.java
View File

@ -672,7 +672,7 @@ public abstract class AccordTask<R> extends Task implements Function<SafeCommand
}
@Override
public void runInternal()
public void run()
{
onRunning();
logger.trace("Running {} with state {}", this, state);

8
src/java/org/apache/cassandra/utils/concurrent/SignalLock.java
View File

@ -424,11 +424,17 @@ public final class SignalLock implements Lock
public boolean incrementAsyncWork(boolean signalIfWaiting)
{
return incrementAsyncWork(signalIfWaiting, 1);
}
public boolean incrementAsyncWork(boolean signalIfWaiting, int increment)
{
Invariants.require(increment >= 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)

48
test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorTest.java
View File

@ -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<AccordExecutor> supplier)
public void executorTest(SerializableSupplier<AccordExecutor> 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<Future<?>> 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<Future<?>> 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<Future<?>> 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<Future<?>> 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));
}
}