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

InterceptingExecutor.schedule returns a simulator-unsafe Future 

patch by Benedict Elliott Smith; reviewed by David Capwell for CASSANDRA-20420
This commit is contained in:
Benedict Elliott Smith 2025-03-08 14:24:49 -08:00 committed by David Capwell
parent fa1157203f
commit ab556ab545
3 changed files with 60 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutor.java
View File

@ -91,43 +91,9 @@ public interface InterceptingExecutor extends OrderOn
OrderOn orderAppliesAfterScheduling();
static class InterceptedScheduledFutureTask<T> extends SyncFutureTask<T> implements ScheduledFuture<T>
interface InterceptableScheduledFuture<T> extends ScheduledFuture<T>, RunnableFuture<T>
{
final long delayNanos;
Runnable onCancel;
public InterceptedScheduledFutureTask(long delayNanos, Callable<T> call)
{
super(call);
this.delayNanos = delayNanos;
}
@Override
public long getDelay(TimeUnit unit)
{
return unit.convert(delayNanos, NANOSECONDS);
}
@Override
public int compareTo(Delayed that)
{
return Long.compare(delayNanos, that.getDelay(NANOSECONDS));
}
void onCancel(Runnable onCancel)
{
this.onCancel = onCancel;
}
@Override
public boolean cancel(boolean b)
{
if (onCancel != null)
{
onCancel.run();
onCancel = null;
}
return super.cancel(b);
}
void onCancel(Runnable onCancel);
}
@PerClassLoader
@ -715,6 +681,45 @@ public interface InterceptingExecutor extends OrderOn
@PerClassLoader
class InterceptingSequentialExecutor extends AbstractSingleThreadedExecutorPlus implements InterceptingExecutor, ScheduledExecutorPlus, OrderOn
{
static class InterceptableScheduledFutureTask<T> extends SyncFutureTask<T> implements InterceptableScheduledFuture<T>
{
final long delayNanos;
Runnable onCancel;
public InterceptableScheduledFutureTask(long delayNanos, Callable<T> call)
{
super(call);
this.delayNanos = delayNanos;
}
@Override
public long getDelay(TimeUnit unit)
{
return unit.convert(delayNanos, NANOSECONDS);
}
@Override
public int compareTo(Delayed that)
{
return Long.compare(delayNanos, that.getDelay(NANOSECONDS));
}
public void onCancel(Runnable onCancel)
{
this.onCancel = onCancel;
}
@Override
public boolean cancel(boolean b)
{
if (onCancel != null)
{
onCancel.run();
onCancel = null;
}
return super.cancel(b);
}
}
InterceptingSequentialExecutor(InterceptorOfExecution interceptorOfExecution, ThreadFactory threadFactory, InterceptingTaskFactory taskFactory)
{
super(interceptorOfExecution, threadFactory, taskFactory);
@ -765,7 +770,7 @@ public interface InterceptingExecutor extends OrderOn
throw new RejectedExecutionException();
long delayNanos = unit.toNanos(delay);
return interceptorOfExecution.intercept().schedule(SCHEDULED_TASK, delayNanos, relativeToGlobalNanos(delayNanos), callable(run, null), this);
return schedule(SCHEDULED_TASK, delayNanos, relativeToGlobalNanos(delayNanos), callable(run, null));
}
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)
@ -774,7 +779,7 @@ public interface InterceptingExecutor extends OrderOn
throw new RejectedExecutionException();
long delayNanos = unit.toNanos(delay);
return interceptorOfExecution.intercept().schedule(SCHEDULED_TASK, delayNanos, relativeToGlobalNanos(delayNanos), callable, this);
return schedule(SCHEDULED_TASK, delayNanos, relativeToGlobalNanos(delayNanos), callable);
}
public ScheduledFuture<?> scheduleTimeoutWithDelay(Runnable run, long delay, TimeUnit unit)
@ -787,7 +792,7 @@ public interface InterceptingExecutor extends OrderOn
if (isShutdown)
throw new RejectedExecutionException();
return interceptorOfExecution.intercept().schedule(SCHEDULED_TASK, localToRelativeNanos(deadlineNanos), localToGlobalNanos(deadlineNanos), callable(run, null), this);
return schedule(SCHEDULED_TASK, localToRelativeNanos(deadlineNanos), localToGlobalNanos(deadlineNanos), callable(run, null));
}
public ScheduledFuture<?> scheduleTimeoutAt(Runnable run, long deadlineNanos)
@ -795,7 +800,7 @@ public interface InterceptingExecutor extends OrderOn
if (isShutdown)
throw new RejectedExecutionException();
return interceptorOfExecution.intercept().schedule(SCHEDULED_TIMEOUT, localToRelativeNanos(deadlineNanos), localToGlobalNanos(deadlineNanos), callable(run, null), this);
return schedule(SCHEDULED_TIMEOUT, localToRelativeNanos(deadlineNanos), localToGlobalNanos(deadlineNanos), callable(run, null));
}
public ScheduledFuture<?> scheduleSelfRecurring(Runnable run, long delay, TimeUnit unit)
@ -804,7 +809,7 @@ public interface InterceptingExecutor extends OrderOn
throw new RejectedExecutionException();
long delayNanos = unit.toNanos(delay);
return interceptorOfExecution.intercept().schedule(SCHEDULED_DAEMON, delayNanos, relativeToGlobalNanos(delayNanos), callable(run, null), this);
return schedule(SCHEDULED_DAEMON, delayNanos, relativeToGlobalNanos(delayNanos), callable(run, null));
}
public ScheduledFuture<?> scheduleAtFixedRate(Runnable run, long initialDelay, long period, TimeUnit unit)
@ -813,7 +818,7 @@ public interface InterceptingExecutor extends OrderOn
throw new RejectedExecutionException();
long delayNanos = unit.toNanos(initialDelay);
return interceptorOfExecution.intercept().schedule(SCHEDULED_DAEMON, delayNanos, relativeToGlobalNanos(delayNanos), new Callable<Object>()
return schedule(SCHEDULED_DAEMON, delayNanos, relativeToGlobalNanos(delayNanos), new Callable<Object>()
{
@Override
public Object call()
@ -829,7 +834,12 @@ public interface InterceptingExecutor extends OrderOn
{
return run.toString();
}
}, this);
});
}
<T> ScheduledFuture<T> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, Callable<T> task)
{
return interceptorOfExecution.intercept().schedule(kind, delayNanos, deadlineNanos, new InterceptableScheduledFutureTask<>(delayNanos, task), this);
}
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable run, long initialDelay, long delay, TimeUnit unit)
@ -843,6 +853,8 @@ public interface InterceptingExecutor extends OrderOn
}
}
@PerClassLoader
class InterceptingPooledLocalAwareExecutor extends InterceptingPooledExecutor implements LocalAwareExecutorPlus
{

4
test/simulator/main/org/apache/cassandra/simulator/systems/InterceptorOfExecution.java
View File

@ -18,10 +18,10 @@
package org.apache.cassandra.simulator.systems;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import java.util.function.Function;
import org.apache.cassandra.simulator.systems.InterceptingExecutor.InterceptableScheduledFuture;
import org.apache.cassandra.simulator.systems.SimulatedAction.Kind;
import org.apache.cassandra.utils.Shared;
import org.apache.cassandra.utils.concurrent.RunnableFuture;
@ -38,7 +38,7 @@ public interface InterceptorOfExecution
interface InterceptExecution
{
<V, T extends RunnableFuture<V>> T addTask(T task, InterceptingExecutor executor);
<T> ScheduledFuture<T> schedule(Kind kind, long delayNanos, long deadlineNanos, Callable<T> runnable, InterceptingExecutor executor);
<T> ScheduledFuture<T> schedule(Kind kind, long delayNanos, long deadlineNanos, InterceptableScheduledFuture<T> task, InterceptingExecutor executor);
Thread start(Kind kind, Function<Runnable, InterceptibleThread> factory, Runnable run);
}
}

