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

Avoid cache modification reentrancy when cancelling loads 

Also fix:
 - Catch exceptions thrown by Task.cleanupExclusive to ensure we do not corrupt SequentialExecutor state
 - ConcurrentModificationException with CommandStore.waitingOnSync
 - Exclude promisedHlc that are in the future, even if not equal to Long.MAX_VALUE
 - Filter CFK on load as not expected to be up to date (for later on demand filtering validation)

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20821
This commit is contained in:
Benedict Elliott Smith 2025-08-06 12:19:06 +01:00
parent bb7e706213
commit 7b12404a4e
11 changed files with 73 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit d5de79b56f792d3b6d868ca46845b830e8906828
Subproject commit 12da4693b449b30d0420673579a06025b7b3e484

3
src/java/org/apache/cassandra/service/accord/AccordCache.java
View File

@ -52,6 +52,7 @@ import accord.utils.IntrusiveLinkedList;
import accord.utils.Invariants;
import accord.utils.QuadFunction;
import accord.utils.TriFunction;
import accord.utils.UnhandledEnum;
import org.agrona.collections.Object2ObjectHashMap;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -255,7 +256,7 @@ public class AccordCache implements CacheSize
Status status = node.status();
switch (status)
{
default: throw new IllegalStateException("Unhandled status " + status);
default: throw new UnhandledEnum(status);
case LOADING:
node.loading().loading.cancel();
case WAITING_TO_LOAD:

8
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -308,7 +308,13 @@ public class AccordCommandStore extends CommandStore
CommandsForKey loadCommandsForKey(RoutableKey key)
{
return CommandsForKeyAccessor.load(id, (TokenKey) key);
CommandsForKey cfk = CommandsForKeyAccessor.load(id, (TokenKey) key);
if (cfk == null)
return null;
RedundantBefore.QuickBounds bounds = unsafeGetRedundantBefore().get(key);
if (bounds == null)
return cfk; // TODO (required): I don't think this should be possible? but we hit it on some test
return cfk.withRedundantBeforeAtLeast(bounds.gcBefore, false);
}
boolean validateCommandsForKey(RoutableKey key, CommandsForKey evicting)

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

@ -36,8 +36,6 @@ import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.base.Functions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -463,7 +461,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
}
void consumeExclusive(Object object)
void consumeExclusive(Submittable object)
{
try
{
@ -538,31 +536,31 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return submitPlainExclusive(null, new SaveRunnable(entry, identity, save));
}
private <P1> void submit(BiConsumer<AccordExecutor, P1> sync, Function<P1, ?> async, P1 p1)
private <P1> void submit(BiConsumer<AccordExecutor, P1> sync, Function<P1, Submittable> async, P1 p1)
{
submit((e, c, p1a, p2a, p3) -> c.accept(e, p1a), (f, p1a, p2a, p3) -> f.apply(p1a), sync, async, p1, null, null);
}
private <P1, P2> void submit(TriConsumer<AccordExecutor, P1, P2> sync, BiFunction<P1, P2, ?> async, P1 p1, P2 p2)
private <P1, P2> void submit(TriConsumer<AccordExecutor, P1, P2> sync, BiFunction<P1, P2, Submittable> async, P1 p1, P2 p2)
{
submit((e, c, p1a, p2a, p3) -> c.accept(e, p1a, p2a), (f, p1a, p2a, p3) -> f.apply(p1a, p2a), sync, async, p1, p2, null);
}
private <P1, P2, P3> void submit(QuadConsumer<AccordExecutor, P1, P2, P3> sync, TriFunction<P1, P2, P3, ?> async, P1 p1, P2 p2, P3 p3)
private <P1, P2, P3> void submit(QuadConsumer<AccordExecutor, P1, P2, P3> sync, TriFunction<P1, P2, P3, Submittable> async, P1 p1, P2 p2, P3 p3)
{
submit((e, c, p1a, p2a, p3a) -> c.accept(e, p1a, p2a, p3a), TriFunction::apply, sync, async, p1, p2, p3);
}
private <P1, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1, P2, P3, P4> sync, QuadFunction<P1, P2, P3, P4, Object> async, P1 p1, P2 p2, P3 p3, P4 p4)
private <P1, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1, P2, P3, P4> sync, QuadFunction<P1, P2, P3, P4, Submittable> async, P1 p1, P2 p2, P3 p3, P4 p4)
{
submit(sync, async, p1, p1, p2, p3, p4);
}
abstract <P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4);
abstract <P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4);
<R> void submit(AccordTask<R> operation)
{
submit(AccordExecutor::submitExclusive, Function.identity(), operation);
submit(AccordExecutor::submitExclusive, i -> i, operation);
}
void submitExclusive(AccordTask<?> task)
@ -775,7 +773,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
private Cancellable submit(Plain task)
{
submit(AccordExecutor::submitPlainExclusive, Functions.identity(), task);
submit(AccordExecutor::submitPlainExclusive, i -> i, task);
return task;
}
@ -898,10 +896,16 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
*/
abstract protected void cleanupExclusive();
void cancelExclusive(AccordExecutor owner) {}
abstract protected void addToQueue(TaskQueue queue);
}
static abstract class SubmittableTask extends Task
interface Submittable
{
}
static abstract class SubmittableTask extends Task implements Submittable
{
final WithResources locals = ExecutorLocals.propagate();
abstract void submitExclusive(AccordExecutor owner);
@ -990,15 +994,18 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
void cleanupTask()
{
task.cleanupExclusive();
running = false;
owner = null;
task = super.poll();
if (task != null)
try { task.cleanupExclusive(); }
finally
{
AccordExecutor.this.running.remove(selfTask);
selfTask.queuePosition = task.queuePosition;
waitingToRun.append(selfTask);
owner = null;
running = false;
task = super.poll();
if (task != null)
{
AccordExecutor.this.running.remove(selfTask);
selfTask.queuePosition = task.queuePosition;
waitingToRun.append(selfTask);
}
}
}
@ -1178,16 +1185,16 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
}
private abstract static class SubmitAsync
private abstract static class SubmitAsync implements Submittable
{
abstract void submitExclusive(AccordExecutor executor);
}
private static class CancelAsync<R> extends SubmitAsync
private static class CancelAsync extends SubmitAsync
{
final AccordTask<R> cancel;
final Task cancel;
private CancelAsync(AccordTask<R> cancel)
private CancelAsync(Task cancel)
{
this.cancel = cancel;
}
@ -1195,7 +1202,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
@Override
void submitExclusive(AccordExecutor executor)
{
executor.cancelExclusive(cancel);
cancel.cancelExclusive(executor);
}
}
@ -1224,17 +1231,20 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
@Override
public void cancel()
{
executeDirectlyWithLock(() -> {
SequentialExecutor executor = executor();
TaskQueue queue = executor == null ? waitingToRun : executor;
if (queue.contains(this))
{
queue.remove(this);
try { fail(new CancellationException()); }
catch (Throwable t) { agent.onUncaughtException(t); }
completeTaskExclusive(this, true);
}
});
submit((e, c) -> c.cancelExclusive(e), CancelAsync::new, this);
}
void cancelExclusive(AccordExecutor owner)
{
SequentialExecutor executor = executor();
TaskQueue queue = executor == null ? waitingToRun : executor;
if (queue.contains(this))
{
queue.remove(this);
completeTaskExclusive(this, true);
try { fail(new CancellationException()); }
catch (Throwable t) { agent.onUncaughtException(t); }
}
}
@Override

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

