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 52329fc7facdc824ac4f0556f4fc9fd0e5dd58a2 Author: Alex Petrov <[email protected]> AuthorDate: Fri Jul 17 21:12:19 2026 +0200 Rough prototype of the pre-IO step --- .../cassandra/simulator_test/IoStepApiTest.java | 211 +++++++++++++++++++-- .../simulator/context/SharedTestState.java | 7 + .../org/apache/cassandra/simulator/Simulator.java | 2 +- .../cassandra/simulator/step/ObservableAction.java | 1 + .../simulator/systems/InterceptedWait.java | 23 ++- .../systems/InterceptingGlobalMethods.java | 21 +- .../systems/InterceptorOfSystemMethods.java | 12 ++ .../simulator/systems/SimulatedAction.java | 7 +- .../cassandra/simulator/io/InstrumentedFile.java | 153 +++++++++++++++ .../simulator/io/IoOperationListener.java | 34 ++++ 10 files changed, 446 insertions(+), 25 deletions(-) diff --git a/integration-test/src/test/java/org/apache/cassandra/simulator_test/IoStepApiTest.java b/integration-test/src/test/java/org/apache/cassandra/simulator_test/IoStepApiTest.java index a9f62dd..be4490b 100644 --- a/integration-test/src/test/java/org/apache/cassandra/simulator_test/IoStepApiTest.java +++ b/integration-test/src/test/java/org/apache/cassandra/simulator_test/IoStepApiTest.java @@ -20,20 +20,29 @@ package org.apache.cassandra.simulator_test; import java.nio.file.Files; import java.nio.file.Path; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; import org.apache.cassandra.simulator.io.InstrumentedFile; import org.apache.cassandra.simulator.io.IoOperationListener; import org.apache.cassandra.simulator.Simulator; import org.apache.cassandra.simulator.context.SharedTestState; import org.apache.cassandra.simulator.context.IIsolatedExecutor.SerializableRunnable; +import org.apache.cassandra.simulator.step.ObservableAction; import org.apache.cassandra.simulator.step.Session; import org.apache.cassandra.simulator.systems.InstrumentedCassandraCountDownLatch; +import org.apache.cassandra.simulator.systems.InterceptorOfSystemMethods; import org.apache.cassandra.utils.concurrent.CassandraCountDownLatch; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; public class IoStepApiTest @@ -45,19 +54,25 @@ public class IoStepApiTest InstrumentedFile.setListener(new IoOperationListener() { @Override - public void beforeIoStep() + public void beforePreIoPark() { SharedTestState.parkCount.incrementAndGet(); } @Override - public void afterIoStep() + public void afterPreIoPark() { SharedTestState.unparkCount.incrementAndGet(); } }); } + @AfterEach + void clearListener() + { + InstrumentedFile.setListener(null); + } + private static Simulator simulator(long seed) { Simulator sim = new Simulator(seed); @@ -89,26 +104,188 @@ public class IoStepApiTest }, "observer").start(); try { done.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - })) + }).enableDebug()) { - assertEquals(0, SharedTestState.parkCount.get(), "step 0: no I/O boundary reached"); - assertEquals(0, SharedTestState.eventCount.get(), "step 0: exists() has not completed"); + ObservableAction preIo = session.stepUntil(ObservableAction.Kind.PRE_IO_CAPTURED).orElseThrow(); + assertTrue(preIo.threadName().contains("exists")); + assertEquals(0, SharedTestState.eventCount.get(), "host I/O has not executed"); - assertTrue(session.stepThrough(1)); - assertEquals(0, SharedTestState.parkCount.get(), "step 1: threads queued, not run"); + ObservableAction observer = session.stepUntil(InstrumentedCassandraCountDownLatch.DECREMENT).orElseThrow(); + assertTrue(observer.threadName().contains("observer")); + assertEquals(0, SharedTestState.eventCount.get(), "another runnable executed before the continuation"); - assertTrue(session.stepThrough(1)); - assertEquals(1, SharedTestState.parkCount.get(), "step 2: exists() reached an I/O step"); - assertEquals(0, SharedTestState.unparkCount.get(), "step 2: exists() is parked before host Files.exists"); - assertEquals(0, SharedTestState.eventCount.get(), "step 2: exists() completion is not visible"); + ObservableAction resumed = session.stepUntil(ObservableAction.Kind.WAKEUP_FIRED).orElseThrow(); + assertEquals(Long.valueOf(preIo.threadId()), resumed.targetThreadId()); + assertEquals(1, SharedTestState.eventCount.get(), "host I/O executed after the continuation fired"); + } + } + finally + { + Files.deleteIfExists(path); + } + } - assertTrue(session.stepThrough(1)); - assertEquals(1, SharedTestState.threadRan.get(), "step 3: another runnable ran while I/O was parked"); - assertEquals(0, SharedTestState.eventCount.get(), "step 3: exists() is still not complete"); + @Test + void immediateContinuationIsValid() throws Exception + { + Path path = Files.createTempFile("sim-io-immediate", ".tmp"); + try + { + String pathString = path.toString(); + try (Simulator sim = simulator(42L); + Session session = sim.byStep((SerializableRunnable) () -> { + if (new InstrumentedFile(Path.of(pathString)).exists()) + SharedTestState.eventCount.incrementAndGet(); + }).enableDebug()) + { + ObservableAction preIo = session.stepUntil(ObservableAction.Kind.PRE_IO_CAPTURED).orElseThrow(); + assertEquals(0, SharedTestState.eventCount.get()); - assertTrue(session.stepThrough(1)); - assertEquals(1, SharedTestState.unparkCount.get(), "step 4: exists() resumed after the I/O step"); - assertEquals(1, SharedTestState.eventCount.get(), "step 4: exists() result is now visible"); + ObservableAction resumed = session.stepUntil(ObservableAction.Kind.WAKEUP_FIRED).orElseThrow(); + assertEquals(Long.valueOf(preIo.threadId()), resumed.targetThreadId()); + assertEquals(1, SharedTestState.eventCount.get()); + assertFalse(session.hasNext()); + } + } + finally + { + Files.deleteIfExists(path); + } + } + + @Test + void continuationPreservesThreadAndInterruptStatus() throws Exception + { + Path path = Files.createTempFile("sim-io-interrupt", ".tmp"); + AtomicReference<Thread> before = new AtomicReference<>(); + AtomicReference<Thread> after = new AtomicReference<>(); + InstrumentedFile.setListener(new IoOperationListener() + { + @Override + public void beforePreIoPark() + { + before.set(Thread.currentThread()); + } + + @Override + public void afterPreIoPark() + { + after.set(Thread.currentThread()); + } + }); + + try + { + String pathString = path.toString(); + try (Simulator sim = simulator(42L); + Session session = sim.byStep((SerializableRunnable) () -> { + Thread.currentThread().interrupt(); + new InstrumentedFile(Path.of(pathString)).exists(); + if (Thread.currentThread().isInterrupted()) + SharedTestState.eventCount.incrementAndGet(); + })) + { + assertTrue(session.step()); + assertNotNull(before.get()); + assertNotSame(Thread.currentThread(), before.get()); + assertNull(after.get()); + + assertTrue(session.step()); + assertSame(before.get(), after.get()); + assertEquals(1, SharedTestState.eventCount.get()); + } + } + finally + { + Files.deleteIfExists(path); + } + } + + @Test + void preIoConsumesNoRandomDecision() throws Exception + { + long seed = 42L; + try (Simulator sim = simulator(seed)) + { + sim.simulate((SerializableRunnable) () -> { + UUID uuid = InterceptorOfSystemMethods.Global.randomUUID(); + SharedTestState.value1.set(uuid.getMostSignificantBits()); + SharedTestState.value2.set(uuid.getLeastSignificantBits()); + }); + } + UUID expected = new UUID(SharedTestState.value1.get(), SharedTestState.value2.get()); + + SharedTestState.reset(); + Path path = Files.createTempFile("sim-io-random", ".tmp"); + try + { + String pathString = path.toString(); + try (Simulator sim = simulator(seed)) + { + sim.simulate((SerializableRunnable) () -> { + new InstrumentedFile(Path.of(pathString)).exists(); + UUID uuid = InterceptorOfSystemMethods.Global.randomUUID(); + SharedTestState.value1.set(uuid.getMostSignificantBits()); + SharedTestState.value2.set(uuid.getLeastSignificantBits()); + }); + } + + assertEquals(expected, new UUID(SharedTestState.value1.get(), SharedTestState.value2.get())); + assertEquals(1, SharedTestState.parkCount.get()); + assertEquals(1, SharedTestState.unparkCount.get()); + } + finally + { + Files.deleteIfExists(path); + } + } + + @Test + void nonSimulatorExecutionDoesNotPause() throws Exception + { + Path path = Files.createTempFile("sim-io-unsimulated", ".tmp"); + try + { + assertTrue(new InstrumentedFile(path).exists()); + assertEquals(1, SharedTestState.parkCount.get()); + assertEquals(1, SharedTestState.unparkCount.get()); + } + finally + { + Files.deleteIfExists(path); + } + } + + @Test + void deterministicEvaluationDoesNotPause() throws Exception + { + Path path = Files.createTempFile("sim-io-deterministic", ".tmp"); + try + { + String pathString = path.toString(); + try (Simulator sim = simulator(42L); + Session session = sim.byStep((SerializableRunnable) () -> { + Runnable operation = () -> { + if (new InstrumentedFile(Path.of(pathString)).exists()) + SharedTestState.eventCount.incrementAndGet(); + }; + try + { + Class.forName("org.apache.cassandra.simulator.systems.InterceptibleThread") + .getMethod("runDeterministic", Runnable.class) + .invoke(null, operation); + } + catch (ReflectiveOperationException e) + { + throw new RuntimeException(e); + } + })) + { + assertTrue(session.step()); + assertEquals(1, SharedTestState.parkCount.get()); + assertEquals(1, SharedTestState.unparkCount.get()); + assertEquals(1, SharedTestState.eventCount.get()); + assertFalse(session.hasNext()); } } finally diff --git a/simulator-context/src/main/java/org/apache/cassandra/simulator/context/SharedTestState.java b/simulator-context/src/main/java/org/apache/cassandra/simulator/context/SharedTestState.java index 861771c..edf132f 100644 --- a/simulator-context/src/main/java/org/apache/cassandra/simulator/context/SharedTestState.java +++ b/simulator-context/src/main/java/org/apache/cassandra/simulator/context/SharedTestState.java @@ -19,6 +19,7 @@ package org.apache.cassandra.simulator.context; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; /** * Shared mutable state for step-by-step simulator tests. @@ -51,6 +52,10 @@ public class SharedTestState /** General-purpose counter for tests that just need "did X happen". */ public static final AtomicInteger eventCount = new AtomicInteger(0); + /** General-purpose values for transferring exact long results across the classloader boundary. */ + public static final AtomicLong value1 = new AtomicLong(0); + public static final AtomicLong value2 = new AtomicLong(0); + /** Reset all counters. Call from {@code @BeforeEach}. */ public static void reset() { @@ -58,5 +63,7 @@ public class SharedTestState parkCount.set(0); unparkCount.set(0); eventCount.set(0); + value1.set(0); + value2.set(0); } } 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 b3868de..4d09335 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 @@ -707,7 +707,7 @@ public class Simulator implements AutoCloseable for (int i = 0; i < bodies.length; i++) { Runnable transferred = transfer.apply(bodies[i]); - actions.add(threadAction("entrypoint-" + i, transferred)); + actions.add(threadAction("-" + i, transferred)); } return new ActionSchedule(simulatedTime, diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableAction.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableAction.java index 42bdfe0..ad5f0c7 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableAction.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/step/ObservableAction.java @@ -47,6 +47,7 @@ public interface ObservableAction WAIT_CAPTURED, NOTIFY_CAPTURED, SLEEP_CAPTURED, + PRE_IO_CAPTURED, TIMEOUT_FIRED, CUSTOM } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedWait.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedWait.java index 631fb41..edc7b21 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedWait.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedWait.java @@ -47,7 +47,28 @@ import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; @Shared(scope = SIMULATION, inner = INTERFACES) public interface InterceptedWait extends NotifyThreadPaused { - enum Kind { SLEEP_UNTIL, WAIT_UNTIL, UNBOUNDED_WAIT, NEMESIS } + enum Kind + { + SLEEP_UNTIL("Sleep until"), + WAIT_UNTIL("Wait until"), + UNBOUNDED_WAIT("unbounded wait"), + NEMESIS("nemesis"), + PRE_IO("pre-I/O"); + + private final String description; + + Kind(String description) + { + this.description = description; + } + + @Override + public String toString() + { + return description; + } + } + enum Trigger { TIMEOUT, INTERRUPT, SIGNAL } interface TriggerListener 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 1fc1cbd..d631dde 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 @@ -37,6 +37,7 @@ import org.apache.cassandra.simulator.utils.Clock; import static org.apache.cassandra.simulator.SimulatorProperties.TEST_SIMULATOR_DETERMINISM_CHECK; import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM; import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.NEMESIS; +import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.PRE_IO; import static org.apache.cassandra.simulator.systems.NonInterceptible.Permit.OPTIONAL; import static org.apache.cassandra.simulator.systems.NonInterceptible.Permit.REQUIRED; @@ -155,10 +156,26 @@ public class InterceptingGlobalMethods extends InterceptingMonitors implements I if (thread == null || thread.isEvaluationDeterministic() || !random.decide(chance)) return; - InterceptedConditionWait signal = new InterceptedConditionWait(NEMESIS, 0L, thread, captureWaitSite(thread), null); + pause(thread, NEMESIS); + } + + @Override + public void preIo() + { + InterceptibleThread thread = ifIntercepted(); + if (thread == null || thread.isEvaluationDeterministic()) + return; + + ObservableActions.emit(ObservableAction.Kind.PRE_IO_CAPTURED); + pause(thread, PRE_IO); + } + + private void pause(InterceptibleThread thread, InterceptedWait.Kind kind) + { + InterceptedConditionWait signal = new InterceptedConditionWait(kind, 0L, thread, captureWaitSite(thread), null); thread.interceptWait(signal); - // save interrupt state to restore afterwards - new ones only arrive if terminating simulation + // Save interrupt state to restore afterwards; new interrupts only arrive when terminating the simulation. boolean restoreInterrupt = Thread.interrupted(); try { diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfSystemMethods.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfSystemMethods.java index c6474d0..cec92df 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfSystemMethods.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfSystemMethods.java @@ -58,6 +58,7 @@ public interface InterceptorOfSystemMethods void notify(Object monitor); void notifyAll(Object monitor); void nemesis(float chance); + void preIo(); void park(); void parkNanos(long nanos); @@ -203,6 +204,12 @@ public interface InterceptorOfSystemMethods methods.nemesis(chance); } + // explicit scheduler boundary before synchronous host I/O + public static void preIo() + { + methods.preIo(); + } + // ThreadLocalRandom#advanceProbe(int), Striped64#advanceProbe(int) [GlobalMethodTransformer, GLOBAL_METHODS || DETERMINISTIC] public static int advanceProbe(int probe) { @@ -374,6 +381,11 @@ public interface InterceptorOfSystemMethods { } + @Override + public void preIo() + { + } + @Override public void park() { 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 56835d5..47e06a6 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 @@ -267,10 +267,9 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon applyToSignal(out, START_TIMEOUT_TASK, "Timeout", wakeupWith, TIMEOUT, wakeupWith.waitTime()); break; case NEMESIS: - applyToSignal(out, WAKE_UP_THREAD, "Nemesis", wakeupWith, SIGNAL, -1L); - break; - default : - applyToSignal(out, WAKE_UP_THREAD, "Continue", wakeupWith, SIGNAL, -1L); + case PRE_IO: + default: + applyToSignal(out, WAKE_UP_THREAD, wakeupWith.kind().toString(), wakeupWith, SIGNAL, -1L); break; } } diff --git a/simulator-io/src/main/java/org/apache/cassandra/simulator/io/InstrumentedFile.java b/simulator-io/src/main/java/org/apache/cassandra/simulator/io/InstrumentedFile.java new file mode 100644 index 0000000..eb43500 --- /dev/null +++ b/simulator-io/src/main/java/org/apache/cassandra/simulator/io/InstrumentedFile.java @@ -0,0 +1,153 @@ +/* + * 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.io; + +import java.io.FilenameFilter; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Predicate; + +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.simulator.systems.InterceptorOfSystemMethods; +import org.apache.cassandra.simulator.systems.PerClassLoader; + +/** + * First-cut simulator I/O wrapper for Cassandra's File abstraction. + * + * The simulator redirects construction of File to this subclass. + * Overridden methods insert a scheduler-controlled boundary immediately before the underlying + * host file-system operation. The operation still runs synchronously on real files, but its + * completion is not visible to the caller until the simulator schedules this thread again. + */ +@PerClassLoader +public class InstrumentedFile extends File +{ + public InstrumentedFile(Path path) + { + super(path); + } + + public InstrumentedFile(java.io.File file) + { + super(file); + } + + public InstrumentedFile(File parent, String child) + { + super(parent, child); + } + + @Override + public boolean exists() + { + ioStep(); + return Files.exists(toPath()); + } + + @Override + public boolean createFileIfNotExists() throws IOException + { + ioStep(); + if (Files.exists(toPath())) + return false; + Files.createFile(toPath()); + return true; + } + + @Override + public void delete() + { + deleteIfExists(); + } + + @Override + public void deleteIfExists() + { + ioStep(); + try + { + Files.deleteIfExists(toPath()); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @Override + public FileChannel newReadChannel() throws IOException + { + ioStep(); + return FileChannel.open(toPath(), StandardOpenOption.READ); + } + + @Override + public String[] listNames(FilenameFilter filter) throws IOException + { + ioStep(); + java.io.File[] files = toPath().toFile().listFiles((dir, name) -> filter.accept(dir, name)); + if (files == null) + return new String[0]; + String[] names = new String[files.length]; + for (int i = 0; i < files.length; i++) + names[i] = files[i].getName(); + return names; + } + + @Override + public List<File> listUnchecked(Predicate<File> predicate) + { + ioStep(); + try (DirectoryStream<Path> stream = Files.newDirectoryStream(toPath())) + { + List<File> out = new ArrayList<>(); + for (Path p : stream) + { + File file = new InstrumentedFile(p); + if (predicate.test(file)) + out.add(file); + } + return out; + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static volatile IoOperationListener listener = IoOperationListener.NONE; + + public static void setListener(IoOperationListener listener) + { + InstrumentedFile.listener = listener == null ? IoOperationListener.NONE : listener; + } + + private static void ioStep() + { + listener.beforePreIoPark(); + InterceptorOfSystemMethods.Global.preIo(); + listener.afterPreIoPark(); + } +} diff --git a/simulator-io/src/main/java/org/apache/cassandra/simulator/io/IoOperationListener.java b/simulator-io/src/main/java/org/apache/cassandra/simulator/io/IoOperationListener.java new file mode 100644 index 0000000..22d1fe3 --- /dev/null +++ b/simulator-io/src/main/java/org/apache/cassandra/simulator/io/IoOperationListener.java @@ -0,0 +1,34 @@ +/* + * 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.io; + +public interface IoOperationListener +{ + IoOperationListener NONE = new IoOperationListener() + { + }; + + default void beforePreIoPark() + { + } + + default void afterPreIoPark() + { + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