7
test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java
View File

@ -29,7 +29,7 @@ import org.apache.cassandra.simulator.Actions;
import org.apache.cassandra.simulator.OrderOn;
import org.apache.cassandra.simulator.systems.InterceptedExecution.InterceptedFutureTaskExecution;
import org.apache.cassandra.simulator.systems.InterceptedExecution.InterceptedThreadStart;
import org.apache.cassandra.simulator.systems.InterceptingExecutor.InterceptedScheduledFutureTask;
import org.apache.cassandra.simulator.systems.InterceptingExecutor.InterceptableScheduledFuture;
import org.apache.cassandra.utils.concurrent.Condition;
import org.apache.cassandra.utils.concurrent.NotScheduledFuture;
import org.apache.cassandra.utils.concurrent.RunnableFuture;
@ -93,7 +93,7 @@ public class SimulatedExecution implements InterceptorOfExecution
return task;
}
public <T> ScheduledFuture<T> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, Callable<T> runnable, InterceptingExecutor executor)
public <T> ScheduledFuture<T> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, InterceptableScheduledFuture<T> task, InterceptingExecutor executor)
{
return new NotScheduledFuture<>();
}
@ -135,10 +135,9 @@ public class SimulatedExecution implements InterceptorOfExecution
return task;
}
public <V> ScheduledFuture<V> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, Callable<V> call, InterceptingExecutor executor)
public <V> ScheduledFuture<V> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, InterceptableScheduledFuture<V> task, InterceptingExecutor executor)
{
assert kind == SCHEDULED_TASK || kind == SCHEDULED_TIMEOUT || kind == SCHEDULED_DAEMON;
InterceptedScheduledFutureTask<V> task = new InterceptedScheduledFutureTask<>(delayNanos, call);
InterceptedFutureTaskExecution<?> intercepted = new InterceptedFutureTaskExecution<>(kind, executor, task, deadlineNanos);
task.onCancel(intercepted::cancel);
intercept.interceptExecution(intercepted, executor.orderAppliesAfterScheduling());