This is an automated email from the ASF dual-hosted git repository. ifesdjeen pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/cassandra-simulator.git
commit bd0a06a0660a781646d469ed2d1c78d5ac85735b Author: Alex Petrov <[email protected]> AuthorDate: Mon Jul 20 16:18:06 2026 +0200 Improve/simplify per step sessions, get rid of singletons --- .../InstrumentedCassandraCountDownLatch.java | 5 +- .../simulator/systems/InstrumentedSemaphore.java | 5 +- .../simulator/systems/InstrumentedWaitQueue.java | 11 ++--- .../org/apache/cassandra/simulator/Simulator.java | 38 +++++++++++---- .../simulator/step/ObservableActions.java | 28 +++++------ .../simulator/systems/InterceptibleThread.java | 23 +++++++-- .../systems/InterceptingCountDownLatch.java | 7 ++- .../systems/InterceptingGlobalMethods.java | 13 ++--- .../simulator/systems/InterceptingMonitors.java | 13 +++-- .../systems/InterceptorOfConsequences.java | 15 +++++- .../systems/InterceptorOfObservations.java | 57 ++++++++++++++++++++++ .../simulator/systems/SimulatedAction.java | 30 +++++++++--- .../simulator/systems/SimulatedActionTask.java | 9 ++-- .../simulator_test/ObservableActionsTest.java | 48 ++++++++++++++++++ .../simulator_test/SimulatorSessionTest.java | 51 +++++++++++++++++++ 15 files changed, 280 insertions(+), 73 deletions(-) diff --git a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedCassandraCountDownLatch.java b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedCassandraCountDownLatch.java index bf30466..542ae0a 100644 --- a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedCassandraCountDownLatch.java +++ b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedCassandraCountDownLatch.java @@ -20,7 +20,6 @@ package org.apache.cassandra.simulator.systems; import java.util.concurrent.atomic.AtomicInteger; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.utils.concurrent.Awaitable; import org.apache.cassandra.utils.concurrent.CassandraCountDownLatch; @@ -40,13 +39,13 @@ public class InstrumentedCassandraCountDownLatch extends InterceptingAwaitable.I public Awaitable await() throws InterruptedException { - ObservableActions.emitCustom(AWAIT, null, Integer.toString(count.get())); + InterceptorOfObservations.observeCurrentCustom(AWAIT, null, Integer.toString(count.get())); return super.await(); } public void decrement() { - ObservableActions.emitCustom(DECREMENT, null, Integer.toString(count.get())); + InterceptorOfObservations.observeCurrentCustom(DECREMENT, null, Integer.toString(count.get())); if (count.decrementAndGet() == 0) signal(); } diff --git a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedSemaphore.java b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedSemaphore.java index 2978bbc..856188f 100644 --- a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedSemaphore.java +++ b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedSemaphore.java @@ -23,7 +23,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.utils.UncheckedInterruptedException; import org.apache.cassandra.utils.concurrent.Semaphore; @@ -89,7 +88,7 @@ public class InstrumentedSemaphore extends Semaphore.Standard return; } - ObservableActions.emitCustom(RELEASE, null, Integer.toString(release)); + InterceptorOfObservations.observeCurrentCustom(RELEASE, null, Integer.toString(release)); int remaining = permits.addAndGet(release); while (!interceptible.isEmpty() && remaining > 0) { @@ -175,7 +174,7 @@ public class InstrumentedSemaphore extends Semaphore.Standard return; } - ObservableActions.emitCustom(ACQUIRE, null, Integer.toString(acquire)); + InterceptorOfObservations.observeCurrentCustom(ACQUIRE, null, Integer.toString(acquire)); while (true) { if (tryAcquire(acquire)) diff --git a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedWaitQueue.java b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedWaitQueue.java index db678bc..61608c6 100644 --- a/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedWaitQueue.java +++ b/integration-test/src/main/java/org/apache/cassandra/simulator/systems/InstrumentedWaitQueue.java @@ -23,7 +23,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.function.Consumer; import java.util.function.Predicate; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.systems.InterceptingAwaitable.InterceptingSignal; import org.apache.cassandra.utils.concurrent.WaitQueue; @@ -52,7 +51,7 @@ public class InstrumentedWaitQueue extends WaitQueue.Standard implements WaitQue @Override public void cancel() { - ObservableActions.emitCustom(CANCEL, null, null); + InterceptorOfObservations.observeCurrentCustom(CANCEL, null, null); super.cancel(); } } @@ -64,7 +63,7 @@ public class InstrumentedWaitQueue extends WaitQueue.Standard implements WaitQue InstrumentedSignal<?> signal = new InstrumentedSignal<>(); interceptible.add(signal); - ObservableActions.emitCustom(REGISTER, null, Integer.toString(getWaiting())); + InterceptorOfObservations.observeCurrentCustom(REGISTER, null, Integer.toString(getWaiting())); return signal; } @@ -75,19 +74,19 @@ public class InstrumentedWaitQueue extends WaitQueue.Standard implements WaitQue InstrumentedSignal<V> signal = new InstrumentedSignal<>(value, consumer); interceptible.add(signal); - ObservableActions.emitCustom(REGISTER, null, Integer.toString(getWaiting())); + InterceptorOfObservations.observeCurrentCustom(REGISTER, null, Integer.toString(getWaiting())); return signal; } public boolean signal() { - ObservableActions.emitCustom(SIGNAL, null, Integer.toString(getWaiting())); + InterceptorOfObservations.observeCurrentCustom(SIGNAL, null, Integer.toString(getWaiting())); return consumeUntil(InstrumentedSignal::doSignal) || super.signal(); } public void signalAll() { - ObservableActions.emitCustom(SIGNAL_ALL, null, Integer.toString(getWaiting())); + InterceptorOfObservations.observeCurrentCustom(SIGNAL_ALL, null, Integer.toString(getWaiting())); consumeUntil(s -> { s.signal(); return false; diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/Simulator.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/Simulator.java index 4d09335..8ff5ee4 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/Simulator.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/Simulator.java @@ -110,6 +110,7 @@ public class Simulator implements AutoCloseable private final float monitorDelayChance; private final Set<InterceptRule> customRules = new LinkedHashSet<>(); private final List<DeferredReg> deferredRegistrations = new ArrayList<>(); + private volatile SessionImpl activeSession; private State state = State.NEW; /** Deferred factory registration: resolved against the simulation classloader at run-time. */ @@ -447,8 +448,9 @@ public class Simulator implements AutoCloseable public int simulate(IIsolatedExecutor.SerializableRunnable... bodies) { ensureStarted(); + ensureNoActiveSession("simulate"); - ActionSchedule schedule = buildSchedule(bodies); + ActionSchedule schedule = buildSchedule(new ObservableActions(), bodies); int count = 0; while (schedule.hasNext()) { @@ -498,21 +500,34 @@ public class Simulator implements AutoCloseable * * @param bodies serializable runnables to execute as simulated threads * @return a session that can be stepped and closed + * @throws IllegalStateException if another step-by-step session is active */ public Session byStep(IIsolatedExecutor.SerializableRunnable... bodies) { ensureStarted(); - return new SessionImpl(buildSchedule(bodies)); + ensureNoActiveSession("byStep"); + ObservableActions observableActions = new ObservableActions(); + SessionImpl session = new SessionImpl(buildSchedule(observableActions, bodies), observableActions); + activeSession = session; + return session; + } + + private void ensureNoActiveSession(String operation) + { + if (activeSession != null) + throw new IllegalStateException("Cannot call " + operation + " while a step-by-step session is active"); } private final class SessionImpl implements Session { private final ActionSchedule schedule; + private final ObservableActions observableActions; private ObservableActions.Sink observations; - SessionImpl(ActionSchedule schedule) + SessionImpl(ActionSchedule schedule, ObservableActions observableActions) { this.schedule = schedule; + this.observableActions = observableActions; } @Override @@ -530,14 +545,14 @@ public class Simulator implements AutoCloseable { if (observations == null) observations = new ObservableActions.Sink(); - ObservableActions.enable(observations); + observableActions.enable(observations); return this; } @Override public Session disableDebug() { - ObservableActions.disable(); + observableActions.disable(); return this; } @@ -661,7 +676,7 @@ public class Simulator implements AutoCloseable } } - private ActionSchedule buildSchedule(IIsolatedExecutor.SerializableRunnable... bodies) + private ActionSchedule buildSchedule(ObservableActions observableActions, IIsolatedExecutor.SerializableRunnable... bodies) { URL[] urls = buildClasspath(); @@ -707,7 +722,7 @@ public class Simulator implements AutoCloseable for (int i = 0; i < bodies.length; i++) { Runnable transferred = transfer.apply(bodies[i]); - actions.add(threadAction("-" + i, transferred)); + actions.add(threadAction("-" + i, transferred, observableActions)); } return new ActionSchedule(simulatedTime, @@ -729,6 +744,7 @@ public class Simulator implements AutoCloseable public int simulate(List<Thread> threads) { ensureStarted(); + ensureNoActiveSession("simulate"); final Field targetField; try { @@ -740,6 +756,7 @@ public class Simulator implements AutoCloseable throw new RuntimeException("Cannot access Thread.target - unexpected JDK version?", e); } + ObservableActions observableActions = new ObservableActions(); List<Action> actions = new ArrayList<>(threads.size()); for (Thread thread : threads) { @@ -754,7 +771,7 @@ public class Simulator implements AutoCloseable } if (body == null) throw new IllegalArgumentException("Thread has no runnable target: " + thread.getName()); - actions.add(threadAction(thread.getName(), body)); + actions.add(threadAction(thread.getName(), body, observableActions)); } return executeSchedule(ActionList.of(actions), Integer.MAX_VALUE, new RunnableActionScheduler.Sequential()); @@ -812,7 +829,7 @@ public class Simulator implements AutoCloseable return count; } - private Action threadAction(String name, Runnable body) + private Action threadAction(String name, Runnable body, ObservableActions observableActions) { InterceptedExecution.InterceptedThreadStart start = new InterceptedExecution.InterceptedThreadStart( @@ -828,7 +845,8 @@ public class Simulator implements AutoCloseable NONE, simulatedTime, debug, - start + start, + observableActions ); } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableActions.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableActions.java index 62b117f..d022962 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableActions.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableActions.java @@ -42,44 +42,38 @@ import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; @Shared(scope = SIMULATION, ancestors = ALL, members = ALL) public final class ObservableActions { - private static volatile Sink sink; + private volatile Sink sink; - private ObservableActions() + public void enable(Sink sink) { + this.sink = Objects.requireNonNull(sink, "sink"); } - public static void enable(Sink sink) - { - ObservableActions.sink = sink; - } - - public static void disable() + public void disable() { sink = null; } - public static boolean isEnabled() + public boolean isEnabled() { return sink != null; } - public static void emit(ObservableAction.Kind kind) + public void emit(ObservableAction.Kind kind) { - Sink sink = ObservableActions.sink; - if (isEnabled()) - sink.emit(kind, null, Thread.currentThread(), null, null); + emit(kind, null, null); } - public static void emit(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail) + public void emit(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail) { - Sink sink = ObservableActions.sink; + Sink sink = this.sink; if (sink != null) sink.emit(kind, null, Thread.currentThread(), target, detail); } - public static void emitCustom(String name, @Nullable Thread target, @Nullable String detail) + public void emitCustom(String name, @Nullable Thread target, @Nullable String detail) { - Sink sink = ObservableActions.sink; + Sink sink = this.sink; if (sink != null) sink.emit(ObservableAction.Kind.CUSTOM, Objects.requireNonNull(name, "name"), Thread.currentThread(), target, detail); diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptibleThread.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptibleThread.java index 43ea024..7f262ee 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptibleThread.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptibleThread.java @@ -24,7 +24,6 @@ import java.util.concurrent.locks.LockSupport; import org.apache.cassandra.simulator.OrderOn; import org.apache.cassandra.simulator.step.ObservableAction; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.systems.InterceptedWait.Trigger; import org.apache.cassandra.simulator.systems.SimulatedTime.LocalTime; import org.apache.cassandra.simulator.utils.Shared; @@ -252,7 +251,7 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc } Parked parked = new Parked(kind, interceptorOfGlobalMethods.captureWaitSite(this), waitTime, interceptor); - ObservableActions.emit(ObservableAction.Kind.PARK_CAPTURED, null, kind.name()); + observe(ObservableAction.Kind.PARK_CAPTURED, null, kind.name()); this.parked = parked; interceptWait(parked); parked.await(); @@ -263,11 +262,11 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc public boolean unpark(InterceptibleThread by) { if (by.interceptor == null) return false; - ObservableActions.emit(ObservableAction.Kind.UNPARK_CAPTURED, this, null); + by.observe(ObservableAction.Kind.UNPARK_CAPTURED, this, null); if (parked == null) hasPendingUnpark = true; else { - ObservableActions.emit(ObservableAction.Kind.WAKEUP_ENQUEUED, this, SIGNAL.name()); + by.observe(ObservableAction.Kind.WAKEUP_ENQUEUED, this, SIGNAL.name()); parked.interceptWakeup(SIGNAL, by); } return true; @@ -321,6 +320,22 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc throw new UnsupportedOperationException(); } + @Override + public void observe(ObservableAction.Kind kind, Thread target, String detail) + { + InterceptorOfConsequences interceptor = this.interceptor; + if (interceptor != null) + interceptor.observe(kind, target, detail); + } + + @Override + public void observeCustom(String name, Thread target, String detail) + { + InterceptorOfConsequences interceptor = this.interceptor; + if (interceptor != null) + interceptor.observeCustom(name, target, detail); + } + @Override public void interceptWakeup(InterceptedWait wakeup, Trigger trigger, InterceptorOfConsequences waitWasInterceptedBy) { diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingCountDownLatch.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingCountDownLatch.java index 9427810..6009924 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingCountDownLatch.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingCountDownLatch.java @@ -22,7 +22,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.cassandra.simulator.step.ObservableAction; -import org.apache.cassandra.simulator.step.ObservableActions; @PerClassLoader public class InterceptingCountDownLatch extends java.util.concurrent.CountDownLatch @@ -40,21 +39,21 @@ public class InterceptingCountDownLatch extends java.util.concurrent.CountDownLa @Override public void await() throws InterruptedException { - ObservableActions.emit(ObservableAction.Kind.LATCH_AWAIT_CAPTURED, null, Integer.toString(count.get())); + InterceptorOfObservations.observeCurrent(ObservableAction.Kind.LATCH_AWAIT_CAPTURED, null, Integer.toString(count.get())); inner.await(); } @Override public boolean await(long timeout, TimeUnit unit) throws InterruptedException { - ObservableActions.emit(ObservableAction.Kind.LATCH_AWAIT_CAPTURED, null, Integer.toString(count.get())); + InterceptorOfObservations.observeCurrent(ObservableAction.Kind.LATCH_AWAIT_CAPTURED, null, Integer.toString(count.get())); inner.await(); return true; } public void decrement() { - ObservableActions.emit(ObservableAction.Kind.LATCH_COUNTDOWN_CAPTURED, null, Integer.toString(count.get())); + InterceptorOfObservations.observeCurrent(ObservableAction.Kind.LATCH_COUNTDOWN_CAPTURED, null, Integer.toString(count.get())); if (count.decrementAndGet() == 0) inner.signal(); } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingGlobalMethods.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingGlobalMethods.java index d631dde..0eb2925 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingGlobalMethods.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingGlobalMethods.java @@ -29,7 +29,6 @@ import javax.annotation.Nullable; import org.apache.cassandra.simulator.RandomSource; import org.apache.cassandra.simulator.step.ObservableAction; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture; import org.apache.cassandra.simulator.systems.InterceptedWait.InterceptedConditionWait; import org.apache.cassandra.simulator.utils.Clock; @@ -75,7 +74,9 @@ public class InterceptingGlobalMethods extends InterceptingMonitors implements I thread.start(); return; } - ObservableActions.emit(ObservableAction.Kind.THREAD_STARTED, thread, null); + + InterceptibleThread cast = (InterceptibleThread) current; + cast.observe(ObservableAction.Kind.THREAD_STARTED, thread, null); executorFactory.startThread(thread.getName(), thread, InterceptingExecutorFactory.SystemThreadTag.NON_DAEMON, InterceptingExecutorFactory.SimulatorThreadTag.JOB); @@ -84,19 +85,19 @@ public class InterceptingGlobalMethods extends InterceptingMonitors implements I @Override public ExecutorService newFixedThreadPool(int threads) { - return (ExecutorService) executorFactory.pooled("jdk-pool-" + poolCounter.getAndIncrement(), threads); + return executorFactory.pooled("jdk-pool-" + poolCounter.getAndIncrement(), threads); } @Override public ExecutorService newSingleThreadExecutor() { - return (ExecutorService) executorFactory.pooled("jdk-single-" + poolCounter.getAndIncrement(), 1); + return executorFactory.pooled("jdk-single-" + poolCounter.getAndIncrement(), 1); } @Override public ExecutorService newCachedThreadPool() { - return (ExecutorService) executorFactory.sequential("jdk-cached-" + poolCounter.getAndIncrement()); + return executorFactory.sequential("jdk-cached-" + poolCounter.getAndIncrement()); } @Override @@ -166,7 +167,7 @@ public class InterceptingGlobalMethods extends InterceptingMonitors implements I if (thread == null || thread.isEvaluationDeterministic()) return; - ObservableActions.emit(ObservableAction.Kind.PRE_IO_CAPTURED); + thread.observe(ObservableAction.Kind.PRE_IO_CAPTURED); pause(thread, PRE_IO); } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingMonitors.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingMonitors.java index ffb0b8c..ca38acf 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingMonitors.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingMonitors.java @@ -31,7 +31,6 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.simulator.RandomSource; import org.apache.cassandra.simulator.step.ObservableAction; -import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.systems.InterceptedWait.InterceptedConditionWait; import org.apache.cassandra.simulator.utils.Clock; import org.apache.cassandra.simulator.utils.Closeable; @@ -518,7 +517,7 @@ public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods throw new InterruptedException(); InterceptedMonitorWait trigger = new InterceptedMonitorWait(SLEEP_UNTIL, deadline, new MonitorState(), thread, captureWaitSite(thread)); - ObservableActions.emit(ObservableAction.Kind.SLEEP_CAPTURED, null, Long.toString(deadline)); + thread.observe(ObservableAction.Kind.SLEEP_CAPTURED, null, Long.toString(deadline)); thread.interceptWait(trigger); synchronized (trigger) { @@ -607,7 +606,7 @@ public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods + " which does not hold the monitor; held by " + (state.heldBy == null ? "nobody" : state.heldBy)); InterceptedMonitorWait trigger = new InterceptedMonitorWait(kind, waitNanos, state, thread, captureWaitSite(thread), monitor); - ObservableActions.emit(ObservableAction.Kind.WAIT_CAPTURED, null, kind.name()); + thread.observe(ObservableAction.Kind.WAIT_CAPTURED, null, kind.name()); state.log("enterwait", thread); state.suspend(trigger); state.waitOn(NOTIFY, trigger); @@ -640,8 +639,8 @@ public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods // TODO: assign ownership on monitorExit assert wake.waitingOn == null; Thread waker = Thread.currentThread(); - ObservableActions.emit(ObservableAction.Kind.NOTIFY_CAPTURED, wake.waiting, null); - ObservableActions.emit(ObservableAction.Kind.WAKEUP_ENQUEUED, wake.waiting, SIGNAL.name()); + thread.observe(ObservableAction.Kind.NOTIFY_CAPTURED, wake.waiting, null); + thread.observe(ObservableAction.Kind.WAKEUP_ENQUEUED, wake.waiting, SIGNAL.name()); wake.interceptWakeup(SIGNAL, waker); state.log("notify", wake.waiting, waker); return; @@ -664,14 +663,14 @@ public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods { NOTIFY.setHead(state, null); Thread waker = Thread.currentThread(); - ObservableActions.emit(ObservableAction.Kind.NOTIFY_CAPTURED, head.waiting, "all"); + InterceptorOfObservations.observeCurrent(ObservableAction.Kind.NOTIFY_CAPTURED, head.waiting, "all"); // Signal the first waiter directly InterceptedMonitorWait cur = head; InterceptedMonitorWait next = cur.next; cur.next = null; cur.waitingOn = null; - ObservableActions.emit(ObservableAction.Kind.WAKEUP_ENQUEUED, cur.waiting, SIGNAL.name()); + InterceptorOfObservations.observeCurrent(ObservableAction.Kind.WAKEUP_ENQUEUED, cur.waiting, SIGNAL.name()); cur.interceptWakeup(SIGNAL, waker); state.log("notify", cur.waiting, waker); diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfConsequences.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfConsequences.java index b98df0c..d1fb54c 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfConsequences.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfConsequences.java @@ -18,14 +18,17 @@ package org.apache.cassandra.simulator.systems; +import javax.annotation.Nullable; + import org.apache.cassandra.simulator.OrderOn; +import org.apache.cassandra.simulator.step.ObservableAction; import org.apache.cassandra.simulator.systems.InterceptedWait.Trigger; import org.apache.cassandra.simulator.utils.Shared; import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; @Shared(scope = SIMULATION) -public interface InterceptorOfConsequences +public interface InterceptorOfConsequences extends InterceptorOfObservations { public static final InterceptorOfConsequences DEFAULT_INTERCEPTOR = new InterceptorOfConsequences() { @@ -34,6 +37,16 @@ public interface InterceptorOfConsequences { } + @Override + public void observe(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail) + { + } + + @Override + public void observeCustom(String name, @Nullable Thread target, @Nullable String detail) + { + } + @Override public void interceptWait(InterceptedWait wakeupWith) { diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfObservations.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfObservations.java new file mode 100644 index 0000000..aaae8f9 --- /dev/null +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfObservations.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.simulator.systems; + +import javax.annotation.Nullable; + +import org.apache.cassandra.simulator.step.ObservableAction; +import org.apache.cassandra.simulator.utils.Shared; + +import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; + +/** + * Receives observations produced in the dynamic scope of a simulated action. + * Observations are different from consequences in that they have no material effect on the rest of the system. + **/ +@Shared(scope = SIMULATION) +public interface InterceptorOfObservations +{ + static void observeCurrent(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail) + { + Thread thread = Thread.currentThread(); + if (thread instanceof InterceptibleThread) + ((InterceptibleThread) thread).observe(kind, target, detail); + } + + static void observeCurrentCustom(String name, @Nullable Thread target, @Nullable String detail) + { + Thread thread = Thread.currentThread(); + if (thread instanceof InterceptibleThread) + ((InterceptibleThread) thread).observeCustom(name, target, detail); + } + + default void observe(ObservableAction.Kind kind) + { + observe(kind, null, null); + } + + void observe(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail); + + void observeCustom(String name, @Nullable Thread target, @Nullable String detail); +} \ No newline at end of file diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedAction.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedAction.java index 47e06a6..9502664 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedAction.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedAction.java @@ -129,10 +129,10 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon @Override protected ActionList performSimple() { - ObservableActions.emit(trigger == TIMEOUT || wakeup.kind() == InterceptedWait.Kind.SLEEP_UNTIL - ? ObservableAction.Kind.TIMEOUT_FIRED - : ObservableAction.Kind.WAKEUP_FIRED, - wakeup.waiting(), trigger.name()); + observe(trigger == TIMEOUT || wakeup.kind() == InterceptedWait.Kind.SLEEP_UNTIL + ? ObservableAction.Kind.TIMEOUT_FIRED + : ObservableAction.Kind.WAKEUP_FIRED, + wakeup.waiting(), trigger.name()); return simulate(() -> wakeup.triggerAndAwaitDone(SimulatedAction.this, trigger)); } @@ -147,6 +147,7 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon protected final SimulatedAction.Kind kind; protected final SimulatedTime time; protected final Debug debug; + private final ObservableActions observableActions; private InterceptibleThread realThread; // unset until first simulation @@ -154,12 +155,12 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon private @Nullable InterceptedWait pausedOn; private boolean realThreadHasTerminated; - public SimulatedAction(Object description, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug) + public SimulatedAction(Object description, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, ObservableActions observableActions) { - this(description, Kind.TASK, OrderOn.NONE, self, transitive, time, debug); + this(description, Kind.TASK, OrderOn.NONE, self, transitive, time, debug, observableActions); } - public SimulatedAction(Object description, Kind kind, OrderOn orderOn, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug) + public SimulatedAction(Object description, Kind kind, OrderOn orderOn, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, ObservableActions observableActions) { super(description, orderOn, self, transitive); Preconditions.checkNotNull(kind); @@ -168,6 +169,19 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon this.kind = kind; this.time = time; this.debug = debug; + this.observableActions = Preconditions.checkNotNull(observableActions); + } + + @Override + public void observe(ObservableAction.Kind kind, @Nullable Thread target, @Nullable String detail) + { + observableActions.emit(kind, target, detail); + } + + @Override + public void observeCustom(String name, @Nullable Thread target, @Nullable String detail) + { + observableActions.emitCustom(name, target, detail); } @Override @@ -309,7 +323,7 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon Action applyTo(Object description, Kind kind, OrderOn orderOn, Modifiers self, InterceptedExecution task) { - return new SimulatedActionTask(description, kind, orderOn, self, NONE, time, debug, task); + return new SimulatedActionTask(description, kind, orderOn, self, NONE, time, debug, task, observableActions); } public Object description() diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedActionTask.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedActionTask.java index 79196f7..52ea6e2 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedActionTask.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedActionTask.java @@ -21,6 +21,7 @@ package org.apache.cassandra.simulator.systems; import org.apache.cassandra.simulator.ActionList; import org.apache.cassandra.simulator.Debug; import org.apache.cassandra.simulator.OrderOn; +import org.apache.cassandra.simulator.step.ObservableActions; import org.apache.cassandra.simulator.utils.Throwables; import static org.apache.cassandra.simulator.systems.SimulatedAction.Kind.TASK; @@ -29,14 +30,14 @@ public class SimulatedActionTask extends SimulatedAction implements Runnable { InterceptedExecution task; - public SimulatedActionTask(Object description, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, InterceptedExecution task) + public SimulatedActionTask(Object description, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, InterceptedExecution task, ObservableActions observableActions) { - this(description, TASK, OrderOn.NONE, self, transitive, time, debug, task); + this(description, TASK, OrderOn.NONE, self, transitive, time, debug, task, observableActions); } - public SimulatedActionTask(Object description, Kind kind, OrderOn orderOn, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, InterceptedExecution task) + public SimulatedActionTask(Object description, Kind kind, OrderOn orderOn, Modifiers self, Modifiers transitive, SimulatedTime time, Debug debug, InterceptedExecution task, ObservableActions observableActions) { - super(description, kind, orderOn, self, transitive, time, debug); + super(description, kind, orderOn, self, transitive, time, debug, observableActions); this.task = task; task.onCancel(this); } diff --git a/simulator-core/src/test/java/org/apache/cassandra/simulator_test/ObservableActionsTest.java b/simulator-core/src/test/java/org/apache/cassandra/simulator_test/ObservableActionsTest.java new file mode 100644 index 0000000..754c89a --- /dev/null +++ b/simulator-core/src/test/java/org/apache/cassandra/simulator_test/ObservableActionsTest.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.simulator_test; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.simulator.step.ObservableAction; +import org.apache.cassandra.simulator.step.ObservableActions; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class ObservableActionsTest +{ + @Test + void channelsAreIndependent() + { + ObservableActions first = new ObservableActions(); + ObservableActions second = new ObservableActions(); + ObservableActions.Sink firstSink = new ObservableActions.Sink(); + ObservableActions.Sink secondSink = new ObservableActions.Sink(); + + first.enable(firstSink); + second.enable(secondSink); + first.emit(ObservableAction.Kind.PRE_IO_CAPTURED); + second.emit(ObservableAction.Kind.THREAD_STARTED); + first.disable(); + second.emit(ObservableAction.Kind.PARK_CAPTURED); + + assertEquals(1, firstSink.drainObservedActions().size()); + assertEquals(2, secondSink.drainObservedActions().size()); + } +} diff --git a/simulator-core/src/test/java/org/apache/cassandra/simulator_test/SimulatorSessionTest.java b/simulator-core/src/test/java/org/apache/cassandra/simulator_test/SimulatorSessionTest.java new file mode 100644 index 0000000..798f8ea --- /dev/null +++ b/simulator-core/src/test/java/org/apache/cassandra/simulator_test/SimulatorSessionTest.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.simulator_test; + +import org.apache.cassandra.simulator.Simulator; +import org.apache.cassandra.simulator.context.IIsolatedExecutor.SerializableRunnable; +import org.apache.cassandra.simulator.step.Session; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class SimulatorSessionTest +{ + @Test + void rejectsOverlappingStepSessions() + { + try (Simulator simulator = new Simulator(1L); + Session ignored = simulator.byStep((SerializableRunnable) () -> {})) + { + assertThrows(IllegalStateException.class, + () -> simulator.byStep((SerializableRunnable) () -> {})); + } + } + + @Test + void rejectsDirectSimulationWhileStepSessionIsActive() + { + try (Simulator simulator = new Simulator(1L); + Session ignored = simulator.byStep((SerializableRunnable) () -> {})) + { + assertThrows(IllegalStateException.class, + () -> simulator.simulate((SerializableRunnable) () -> {})); + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