@ -33,7 +33,7 @@ import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_L
abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
{
final ConcurrentLinkedStack<Object> submitted = new ConcurrentLinkedStack<>();
final ConcurrentLinkedStack<Submittable> submitted = new ConcurrentLinkedStack<>();
boolean isHeldByExecutor;
boolean shutdown;
@ -47,17 +47,17 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
abstract void awaitExclusive() throws InterruptedException;
abstract AccordExecutorLoops loops();
abstract boolean isInLoop();
abstract <P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4);
abstract <P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4);
<P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
<P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
{
// if we're a loop thread, we will poll the waitingToRun queue when we come around
// NOTE: this assumes no synchronous blocking tasks are submitted to this executor
if (isInLoop()) submitted.push(async.apply(p1a, p2, p3, p4));
if (isInLoop() || isOwningThread()) submitted.push(async.apply(p1a, p2, p3, p4));
else submitExternal(sync, async, p1s, p1a, p2, p3, p4);
}
<P1s, P1a, P2, P3, P4> void submitExternalExclusive(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
<P1s, P1a, P2, P3, P4> void submitExternalExclusive(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
{
try
{

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

@ -34,7 +34,7 @@ abstract class AccordExecutorAbstractSemiSyncSubmit extends AccordExecutorAbstra
abstract void awaitExclusive() throws InterruptedException;
<P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
<P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
{
if (!lock.tryLock())
{

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

@ -111,7 +111,7 @@ class AccordExecutorSimple extends AccordExecutor
}
@Override
<P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
<P1s, P1a, P2, P3, P4> void submit(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
{
lock.lock();
try

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

@ -96,7 +96,7 @@ class AccordExecutorSyncSubmit extends AccordExecutorAbstractLockLoop
hasWork.signal();
}
<P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Object> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
<P1s, P1a, P2, P3, P4> void submitExternal(QuintConsumer<AccordExecutor, P1s, P2, P3, P4> sync, QuadFunction<P1a, P2, P3, P4, Submittable> async, P1s p1s, P1a p1a, P2 p2, P3 p3, P4 p4)
{
lock.lock();
try

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

@ -784,6 +784,11 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
callback.accept(null, new CancellationException());
}
void cancelExclusive(AccordExecutor owner)
{
owner.cancel(this);
}
public State state()
{
return state;

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

@ -246,13 +246,13 @@ public class AccordAgent implements Agent
Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch());
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
long oneSecond = SECONDS.toMicros(1L);
long promisedHlc = command.promised().hlc();
if (promisedHlc == Long.MAX_VALUE)
if (promisedHlc > nowMicros + TimeUnit.MINUTES.toMicros(1))
promisedHlc = 0;
long mostRecentStart = Math.max(command.txnId().hlc(), promisedHlc);
long waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS);
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
if (mostRecentStart > nowMicros + SECONDS.toMicros(1L))
logger.warn("max({},{})>{}", command.txnId(), command.promised(), nowMicros);
long startTime = mostRecentStart + waitMicros;

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

@ -43,6 +43,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
@ -258,7 +259,8 @@ public class AccordLoadTest extends AccordTestBase
try
{
i.acceptOnInstance(name -> {
AccordKeyspace.AccordColumnFamilyStores.commandsForKey.forceFlush(UNIT_TESTS);
if (CommitLog.instance.isStarted())
AccordKeyspace.AccordColumnFamilyStores.commandsForKey.forceFlush(UNIT_TESTS);
}, accordTableName);
}
catch (Throwable t)