From f02ab5608154e828cb0220e75c78d48725e8b3ee Mon Sep 17 00:00:00 2001 From: Esau Rodriguez Date: Thu, 1 Oct 2026 10:43:19 +0200 Subject: [PATCH] Add concurrent stateful testing `Stateful.run(machine, tc, Stateful.options().maxConcurrency(n))` runs a state machine on engine-drawn worker threads, following hegel-rust's concurrent runner: rounds are driven on the root handle, each worker pulls rules through a fresh per-round clone, invariants run at the join points, and a round's outcome is resolved in Rust's precedence (control error, then overrun/invalid, then the lowest worker's failure). `@Rule(group)` assigns concurrency groups; `ConcurrentPool` is the thread-safe pool for concurrent rules. Worker output is buffered per round and reported under the round header stamped `[worker N +X.XXXms]`. The sequential path is unchanged, so existing stateful tests and their stored failures are unaffected. `Libhegel` gains `testCaseClone` (a draw: it can report `E_STOP_TEST` on a shorter replay, hence the raw-rc convention) and `testCaseSetWorker`. Co-Authored-By: Claude Fable 5.1 --- .claude/CLAUDE.md | 70 ++- README.md | 28 + RELEASE.md | 88 +++ .../src/main/java/dev/hegel/JnaLibhegel.java | 19 + .../test/java/dev/hegel/JnaLibhegelTest.java | 4 + .../java/dev/hegel/lowlevel/Libhegel.java | 22 + .../src/main/java/dev/hegel/RealLibhegel.java | 24 + .../test/java/dev/hegel/RealLibhegelTest.java | 4 + .../main/java/dev/hegel/ConcurrentPool.java | 97 ++++ .../src/main/java/dev/hegel/DataSource.java | 38 +- .../main/java/dev/hegel/LiveDataSource.java | 51 +- shared/src/main/java/dev/hegel/Pool.java | 4 + shared/src/main/java/dev/hegel/Rule.java | 12 + shared/src/main/java/dev/hegel/Stateful.java | 546 +++++++++++++++++- shared/src/main/java/dev/hegel/TestCase.java | 178 +++++- .../java/dev/hegel/BindingErrorPathsTest.java | 82 ++- .../dev/hegel/ConcurrentCounterExample.java | 66 +++ .../hegel/ConcurrentStatefulDriverTest.java | 428 ++++++++++++++ .../dev/hegel/ConcurrentStatefulTest.java | 458 +++++++++++++++ .../src/test/java/dev/hegel/FakeLibhegel.java | 82 ++- .../src/test/java/dev/hegel/StatefulTest.java | 10 +- 21 files changed, 2194 insertions(+), 117 deletions(-) create mode 100644 RELEASE.md create mode 100644 shared/src/main/java/dev/hegel/ConcurrentPool.java create mode 100644 shared/src/test/java/dev/hegel/ConcurrentCounterExample.java create mode 100644 shared/src/test/java/dev/hegel/ConcurrentStatefulDriverTest.java create mode 100644 shared/src/test/java/dev/hegel/ConcurrentStatefulTest.java diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index df5d39c..f4783ed 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -171,26 +171,56 @@ public `Generator`/`TestCase`/`Generators`/`Hegel`/`Stateful` surface stays in ` `@HegelTest` annotation + `HegelTestExtension` (a JUnit 5 `TestTemplateInvocationContextProvider` that drives the engine loop and invokes the user method per case). -- **Stateful testing** — `Stateful.run(machine, tc)` reflects `@Rule`/`@Invariant` methods (sorted - by name for determinism) and registers them via `hegel_new_state_machine` as a *sequential* - machine (every rule in group 0, concurrency fixed at `1, 1`, worker index 0, and a per-machine - `step_count` — `Stateful.DEFAULT_STEP_COUNT` = 50, or the `run(machine, tc, stepCount)` overload; - the engine has no default). `@Rule(weight = w)` becomes the `rule_weights` array (validated - finite and positive in Java; `null` — the engine's all-equal default — when every rule is at 1); - the engine samples proportionally among the *enabled* rules, so weights are hints, and tests - check them per case, not in aggregate. It then drives the - engine's round protocol: `hegel_state_machine_next_group` opens each round (or reports - `HEGEL_STATE_MACHINE_DONE`, which is `INT64_MIN`), `hegel_state_machine_next_rule` hands out the - round's rules until the join point, a rule that fails its own assumption is reported with - `hegel_state_machine_rule_rejected` (and the round's STATEFUL_RULE span discarded), and at each - join point `hegel_state_machine_should_check_invariant` decides which invariants run — always - for `@Invariant(alwaysRun = true)`, sampled otherwise; the initial and final checks run every - invariant unconditionally. The machine handle is freed in a `finally`. An engine-level - `E_ASSUME` inside a rule (the data source is aborted) unwinds the whole case as invalid instead - of being reported as the rule's rejection. (Since engine 0.44 concurrency alone no longer marks - a run nondeterministic, and `HEGEL_RUN_STATUS_FAILED_NONDETERMINISTIC` is retired; this binding - drives machines sequentially anyway.) `Pool` tracks previously generated values over - the engine's pool primitives so rules can reuse or consume them. +- **Stateful testing** — `Stateful.run(machine, tc[, options])` reflects `@Rule`/`@Invariant` + methods (sorted by name for determinism) and registers them via `hegel_new_state_machine`: + `@Rule(group = "g")` names become `rule_groups` ids (0.. in first-appearance order; rules that + name none share `Stateful.ANONYMOUS_GROUP`, so a plain machine is all-zeros), `@Rule(weight = w)` + becomes the `rule_weights` array (validated finite and positive in Java; `null` — the engine's + all-equal default — when every rule is at 1; the engine samples proportionally among the + *enabled* rules, so weights are hints, and tests check them per case, not in aggregate), and + `Stateful.Options` carries the per-machine `step_count` (`Stateful.DEFAULT_STEP_COUNT` = 50; the + engine has no default) and the concurrency bounds (default `1, 1`). The engine draws the + concurrency level and the driver runs exactly that many workers. Two drivers share the + registration and the round protocol — `hegel_state_machine_next_group` opens each round on the + root handle (or reports `HEGEL_STATE_MACHINE_DONE`, which is `INT64_MIN`), + `hegel_state_machine_next_rule(worker)` hands out the round's rules until the worker's join point, + a rule that fails its own assumption is reported with `hegel_state_machine_rule_rejected(worker)`, + and at each join point `hegel_state_machine_should_check_invariant` (root handle) decides which + invariants run — always for `@Invariant(alwaysRun = true)`, sampled otherwise; the initial and + final checks run every invariant unconditionally: + - `maxConcurrency == 1` → the **sequential** driver, unchanged from before concurrency existed: + everything on the calling thread and the root handle as worker 0, one `STATEFUL_RULE` span + per round (discarded on rejection), `Step N: rule` notes. + - `maxConcurrency > 1` → the **concurrent** driver, modelled on hegel-rust's `run_concurrent_machine` + (even when the drawn level is 1, so a run's output format is uniform). Persistent daemon + worker threads (`hegel-worker-N`) live for the test case; per round the root thread notes the + header (`---------------- Round k: group "name" ----------------`), makes a **fresh clone per + worker** (`hegel_test_case_clone` + `hegel_test_case_set_worker`; cloning is a draw and can + return `E_STOP_TEST` on a replay, hence the raw-rc convention for `Libhegel.testCaseClone`), + hands each worker its clone through a queue, waits for one event per worker (the join point), + absorbs the workers' buffered draws/notes into the root in worker order (each line stamped + `[worker N +X.XXXms]` client-side — Java notes never go through `hegel_note`, so the engine's + own attribution is invisible here; draw names stay unique family-wide through the root's + synchronized name counter and are recorded plain), frees the clones, and resolves the round in + Rust's precedence: a control error (`LibhegelException`/`IllegalArgumentException`, or a worker + that exited without reporting) is rethrown first; then an overrun or an engine-level `E_ASSUME` + concludes the case, dropping any failures found alongside with a `Dropped concurrent failure + from worker N` note; otherwise the lowest worker's failure is rethrown as-is (its stack trace + is the worker's, so `originOf` still works) and the rest are noted as dropped. There is no + cancellation: a failing worker ends its own round and the others finish theirs (an overrun + aborts the family engine-side anyway). No spans are opened in the concurrent path (none of the + bindings do; the engine owns rule structure). The `Reporter` is only ever called from the + driving thread: verbose output of worker lines is delivered at the join. Workers are stopped + and joined in a `finally` before any leftover clone is freed. + The machine handle is freed in a `finally`. `Pool` (sequential; bound to its creating handle, + unsynchronised) and `ConcurrentPool` (one monitor across the engine call and the map update; + `add(tc, value)` draws through the calling rule's handle) track previously generated values over + the engine's pool primitives so rules can reuse or consume them. `LiveDataSource` maps + `E_CONCURRENT_USE` to a `HegelException` naming the fix (draw through the rule's own `TestCase`, + use `ConcurrentPool`). `FakeLibhegel.concurrentRounds` scripts per-worker rule queues so the + driver's join-point layout and every resolution path are tested deterministically + (`ConcurrentStatefulDriverTest`); `ConcurrentStatefulTest` covers the real engine (parallelism, + group exclusion, a found lost-update race, pools, blobs). - **Derivation** — `dev.hegel.generators.Derive` + `RecordGenerator` build generators from records, enums, scalars, and generic `List`/`Set`/`Optional`/`Map` by reflection. diff --git a/README.md b/README.md index 87167d9..0d28b88 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,34 @@ Hegel reports the minimal example showing that our sort is incorrectly dropping The optional `"xs"` label passed to `draw` names the value in the falsifying-example output. See the [API documentation](https://javadoc.io/doc/dev.hegel/hegel) for a full tour of generators, combinators, control functions, and settings. +## Stateful testing + +Model-based tests drive a state machine whose `@Rule` methods are the actions and whose `@Invariant` +methods are the properties, with the engine choosing and shrinking the action sequence: + +```java +Stateful.run(new IntegerStack(), tc); +``` + +Run the same machine with a concurrency bound to look for races: the engine draws how many worker +threads apply the rules at once, `@Rule(group = "...")` says which rules may overlap, and a +`ConcurrentPool` passes generated values between concurrent rules. + +```java +Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4)); +``` + +See the `Stateful` Javadoc for complete sequential and concurrent examples. The concurrent one, a +counter that loses updates under contention, is checked in as a runnable demonstration that is +skipped unless asked for; run it from a checkout with + +```bash +mvn test -pl hegel -am -Dtest=ConcurrentCounterExample -Dhegel.examples=true -Dsurefire.failIfNoSpecifiedTests=false +``` + +(or `-pl hegel-jna -am` for the JNA frontend) and it fails with a report of the rounds and the +worker-stamped rule and draw lines that led to the lost update. + ## Using Hegel as a library Frontends for other JVM languages (or custom runners) use `Hegel.run` instead of `Hegel.test`. It diff --git a/RELEASE.md b/RELEASE.md new file mode 100644 index 0000000..7cededc --- /dev/null +++ b/RELEASE.md @@ -0,0 +1,88 @@ +RELEASE_TYPE: minor + +This release adds concurrent stateful testing. Running a state machine with +`Stateful.Options.maxConcurrency` above 1 makes the engine draw a concurrency level for each test +case and the driver run that many worker threads, which apply the machine's rules at the same time +on the same machine object, so races and lost updates surface as invariant failures like any other +bug: + +```java +class Counter { + private final Map store = new ConcurrentHashMap<>(); + private final AtomicInteger increments = new AtomicInteger(); + private final ConcurrentPool keys; + + Counter(TestCase tc) { + keys = new ConcurrentPool<>(tc); + } + + @Rule(group = "ops") + void register(TestCase tc) { + String key = tc.draw(text().minSize(1).maxSize(3)); + store.putIfAbsent(key, 0); + keys.add(tc, key); + } + + @Rule(group = "ops", weight = 3) + void increment(TestCase tc) { + String key = tc.draw(keys.reusable()); + store.put(key, store.get(key) + 1); // racy: a lost update + increments.incrementAndGet(); + } + + @Rule(group = "audit") + void audit(TestCase tc) { + tc.note("store holds " + store.size() + " keys"); + } + + @Invariant + void noLostUpdates(TestCase tc) { + assertEquals(increments.get(), store.values().stream().mapToInt(Integer::intValue).sum()); + } +} + +@HegelTest +void counterUnderContention(TestCase tc) { + Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4)); +} +``` + +Execution proceeds in rounds. Each round the engine picks one concurrency group (`@Rule(group = +"...")`; rules that name none share the anonymous group), every worker applies a few of that group's +rules concurrently, and once all workers have finished the round the sampled invariants run on the +driving thread. Rules in the same group may therefore overlap in time, rules in different groups +never do, and invariants never overlap a rule. Rules run on the same machine object from several +threads, so its state must be safe for concurrent access; each rule receives its worker's own +`TestCase` and must draw only through it. The new `ConcurrentPool` is the thread-safe counterpart of +`Pool` for passing generated values between concurrent rules; its `add` takes the calling rule's test +case. A failure report groups each round's draws and notes by worker under the round's header, every +line stamped `[worker N +X.XXXms]` with the time since the machine started: + +``` +Concurrency level: 2 +---------------- Round 1: group "ops" ---------------- +[worker 0 +0.412ms] Rule: increment +[worker 0 +0.437ms] draw_1 = "a"; +[worker 1 +0.415ms] Rule: increment +[worker 1 +0.440ms] draw_2 = "a"; + +java.lang.AssertionError: expected: <2> but was: <1> +``` + +When a rule fails in one worker the other workers finish their round before the failure is +reported, and when several workers fail in the same round the lowest-numbered worker's exception is +reported with the others noted as dropped. A concurrency bug that depends on thread scheduling may +not reproduce on every replay; the engine then confirms it by repeated replay and reports it with a +caveat, as for any nondeterministic failure. + +`Stateful.options()` is the new way to configure a run: +`Stateful.run(machine, tc, Stateful.options().stepCount(30).minConcurrency(2).maxConcurrency(4))`. +The existing `run(machine, tc)` and `run(machine, tc, stepCount)` overloads are unchanged and keep +running the machine sequentially on the calling thread, with the same output and choice sequences +as before, so stored failures for existing stateful tests remain valid. + +In `dev.hegel:hegel-lowlevel`, `Libhegel` gains `testCaseClone` (`hegel_test_case_clone`; a draw, +so it returns the raw return code and reports `E_STOP_TEST` on a replay of a shorter sequence) and +`testCaseSetWorker` (`hegel_test_case_set_worker`). Bindings implementing `Libhegel` must add the +two methods, which is what makes this a minor release; the `dev.hegel:hegel` and +`dev.hegel:hegel-jna` frontends only gain API. diff --git a/hegel-jna/src/main/java/dev/hegel/JnaLibhegel.java b/hegel-jna/src/main/java/dev/hegel/JnaLibhegel.java index 4c1a2f5..a3cd87f 100644 --- a/hegel-jna/src/main/java/dev/hegel/JnaLibhegel.java +++ b/hegel-jna/src/main/java/dev/hegel/JnaLibhegel.java @@ -122,6 +122,10 @@ int hegel_test_case_from_blob( int hegel_test_case_should_capture(Pointer ctx, Pointer tc, ByteByReference out); + int hegel_test_case_clone(Pointer ctx, Pointer tc, PointerByReference out); + + int hegel_test_case_set_worker(Pointer ctx, Pointer tc, long workerIndex); + int hegel_generate_boolean(Pointer ctx, Pointer tc, double p, byte hasForced, byte forced, ByteByReference out); int hegel_generate_integer(Pointer ctx, Pointer tc, long min, long max, LongByReference out); @@ -548,6 +552,21 @@ public boolean testCaseShouldCapture(long tc) { return out.getValue() != 0; } + @Override + public int testCaseClone(long tc, long[] out) { + PointerByReference ref = new PointerByReference(); + int code = lib.hegel_test_case_clone(ctx(), pointer(tc), ref); + if (code == Abi.OK) { + out[0] = address(ref.getValue()); + } + return code; + } + + @Override + public void testCaseSetWorker(long tc, long workerIndex) { + check("hegel_test_case_set_worker", lib.hegel_test_case_set_worker(ctx(), pointer(tc), workerIndex)); + } + // --- draws --- @Override diff --git a/hegel-jna/src/test/java/dev/hegel/JnaLibhegelTest.java b/hegel-jna/src/test/java/dev/hegel/JnaLibhegelTest.java index 86a6f7b..b7a81bc 100644 --- a/hegel-jna/src/test/java/dev/hegel/JnaLibhegelTest.java +++ b/hegel-jna/src/test/java/dev/hegel/JnaLibhegelTest.java @@ -74,6 +74,10 @@ void infrastructureCallsReportNullHandles() { assertThrows(HegelException.class, () -> lib.runStart(0, null)); assertThrows(HegelException.class, () -> lib.runStartBlob(0, "blob", null)); assertThrows(HegelException.class, () -> lib.testCaseShouldCapture(0)); + assertThrows(HegelException.class, () -> lib.testCaseSetWorker(0, 0)); + long[] clone = {7}; + assertEquals(Abi.E_INVALID_HANDLE, lib.testCaseClone(0, clone)); + assertEquals(7, clone[0]); } @Test diff --git a/hegel-lowlevel/src/main/java/dev/hegel/lowlevel/Libhegel.java b/hegel-lowlevel/src/main/java/dev/hegel/lowlevel/Libhegel.java index 0bf169b..113528d 100644 --- a/hegel-lowlevel/src/main/java/dev/hegel/lowlevel/Libhegel.java +++ b/hegel-lowlevel/src/main/java/dev/hegel/lowlevel/Libhegel.java @@ -200,6 +200,28 @@ static Libhegel load(Path library, Iterable backends) */ boolean testCaseShouldCapture(long tc); + /** + * {@code hegel_test_case_clone}: a caller-owned handle onto an independent choice stream of + * the same test case. The clone shares the case's outcome and budgets (and its collections, + * pools and state machines, which are family-wide) but draws from its own sequence, so a clone + * and its source can be driven concurrently from different threads while staying + * deterministic under replay. Release it with {@link #testCaseFree}. + * + *

Cloning consumes one choice position on the source stream, so it is a draw: it returns + * the raw rc — {@link Abi#E_STOP_TEST} when the choice budget is exhausted (as on a replay of + * a shorter sequence), {@link Abi#E_CONCURRENT_USE} if the source is mid-operation on another + * thread, {@link Abi#E_ALREADY_COMPLETE} once the case has completed — and on OK writes the + * handle into {@code out[0]}. + */ + int testCaseClone(long tc, long[] out); + + /** + * {@code hegel_test_case_set_worker}: attribute the handle's engine-side output (notes and + * printer lines) to concurrent worker {@code workerIndex}, which must be non-negative. Blocks + * and clones derived from the handle afterwards inherit the attribution. + */ + void testCaseSetWorker(long tc, long workerIndex); + // Per-test-case draws. Each returns the raw rc. int generateBoolean(long tc, double p, boolean[] out); diff --git a/hegel/src/main/java/dev/hegel/RealLibhegel.java b/hegel/src/main/java/dev/hegel/RealLibhegel.java index c965c9f..ad99803 100644 --- a/hegel/src/main/java/dev/hegel/RealLibhegel.java +++ b/hegel/src/main/java/dev/hegel/RealLibhegel.java @@ -123,6 +123,8 @@ private static MethodHandle findEmitLine() { private final MethodHandle testCaseFromBlob; private final MethodHandle testCaseFree; private final MethodHandle testCaseShouldCapture; + private final MethodHandle testCaseClone; + private final MethodHandle testCaseSetWorker; private final MethodHandle generateBoolean; private final MethodHandle generateInteger; private final MethodHandle generateFloat; @@ -288,6 +290,13 @@ private static MethodHandle findEmitLine() { lookup, "hegel_test_case_should_capture", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS)); + this.testCaseClone = + h(linker, lookup, "hegel_test_case_clone", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, ADDRESS)); + this.testCaseSetWorker = h( + linker, + lookup, + "hegel_test_case_set_worker", + FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS, JAVA_LONG)); this.generateBoolean = h( linker, lookup, @@ -736,6 +745,21 @@ public boolean testCaseShouldCapture(long tc) { return out.get(JAVA_BOOLEAN, 0); } + @Override + public int testCaseClone(long tc, long[] out) { + MemorySegment seg = Arena.ofAuto().allocate(ADDRESS); + int code = rc(testCaseClone, segment(tc), seg); + if (code == Abi.OK) { + out[0] = seg.get(ADDRESS, 0).address(); + } + return code; + } + + @Override + public void testCaseSetWorker(long tc, long workerIndex) { + check("hegel_test_case_set_worker", rc(testCaseSetWorker, segment(tc), workerIndex)); + } + // --- draws --- @Override diff --git a/hegel/src/test/java/dev/hegel/RealLibhegelTest.java b/hegel/src/test/java/dev/hegel/RealLibhegelTest.java index 5a07383..52c84a1 100644 --- a/hegel/src/test/java/dev/hegel/RealLibhegelTest.java +++ b/hegel/src/test/java/dev/hegel/RealLibhegelTest.java @@ -83,6 +83,10 @@ void infrastructureCallsReportNullHandles() { assertThrows(HegelException.class, () -> lib.runStart(0, null)); assertThrows(HegelException.class, () -> lib.runStartBlob(0, "blob", null)); assertThrows(HegelException.class, () -> lib.testCaseShouldCapture(0)); + assertThrows(HegelException.class, () -> lib.testCaseSetWorker(0, 0)); + long[] clone = {7}; + assertEquals(Abi.E_INVALID_HANDLE, lib.testCaseClone(0, clone)); + assertEquals(7, clone[0]); } @Test diff --git a/shared/src/main/java/dev/hegel/ConcurrentPool.java b/shared/src/main/java/dev/hegel/ConcurrentPool.java new file mode 100644 index 0000000..ee4c84f --- /dev/null +++ b/shared/src/main/java/dev/hegel/ConcurrentPool.java @@ -0,0 +1,97 @@ +package dev.hegel; + +import java.util.HashMap; +import java.util.Map; + +/** + * A {@link Pool} for concurrent state machines: previously generated values that rules running on + * different worker threads can add, reuse and consume, with the engine choosing and shrinking over + * which value a rule gets. + * + *

Create one per test case on the test case the machine is run under, and have rules go through + * the {@link TestCase} they were handed: {@link #add(TestCase, Object)} records a value through the + * calling rule's handle, and the generators from {@link #reusable()} and {@link #consuming()} draw + * through whichever handle draws them. Every operation holds the pool's lock across the engine + * call and the bookkeeping, so the engine's choice and the Java-side values never disagree even + * when workers race. Drawing from an empty pool rejects the current rule (as if by {@code + * assume(false)}), so the engine retries the slot with another rule. + * + *

A {@link #reusable()} draw hands out the stored reference itself; do not mutate a pooled value + * from a rule, since another worker may be reading it at the same time. + * + * @param the type of pooled values + * @see Stateful + */ +public final class ConcurrentPool { + private final long poolId; + private final Map values = new HashMap<>(); + + /** + * Create a pool tracked by the current test case. + * + * @param tc the test case the machine is run under + */ + public ConcurrentPool(TestCase tc) { + this.poolId = tc.newPool(); + } + + /** + * @return whether no values are in the pool + */ + public synchronized boolean isEmpty() { + return values.isEmpty(); + } + + /** + * @return the number of values currently in the pool + */ + public synchronized int size() { + return values.size(); + } + + /** + * Add a value to the pool. + * + * @param tc the test case of the rule adding the value: its own worker's handle + * @param value the value to add + */ + public synchronized void add(TestCase tc, T value) { + values.put(tc.poolAdd(poolId), value); + } + + /** + * A generator over the values in the pool that yields a value without removing it. + * + * @return the reusing generator + */ + public Generator reusable() { + return new PoolGenerator(false); + } + + /** + * A generator that consumes values from the pool: it removes the value it yields, so once + * consumed a value is never drawn again. + * + * @return the consuming generator + */ + public Generator consuming() { + return new PoolGenerator(true); + } + + private final class PoolGenerator implements Generator { + private final boolean consume; + + PoolGenerator(boolean consume) { + this.consume = consume; + } + + @Override + public T doDraw(TestCase tc) { + synchronized (ConcurrentPool.this) { + tc.assume(!values.isEmpty()); + long variableId = tc.poolGenerate(poolId, consume); + return consume ? values.remove(variableId) : values.get(variableId); + } + } + } +} diff --git a/shared/src/main/java/dev/hegel/DataSource.java b/shared/src/main/java/dev/hegel/DataSource.java index 5df54ab..6712da4 100644 --- a/shared/src/main/java/dev/hegel/DataSource.java +++ b/shared/src/main/java/dev/hegel/DataSource.java @@ -91,24 +91,48 @@ StringGeneratorHandle textGenerator( long poolGenerate(long poolId, boolean consume); - // Stateful testing. Machines are driven sequentially (one worker, one concurrency group). + // Stateful testing. The root source registers the machine, advances rounds and samples + // invariants; at concurrency > 1 each worker pulls its rules through its own clone. - /** {@code ruleWeights} is parallel to {@code ruleNames}, or {@code null} for equal weights. */ - long newStateMachine( + /** What registering a state machine yields: its handle and the concurrency level the engine drew. */ + record StateMachine(long id, int concurrency) {} + + /** + * {@code ruleGroups} and {@code ruleWeights} are parallel to {@code ruleNames} ({@code null} + * weights = all equal); {@code invariantAlwaysCheck} is parallel to {@code invariantNames}. + * The engine draws the concurrency level in {@code [minConcurrency, maxConcurrency]}. + */ + StateMachine newStateMachine( List ruleNames, + long[] ruleGroups, double[] ruleWeights, List invariantNames, boolean[] invariantAlwaysCheck, + long minConcurrency, + long maxConcurrency, int stepCount); /** Start the next round: its group id, or {@link Abi#STATE_MACHINE_DONE} when the machine is done. */ long stateMachineNextGroup(long stateMachineId); - /** The next rule index for this round, or {@link Abi#STATE_MACHINE_DONE} at the round's join point. */ - long stateMachineNextRule(long stateMachineId); + /** + * The next rule index for worker {@code workerIndex} this round, or {@link + * Abi#STATE_MACHINE_DONE} at the worker's join point. + */ + long stateMachineNextRule(long stateMachineId, long workerIndex); + + /** The rule most recently handed to worker {@code workerIndex} failed an assumption: retry the slot. */ + void stateMachineRuleRejected(long stateMachineId, long workerIndex); + + /** + * A source over an independent choice stream of the same test case ({@code + * hegel_test_case_clone}), attributed to concurrent worker {@code workerIndex}, for a worker + * thread to draw through. Release it with {@link #release()} once the worker is done with it. + */ + DataSource cloneForWorker(long workerIndex); - /** The rule most recently handed out failed an assumption: do not count it as a step. */ - void stateMachineRuleRejected(long stateMachineId); + /** Free the handle behind a source from {@link #cloneForWorker}. Safe once the case is aborted. */ + void release(); /** The engine's sampling decision for invariant {@code invariantIndex} at this join point. */ boolean stateMachineShouldCheckInvariant(long stateMachineId, long invariantIndex); diff --git a/shared/src/main/java/dev/hegel/LiveDataSource.java b/shared/src/main/java/dev/hegel/LiveDataSource.java index 2dedd5d..9e026c2 100644 --- a/shared/src/main/java/dev/hegel/LiveDataSource.java +++ b/shared/src/main/java/dev/hegel/LiveDataSource.java @@ -43,6 +43,11 @@ private void translate(int rc, String op) { throw new AssumeRejected(); case Abi.E_INVALID_ARG: throw new IllegalArgumentException(nullToEmpty(lib.lastErrorMessage())); + case Abi.E_CONCURRENT_USE: + throw new HegelException("hegel_" + op + + ": the test-case handle was used from two threads at once. In a concurrent" + + " state machine, draw only through the TestCase handed to the rule and share" + + " generated values through a ConcurrentPool."); default: throw new HegelException( "hegel_" + op + " failed (rc=" + rc + "): " + nullToEmpty(lib.lastErrorMessage())); @@ -302,33 +307,33 @@ public long poolGenerate(long poolId, boolean consume) { } @Override - public long newStateMachine( + public StateMachine newStateMachine( List ruleNames, + long[] ruleGroups, double[] ruleWeights, List invariantNames, boolean[] invariantAlwaysCheck, + long minConcurrency, + long maxConcurrency, int stepCount) { checkLive(); - // Sequential: every rule in group 0 and exactly one worker, so the drawn concurrency level - // is always 1 and every rule/rejection call below is made on behalf of worker 0. - long[] groups = new long[ruleNames.size()]; long[] id = new long[1]; long[] concurrency = new long[1]; translate( lib.newStateMachine( tc, ruleNames, - groups, + ruleGroups, ruleWeights, invariantNames, invariantAlwaysCheck, - 1, - 1, + minConcurrency, + maxConcurrency, stepCount, id, concurrency), "new_state_machine"); - return id[0]; + return new StateMachine(id[0], (int) concurrency[0]); } @Override @@ -340,17 +345,39 @@ public long stateMachineNextGroup(long stateMachineId) { } @Override - public long stateMachineNextRule(long stateMachineId) { + public long stateMachineNextRule(long stateMachineId, long workerIndex) { checkLive(); long[] index = new long[1]; - translate(lib.stateMachineNextRule(tc, stateMachineId, 0, index), "state_machine_next_rule"); + translate(lib.stateMachineNextRule(tc, stateMachineId, workerIndex, index), "state_machine_next_rule"); return index[0]; } @Override - public void stateMachineRuleRejected(long stateMachineId) { + public void stateMachineRuleRejected(long stateMachineId, long workerIndex) { + checkLive(); + translate(lib.stateMachineRuleRejected(tc, stateMachineId, workerIndex), "state_machine_rule_rejected"); + } + + @Override + public DataSource cloneForWorker(long workerIndex) { checkLive(); - translate(lib.stateMachineRuleRejected(tc, stateMachineId, 0), "state_machine_rule_rejected"); + // Cloning consumes a choice position, so it is a draw: on a replay of a shorter sequence + // it reports STOP_TEST like any other draw, and the case is an overrun. + long[] clone = new long[1]; + translate(lib.testCaseClone(tc, clone), "test_case_clone"); + try { + lib.testCaseSetWorker(clone[0], workerIndex); + } catch (RuntimeException e) { + lib.testCaseFree(clone[0]); + throw e; + } + return new LiveDataSource(lib, clone[0]); + } + + @Override + public void release() { + // Not gated on `aborted`: the clone handle is caller-owned and must be released exactly once. + lib.testCaseFree(tc); } @Override diff --git a/shared/src/main/java/dev/hegel/Pool.java b/shared/src/main/java/dev/hegel/Pool.java index 3a86052..31653df 100644 --- a/shared/src/main/java/dev/hegel/Pool.java +++ b/shared/src/main/java/dev/hegel/Pool.java @@ -15,6 +15,10 @@ * shrinks like any other draw. Drawing from an empty pool rejects the current test case (as if by * {@code assume(false)}). * + *

A pool is bound to the test case it was created on and is not thread-safe: in a machine run + * with {@link Stateful.Options#maxConcurrency maxConcurrency} above 1, use a {@link + * ConcurrentPool} instead. + * * @param the type of pooled values */ public final class Pool { diff --git a/shared/src/main/java/dev/hegel/Rule.java b/shared/src/main/java/dev/hegel/Rule.java index a662ca7..f04e459 100644 --- a/shared/src/main/java/dev/hegel/Rule.java +++ b/shared/src/main/java/dev/hegel/Rule.java @@ -18,6 +18,18 @@ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface Rule { + /** + * The rule's concurrency group, for machines run with {@link Stateful.Options#maxConcurrency + * maxConcurrency} above 1. Each round the engine picks one group and hands out only that + * group's rules, so rules in the same group may run concurrently with each other and rules in + * different groups never overlap. Rules that name no group share the {@link + * Stateful#ANONYMOUS_GROUP anonymous group}. Group names are arbitrary and only appear in the + * round headers of a failure report. Groups have no observable effect on a sequential machine. + * + * @return the group name, or the empty string for the anonymous group + */ + String group() default ""; + /** * How often the engine should pick this rule relative to the machine's other rules: a rule of * weight 5 is offered about five times as often as a rule of weight 1. The weight is a hint, diff --git a/shared/src/main/java/dev/hegel/Stateful.java b/shared/src/main/java/dev/hegel/Stateful.java index c4c74c5..fbe095c 100644 --- a/shared/src/main/java/dev/hegel/Stateful.java +++ b/shared/src/main/java/dev/hegel/Stateful.java @@ -1,11 +1,14 @@ package dev.hegel; import dev.hegel.lowlevel.Abi; +import dev.hegel.lowlevel.LibhegelException; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Comparator; import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; /** * Stateful (model-based) testing: the engine picks which action runs next and this driver applies @@ -48,11 +51,82 @@ * are checked in full on the machine's initial and final state and sampled in between: after any * given rule each invariant runs with probability {@code 1 / stepCount}, so its expected cost per * test case stays constant as the step count grows. The step count is the target number of rules - * per test case, {@link #DEFAULT_STEP_COUNT} unless {@link #run(Object, TestCase, int)} is given - * one. Mark an invariant {@code @Invariant(alwaysRun = - * true)} to check it after every rule instead. Give a rule that should run more often than the - * others a {@link Rule#weight() weight}: {@code @Rule(weight = 5)} is offered about five times as - * often as a plain {@code @Rule}. Use a {@link Pool} to act on previously generated values. + * per test case, {@link #DEFAULT_STEP_COUNT} unless {@link Options#stepCount} sets another. Mark + * an invariant {@code @Invariant(alwaysRun = true)} to check it after every rule instead. Give a + * rule that should run more often than the others a {@link Rule#weight() weight}: {@code + * @Rule(weight = 5)} is offered about five times as often as a plain {@code @Rule}. Use a {@link + * Pool} to act on previously generated values. + * + *

Concurrent machines

+ * + *

To look for concurrency bugs, run the machine with {@link Options#maxConcurrency} above 1: + * + * 

{@code
+ * class Counter {
+ *   private final Map store = new ConcurrentHashMap<>();
+ *   private final AtomicInteger increments = new AtomicInteger();
+ *   private final ConcurrentPool keys;
+ *
+ *   Counter(TestCase tc) {
+ *     keys = new ConcurrentPool<>(tc);
+ *   }
+ *
+ *   @Rule(group = "ops")
+ *   void register(TestCase tc) {
+ *     String key = tc.draw(text().minSize(1).maxSize(3));
+ *     store.putIfAbsent(key, 0);
+ *     keys.add(tc, key);
+ *   }
+ *
+ *   @Rule(group = "ops", weight = 3)
+ *   void increment(TestCase tc) {
+ *     String key = tc.draw(keys.reusable());
+ *     store.put(key, store.get(key) + 1); // racy: a lost update
+ *     increments.incrementAndGet();
+ *   }
+ *
+ *   @Rule(group = "audit")
+ *   void audit(TestCase tc) {
+ *     tc.note("store holds " + store.size() + " keys");
+ *   }
+ *
+ *   @Invariant
+ *   void noLostUpdates(TestCase tc) {
+ *     int total = store.values().stream().mapToInt(Integer::intValue).sum();
+ *     assertEquals(increments.get(), total);
+ *   }
+ * }
+ *
+ * @HegelTest
+ * void counterUnderContention(TestCase tc) {
+ *   Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4));
+ * }
+ * }
+ * + *

For each test case the engine draws a concurrency level between {@link + * Options#minConcurrency} and {@link Options#maxConcurrency} (weighted toward the maximum) and the + * driver runs that many worker threads. Execution proceeds in rounds: the engine picks one + * {@linkplain Rule#group() concurrency group} per round, every worker pulls and applies a few of + * that group's rules concurrently, and once every worker has finished the round the sampled + * invariants run on the driving thread. Rules in the same group may therefore overlap in time; + * rules in different groups never do, and invariants never overlap a rule. The step count bounds + * the number of rounds. + * + *

Rules run concurrently on the same machine object, so any state a rule reads or writes must + * be safe for concurrent access (atomics, {@code synchronized}, concurrent collections). Each rule + * receives its worker's own {@link TestCase}: draw only through it, never retain it past the rule + * or share it with another worker, and use a {@link ConcurrentPool} rather than a {@link Pool} to + * pass generated values between rules. Invariants receive the driving thread's test case. Rules + * must not depend on their relative ordering within a round. + * + *

When a rule fails in some worker, the other workers finish their round before the failure is + * reported, so a rule that blocks forever hangs the test case as it would sequentially. When + * several workers fail in the same round the lowest-numbered worker's exception is reported and the + * others are noted as dropped. The report groups each round's output by worker under the round's + * header, each line stamped {@code [worker N +X.XXXms]} with the time since the machine started, so + * it can be read across workers. A concurrency bug that depends on thread scheduling may not + * reproduce on replay; the engine then confirms it by repeated replay and reports it with a + * caveat (see {@link Failure#caveat()}). */ public final class Stateful { private Stateful() {} @@ -63,9 +137,107 @@ private Stateful() {} */ public static final int DEFAULT_STEP_COUNT = 50; + /** The concurrency group of every {@link Rule @Rule} that names none. */ + public static final String ANONYMOUS_GROUP = ""; + + /** + * How to run a machine: the step count and the concurrency bounds. Immutable; each setter + * returns a new instance. Start from {@link Stateful#options()}. + */ + public static final class Options { + static final Options DEFAULT = new Options(DEFAULT_STEP_COUNT, 1, 1); + + final int stepCount; + final int minConcurrency; + final int maxConcurrency; + + private Options(int stepCount, int minConcurrency, int maxConcurrency) { + this.stepCount = stepCount; + this.minConcurrency = minConcurrency; + this.maxConcurrency = maxConcurrency; + } + + /** + * The target number of steps per test case: rules for a sequential machine, rounds for a + * concurrent one. Every test case runs at least one step and at most {@code stepCount}; + * the engine decides when to stop within that budget. After each step a sampled invariant + * is checked with probability {@code 1 / stepCount}. + * + * @param stepCount the step budget, at least 1 + * @return options with the step count set + * @throws IllegalArgumentException if {@code stepCount} is less than 1 + */ + public Options stepCount(int stepCount) { + if (stepCount < 1) { + throw new IllegalArgumentException("stepCount must be at least 1, got " + stepCount); + } + return new Options(stepCount, minConcurrency, maxConcurrency); + } + + /** + * The fewest worker threads a test case runs. Defaults to 1. + * + * @param minConcurrency the lower concurrency bound, at least 1 + * @return options with the bound set + * @throws IllegalArgumentException if {@code minConcurrency} is less than 1 + */ + public Options minConcurrency(int minConcurrency) { + if (minConcurrency < 1) { + throw new IllegalArgumentException("minConcurrency must be at least 1, got " + minConcurrency); + } + return new Options(stepCount, minConcurrency, maxConcurrency); + } + + /** + * The most worker threads a test case runs. Defaults to 1, which runs the machine + * sequentially on the calling thread; any higher value runs it concurrently (see {@link + * Stateful}), with the engine drawing each test case's level between the bounds. + * + * @param maxConcurrency the upper concurrency bound, at least 1 + * @return options with the bound set + * @throws IllegalArgumentException if {@code maxConcurrency} is less than 1 + */ + public Options maxConcurrency(int maxConcurrency) { + if (maxConcurrency < 1) { + throw new IllegalArgumentException("maxConcurrency must be at least 1, got " + maxConcurrency); + } + return new Options(stepCount, minConcurrency, maxConcurrency); + } + + /** + * @return the step budget per test case + */ + public int stepCount() { + return stepCount; + } + + /** + * @return the lower concurrency bound + */ + public int minConcurrency() { + return minConcurrency; + } + + /** + * @return the upper concurrency bound + */ + public int maxConcurrency() { + return maxConcurrency; + } + } + + /** + * The default options: {@link #DEFAULT_STEP_COUNT} steps, sequential. + * + * @return the defaults, to refine with the {@link Options} setters + */ + public static Options options() { + return Options.DEFAULT; + } + /** * Run {@code machine}'s rules and invariants under {@code tc} for up to {@link - * #DEFAULT_STEP_COUNT} steps. + * #DEFAULT_STEP_COUNT} steps, sequentially. * *

The machine's {@code @Rule}/{@code @Invariant} methods are discovered reflectively from * its class (superclass methods are not considered) and ordered by name, so rule numbering is @@ -75,16 +247,12 @@ private Stateful() {} * @param tc the current test case */ public static void run(Object machine, TestCase tc) { - run(machine, tc, DEFAULT_STEP_COUNT); + run(machine, tc, Options.DEFAULT); } /** * Run {@code machine}'s rules and invariants under {@code tc} for up to {@code stepCount} - * steps. - * - *

Every test case runs at least one step and at most {@code stepCount}; the engine decides - * when to stop within that budget. After each step a sampled invariant is checked with - * probability {@code 1 / stepCount}. + * steps, sequentially. Equivalent to {@code run(machine, tc, options().stepCount(stepCount))}. * * @param machine the state machine to drive * @param tc the current test case @@ -92,8 +260,25 @@ public static void run(Object machine, TestCase tc) { * @throws IllegalArgumentException if {@code stepCount} is less than 1 */ public static void run(Object machine, TestCase tc, int stepCount) { - if (stepCount < 1) { - throw new IllegalArgumentException("stepCount must be at least 1, got " + stepCount); + run(machine, tc, Options.DEFAULT.stepCount(stepCount)); + } + + /** + * Run {@code machine}'s rules and invariants under {@code tc} as {@code options} say: + * sequentially on the calling thread when {@link Options#maxConcurrency} is 1, on worker + * threads otherwise (see {@link Stateful}). + * + * @param machine the state machine to drive + * @param tc the current test case + * @param options the step budget and concurrency bounds + * @throws IllegalArgumentException if the machine has no rules, a rule or invariant has the + * wrong signature or an invalid weight, or {@code minConcurrency} exceeds {@code + * maxConcurrency} + */ + public static void run(Object machine, TestCase tc, Options options) { + if (options.minConcurrency > options.maxConcurrency) { + throw new IllegalArgumentException("minConcurrency (" + options.minConcurrency + + ") must not exceed maxConcurrency (" + options.maxConcurrency + ")"); } List rules = annotated(machine, Rule.class); List invariants = annotated(machine, Invariant.class); @@ -101,12 +286,25 @@ public static void run(Object machine, TestCase tc, int stepCount) { throw new IllegalArgumentException( machine.getClass().getName() + " has no @Rule methods; a state machine needs at least one"); } - long machineId = - tc.newStateMachine(names(rules), weights(rules), names(invariants), alwaysRun(invariants), stepCount); + List groupNames = new ArrayList<>(); + long[] groups = groups(rules, groupNames); + DataSource.StateMachine sm = tc.newStateMachine( + names(rules), + groups, + weights(rules), + names(invariants), + alwaysRun(invariants), + options.minConcurrency, + options.maxConcurrency, + options.stepCount); try { - drive(machine, rules, invariants, tc, machineId); + if (options.maxConcurrency == 1) { + driveSequential(machine, rules, invariants, tc, sm.id()); + } else { + driveConcurrent(machine, rules, invariants, groupNames, tc, sm.id(), sm.concurrency()); + } } finally { - tc.stateMachineFree(machineId); + tc.stateMachineFree(sm.id()); } } @@ -116,7 +314,7 @@ public static void run(Object machine, TestCase tc, int stepCount) { * next_rule} until the join point), and then samples which invariants to check. The start and * end of the machine are unconditional join points where every invariant runs. */ - private static void drive( + private static void driveSequential( Object machine, List rules, List invariants, TestCase tc, long machineId) { if (!invariants.isEmpty()) { tc.note("Checking invariants on the initial state."); @@ -134,14 +332,11 @@ private static void drive( // not protocol: pull rules until the join point. boolean roundRejected = false; while (true) { - long index = tc.stateMachineNextRule(machineId); + long index = tc.stateMachineNextRule(machineId, 0); if (index == Abi.STATE_MACHINE_DONE) { break; } - if (index < 0 || index >= rules.size()) { - throw new HegelException("internal error: state machine chose out-of-range rule index " + index); - } - Method rule = rules.get((int) index); + Method rule = rules.get(ruleIndex(index, rules)); step++; tc.note("Step " + step + ": " + rule.getName()); try { @@ -155,7 +350,7 @@ private static void drive( // The rule's own precondition failed: tell the engine not to count it as a // step, discard the round's span so it retries from before the step, and pull // the next rule. - tc.stateMachineRuleRejected(machineId); + tc.stateMachineRuleRejected(machineId, 0); roundRejected = true; tc.note("Rule stopped early due to violated assumption."); } catch (RuntimeException | Error e) { @@ -178,10 +373,276 @@ private static void drive( } /** - * Run the invariants at a join point: all of them for the guaranteed initial/final checks, or - * only those the engine samples in (always-run invariants unconditionally) when {@code - * sampled}. + * The same protocol driven by worker threads, mirroring hegel-rust's concurrent runner. The + * calling thread owns the root handle: it advances rounds, clones a fresh per-round handle for + * each worker right after the round's header (so the round's lines land under it), waits for + * every worker to report, absorbs their output, resolves the round's outcome and samples the + * invariants. Workers persist for the whole test case and only ever touch their own handle. + * No spans are opened: the engine owns rule structure through {@code next_rule} / {@code + * rule_rejected}, and a round's draws are spread over several streams. */ + private static void driveConcurrent( + Object machine, + List rules, + List invariants, + List groupNames, + TestCase tc, + long machineId, + int concurrency) { + long startNanos = System.nanoTime(); + tc.note("Concurrency level: " + concurrency); + if (!invariants.isEmpty()) { + tc.note("Checking invariants on the initial state."); + } + checkInvariants(machine, invariants, tc, machineId, false); + + Worker[] workers = new Worker[concurrency]; + for (int w = 0; w < concurrency; w++) { + workers[w] = new Worker(w, machine, rules, machineId); + } + List handles = new ArrayList<>(); + try { + for (Worker worker : workers) { + worker.thread.start(); + } + int round = 0; + while (true) { + long group = tc.stateMachineNextGroup(machineId); + if (group == Abi.STATE_MACHINE_DONE) { + break; + } + round++; + tc.note("---------------- Round " + round + ": group \"" + groupNames.get((int) group) + + "\" ----------------"); + for (Worker worker : workers) { + handles.add(tc.forWorker(worker.index, startNanos)); + } + for (int w = 0; w < concurrency; w++) { + workers[w].inbox.add(handles.get(w)); + } + WorkerEvent[] events = new WorkerEvent[concurrency]; + for (int w = 0; w < concurrency; w++) { + events[w] = workers[w].await(); + } + // Every worker has reported, so no handle is in use any more: fold the round's + // output into the report and release the clones before judging the round. + for (TestCase handle : handles) { + tc.absorb(handle); + handle.release(); + } + handles.clear(); + resolveRound(events, tc); + checkInvariants(machine, invariants, tc, machineId, true); + } + if (!invariants.isEmpty()) { + tc.note("Checking invariants on the final state."); + } + checkInvariants(machine, invariants, tc, machineId, false); + } finally { + // Stop the workers first: a handle may only be released once its worker is done. + for (Worker worker : workers) { + worker.stop(); + } + for (TestCase handle : handles) { + handle.release(); + } + } + } + + /** + * Judge a finished round from its workers' events, in hegel-rust's precedence: a binding or + * usage error (or a worker that died without reporting) is rethrown first; then an overrun or + * an engine-level rejection concludes the whole case, dropping any failures found alongside + * (an abandoned rule can leave the machine's locks and state inconsistent, so those are not + * trustworthy); otherwise the lowest-numbered failing worker's exception is rethrown as-is and + * the rest are noted as dropped. + */ + private static void resolveRound(WorkerEvent[] events, TestCase tc) { + for (WorkerEvent event : events) { + if (event.kind == WorkerEvent.Kind.CONTROL) { + throw event.error; + } + } + boolean overrun = false; + boolean invalid = false; + for (WorkerEvent event : events) { + overrun |= event.kind == WorkerEvent.Kind.OVERRUN; + invalid |= event.kind == WorkerEvent.Kind.INVALID; + } + Throwable winner = null; + for (int w = 0; w < events.length; w++) { + WorkerEvent event = events[w]; + if (event.kind != WorkerEvent.Kind.FAILED) { + continue; + } + if (overrun || invalid || winner != null) { + tc.note("Dropped concurrent failure from worker " + w + ": " + event.failure); + } else { + winner = event.failure; + } + } + if (overrun) { + throw new StopTest(); + } + if (invalid) { + throw new AssumeRejected(); + } + if (winner instanceof Error error) { + throw error; + } + if (winner != null) { + throw (RuntimeException) winner; + } + } + + /** What a worker reports at the end of a round. */ + private static final class WorkerEvent { + enum Kind { + /** The worker's rule stream is exhausted. */ + DONE, + /** A binding or usage error, or the worker died: {@link #error} is rethrown verbatim. */ + CONTROL, + /** The engine's choice budget ran out ({@link StopTest}). */ + OVERRUN, + /** The engine concluded the case invalid ({@link AssumeRejected} outside a rule's own precondition). */ + INVALID, + /** A rule threw: {@link #failure} is the property's failure. */ + FAILED + } + + static final WorkerEvent DONE = new WorkerEvent(Kind.DONE, null, null); + static final WorkerEvent OVERRUN = new WorkerEvent(Kind.OVERRUN, null, null); + static final WorkerEvent INVALID = new WorkerEvent(Kind.INVALID, null, null); + + final Kind kind; + final RuntimeException error; + /** A {@link RuntimeException} or an {@link Error}: {@link #invokeMachineMethod} wraps everything else. */ + final Throwable failure; + + WorkerEvent(Kind kind, RuntimeException error, Throwable failure) { + this.kind = kind; + this.error = error; + this.failure = failure; + } + + static WorkerEvent control(RuntimeException error) { + return new WorkerEvent(Kind.CONTROL, error, null); + } + + static WorkerEvent failed(Throwable failure) { + return new WorkerEvent(Kind.FAILED, null, failure); + } + } + + /** + * A worker thread: per round it receives a handle through its inbox, pulls and applies rules + * on it until the engine signals its join point, and reports one event through its outbox. + * Whatever happens, it reports — a worker that exits its loop leaves a control error behind — + * so the driving thread never waits forever on a live queue. + */ + private static final class Worker { + /** Inbox sentinel: the test case is over, exit. */ + private static final Object STOP = new Object(); + + final int index; + final Thread thread; + final BlockingQueue inbox = new LinkedBlockingQueue<>(); + final BlockingQueue outbox = new LinkedBlockingQueue<>(); + private final Object machine; + private final List rules; + private final long machineId; + + Worker(int index, Object machine, List rules, long machineId) { + this.index = index; + this.machine = machine; + this.rules = rules; + this.machineId = machineId; + this.thread = new Thread(this::loop, "hegel-worker-" + index); + this.thread.setDaemon(true); + } + + private void loop() { + try { + while (true) { + Object command = inbox.take(); + if (command == STOP) { + return; + } + outbox.add(runRound((TestCase) command)); + } + } catch (InterruptedException e) { + // Interrupted between rounds (a rule set the flag): fall through and report it. + Thread.currentThread().interrupt(); + } finally { + // Whatever ended the loop, leave an event behind so the driver never waits forever. + outbox.add(WorkerEvent.control(new HegelException( + "internal error: concurrent worker " + index + " exited without reporting its round"))); + } + } + + /** Pull and apply rules on {@code wtc} until the engine ends this worker's round. */ + private WorkerEvent runRound(TestCase wtc) { + try { + while (true) { + long index = wtc.stateMachineNextRule(machineId, this.index); + if (index == Abi.STATE_MACHINE_DONE) { + return WorkerEvent.DONE; + } + Method rule = rules.get(ruleIndex(index, rules)); + wtc.note("Rule: " + rule.getName()); + try { + invokeMachineMethod(machine, rule, wtc); + } catch (AssumeRejected e) { + if (wtc.isAborted()) { + // The engine concluded the case invalid inside the rule, not the + // rule's own precondition: classified below. + throw e; + } + // The engine does not count the slot; the next next_rule retries it. + wtc.stateMachineRuleRejected(machineId, this.index); + wtc.note("Rule stopped early due to violated assumption."); + } + } + } catch (AssumeRejected e) { + return WorkerEvent.INVALID; + } catch (StopTest e) { + return WorkerEvent.OVERRUN; + } catch (LibhegelException | IllegalArgumentException e) { + return WorkerEvent.control(e); + } catch (Throwable e) { + return WorkerEvent.failed(e); + } + } + + /** Block until this worker reports its round. */ + WorkerEvent await() { + try { + return outbox.take(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HegelException("interrupted while waiting for concurrent worker " + index, e); + } + } + + /** Tell the worker the test case is over and wait for it to exit. */ + void stop() { + inbox.add(STOP); + boolean interrupted = false; + while (true) { + try { + thread.join(); + break; + } catch (InterruptedException e) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + + /** Run the invariants at a join point: all of them, or only those the engine samples in. */ private static void checkInvariants( Object machine, List invariants, TestCase tc, long machineId, boolean sampled) { for (int i = 0; i < invariants.size(); i++) { @@ -192,6 +653,13 @@ private static void checkInvariants( } } + private static int ruleIndex(long index, List rules) { + if (index < 0 || index >= rules.size()) { + throw new HegelException("internal error: state machine chose out-of-range rule index " + index); + } + return (int) index; + } + /** * Invoke a rule/invariant method, unwrapping reflection's exception wrapper so the method's own * exception propagates. A checked exception (which a rule may declare but this driver cannot @@ -243,6 +711,28 @@ private static List names(List methods) { return methods.stream().map(Method::getName).toList(); } + /** + * The rules' concurrency-group ids, parallel to {@code rules}: distinct group names numbered + * from 0 in first-appearance order, appended to {@code groupNames} so the id indexes it. A + * sequential machine's all-anonymous rules land in one group, id 0. + */ + private static long[] groups(List rules, List groupNames) { + long[] ids = new long[rules.size()]; + for (int i = 0; i < ids.length; i++) { + String group = rules.get(i).getAnnotation(Rule.class).group(); + if (group.isEmpty()) { + group = ANONYMOUS_GROUP; + } + int id = groupNames.indexOf(group); + if (id < 0) { + id = groupNames.size(); + groupNames.add(group); + } + ids[i] = id; + } + return ids; + } + /** * The rules' selection weights, parallel to {@code rules}, or {@code null} — the engine's * all-equal default — when no rule asks for anything but weight 1. diff --git a/shared/src/main/java/dev/hegel/TestCase.java b/shared/src/main/java/dev/hegel/TestCase.java index 674c32d..0e40dca 100644 --- a/shared/src/main/java/dev/hegel/TestCase.java +++ b/shared/src/main/java/dev/hegel/TestCase.java @@ -8,9 +8,9 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.UUID; -import java.util.function.Consumer; import java.util.function.Supplier; /** @@ -40,17 +40,59 @@ public final class TestCase { private final boolean reporting; private final Reporter reporter; + /** + * The root handle of this test case, whose draw-name counter every handle of the case shares: + * {@code this} for the root, the root for a worker handle (see {@link #forWorker}). + */ + private final TestCase root; + /** The concurrent worker this is a per-round handle for, or {@code -1} for the root. */ + private final int worker; + /** Reference instant for a worker handle's line stamps ({@link System#nanoTime()}). */ + private final long startNanos; + private final Map draws = new LinkedHashMap<>(); private final List notes = new ArrayList<>(); - /** The recorded draws and notes in report order, each as the reporter call that replays it. */ - private final List> events = new ArrayList<>(); - /** Uses per draw name, for numbering repeats ({@code x}, {@code x_2}, ...; {@code draw_N}). */ + /** + * The recorded draws and notes in report order. On a worker handle this is the round's + * buffer, absorbed by the root at the join point ({@link #absorb}). + */ + private final List events = new ArrayList<>(); + /** + * Uses per draw name, for numbering repeats ({@code x}, {@code x_2}, ...; {@code draw_N}). + * Only the root's is used; worker handles go through {@link #nextName} on the root. + */ private final Map nameUses = new HashMap<>(); /** Notes made while a top-level draw is in progress; flushed after that draw's line. */ private final List pendingNotes = new ArrayList<>(); private int drawDepth; + /** A recorded draw ({@code message == null}) or note, as the reporter would be told about it. */ + private static final class Event { + /** The draw's name as reported (stamped with the worker prefix on a worker handle). */ + final String name; + /** The draw's plain name, the key it is recorded under. */ + final String key; + + final Object value; + final String message; + + Event(String name, String key, Object value, String message) { + this.name = name; + this.key = key; + this.value = value; + this.message = message; + } + + void replay(Reporter reporter, boolean finalReplay) { + if (message != null) { + reporter.note(message, finalReplay); + } else { + reporter.draw(name, value, finalReplay); + } + } + } + TestCase(DataSource source, boolean captured, Reporter reporter) { this(source, captured, false, reporter); } @@ -60,6 +102,84 @@ public final class TestCase { this.captured = captured; this.reporting = verbose; this.reporter = reporter; + this.root = this; + this.worker = -1; + this.startNanos = 0; + } + + private TestCase(DataSource source, TestCase root, int worker, long startNanos) { + this.source = source; + this.captured = root.captured; + this.reporting = root.reporting; + this.reporter = root.reporter; + this.root = root; + this.worker = worker; + this.startNanos = startNanos; + } + + /** + * A handle for concurrent worker {@code worker} to draw through for one round of a stateful + * machine: an independent choice stream of the same case (see {@link + * DataSource#cloneForWorker}) whose draws and notes are buffered, stamped {@code [worker N + * +X.XXXms]} with the time since {@code startNanos}, and handed to this root at the join point + * by {@link #absorb}. Release it with {@link #release()} once the round is over. + */ + TestCase forWorker(int worker, long startNanos) { + return new TestCase(source.cloneForWorker(worker), this, worker, startNanos); + } + + /** The worker this handle belongs to, or {@code -1} for the root. */ + int worker() { + return worker; + } + + /** Free the clone handle behind a worker handle from {@link #forWorker}. */ + void release() { + source.release(); + } + + /** + * Take over a worker handle's buffered draws and notes: record them here (on a captured + * execution) and hand them to the reporter (on a verbose one), in the order the worker made + * them. Called on the root at a join point, once the worker has finished its round, so the + * reporter only ever hears from the driving thread. + */ + void absorb(TestCase workerHandle) { + for (Event event : workerHandle.events) { + record(event); + } + workerHandle.events.clear(); + } + + /** The stamp a worker handle's lines carry; empty on the root. */ + private String prefix() { + if (worker < 0) { + return ""; + } + double elapsedMillis = (System.nanoTime() - startNanos) / 1e6; + return String.format(Locale.ROOT, "[worker %d +%.3fms] ", worker, elapsedMillis); + } + + /** + * Deliver a draw or note: buffered on a worker handle until the root absorbs it; on the root, + * kept for the report when captured and reported live when verbose. + */ + private void record(Event event) { + if (worker >= 0) { + events.add(event); + return; + } + if (captured) { + events.add(event); + if (event.message != null) { + notes.add(event.message); + } else { + draws.put(event.key, event.value); + } + } + if (reporting) { + event.replay(reporter, false); + } } /** @@ -112,14 +232,8 @@ public T draw(Generator generator, String label) { } if (top) { if (captured || reporting) { - String name = displayName(label); - if (captured) { - draws.put(name, value); - events.add(r -> r.draw(name, value, true)); - } - if (reporting) { - reporter.draw(name, value, false); - } + String name = root.nextName(label); + record(new Event(prefix() + name, name, value, null)); } flushPendingNotes(); } @@ -129,9 +243,10 @@ public T draw(Generator generator, String label) { /** * The name a top-level draw reports under: a label prints bare the first time and numbered from * its second use ({@code x}, {@code x_2}, ...); an unlabelled draw is {@code draw_N}, counting - * unlabelled draws only. + * unlabelled draws only. Names are unique across the whole case — worker handles number + * through the root — hence synchronized: concurrent workers name their draws at the same time. */ - private String displayName(String label) { + private synchronized String nextName(String label) { String base = label != null ? label : "draw"; int uses = nameUses.merge(base, 1, Integer::sum); if (label == null) { @@ -148,13 +263,7 @@ private void flushPendingNotes() { } private void emitNote(String message) { - if (captured) { - notes.add(message); - events.add(r -> r.note(message, true)); - } - if (reporting) { - reporter.note(message, false); - } + record(new Event(null, null, null, prefix() + message)); } /** @@ -394,25 +503,36 @@ long poolGenerate(long poolId, boolean consume) { return source.poolGenerate(poolId, consume); } - long newStateMachine( + DataSource.StateMachine newStateMachine( List ruleNames, + long[] ruleGroups, double[] ruleWeights, List invariantNames, boolean[] invariantAlwaysCheck, + long minConcurrency, + long maxConcurrency, int stepCount) { - return source.newStateMachine(ruleNames, ruleWeights, invariantNames, invariantAlwaysCheck, stepCount); + return source.newStateMachine( + ruleNames, + ruleGroups, + ruleWeights, + invariantNames, + invariantAlwaysCheck, + minConcurrency, + maxConcurrency, + stepCount); } long stateMachineNextGroup(long stateMachineId) { return source.stateMachineNextGroup(stateMachineId); } - long stateMachineNextRule(long stateMachineId) { - return source.stateMachineNextRule(stateMachineId); + long stateMachineNextRule(long stateMachineId, long workerIndex) { + return source.stateMachineNextRule(stateMachineId, workerIndex); } - void stateMachineRuleRejected(long stateMachineId) { - source.stateMachineRuleRejected(stateMachineId); + void stateMachineRuleRejected(long stateMachineId, long workerIndex) { + source.stateMachineRuleRejected(stateMachineId, workerIndex); } boolean stateMachineShouldCheckInvariant(long stateMachineId, long invariantIndex) { @@ -440,8 +560,8 @@ List notes() { /** Hand the recorded draws and notes to {@code reporter}, in report order, flagged as a replay. */ void replayTo(Reporter reporter) { - for (Consumer event : events) { - event.accept(reporter); + for (Event event : events) { + event.replay(reporter, true); } } diff --git a/shared/src/test/java/dev/hegel/BindingErrorPathsTest.java b/shared/src/test/java/dev/hegel/BindingErrorPathsTest.java index 1b1ceea..797960a 100644 --- a/shared/src/test/java/dev/hegel/BindingErrorPathsTest.java +++ b/shared/src/test/java/dev/hegel/BindingErrorPathsTest.java @@ -53,15 +53,76 @@ void okPathsReturnTheEngineValues() { assertEquals(3, ds.newPool()); assertEquals(0, ds.poolAdd(3)); assertEquals(0, ds.poolGenerate(3, true)); - assertEquals(5, ds.newStateMachine(List.of("r"), null, List.of("i"), new boolean[] {false}, 50)); + fake.stateMachineConcurrency = 3; + DataSource.StateMachine sm = + ds.newStateMachine(List.of("r"), new long[] {0}, null, List.of("i"), new boolean[] {false}, 1, 4, 50); + assertEquals(5, sm.id()); + assertEquals(3, sm.concurrency()); + assertEquals(1, fake.stateMachineMinConcurrency); + assertEquals(4, fake.stateMachineMaxConcurrency); assertEquals(Abi.STATE_MACHINE_DONE, ds.stateMachineNextGroup(5)); - assertEquals(Abi.STATE_MACHINE_DONE, ds.stateMachineNextRule(5)); - ds.stateMachineRuleRejected(5); + assertEquals(Abi.STATE_MACHINE_DONE, ds.stateMachineNextRule(5, 2)); + ds.stateMachineRuleRejected(5, 2); + assertEquals(List.of(2L), fake.nextRuleWorkers); + assertEquals(List.of(2L), fake.rejectedWorkers); assertTrue(ds.stateMachineShouldCheckInvariant(5, 0)); ds.stateMachineFree(5); assertEquals(1, fake.freedStateMachines); } + @Test + void clonesDrawThroughTheirOwnHandleAndAreReleased() { + FakeLibhegel fake = new FakeLibhegel(); + fake.integerValue = 4L; + LiveDataSource ds = source(fake); + DataSource clone = ds.cloneForWorker(2); + assertEquals(List.of(FakeLibhegel.TC), fake.clonesMade); + assertEquals(2L, fake.workersSet.get(FakeLibhegel.CLONE_BASE)); + assertEquals(4, clone.generateInteger(0, 10)); + clone.release(); + assertEquals(List.of(FakeLibhegel.CLONE_BASE), fake.freedClones); + // The root handle is untouched by the clone's release. + assertEquals(1, fake.freedTestCases); + } + + @Test + void cloneFailuresPropagateAndLeakNothing() { + FakeLibhegel cloneFails = new FakeLibhegel(); + cloneFails.cloneRc = Abi.E_BACKEND; + assertThrows(HegelException.class, () -> source(cloneFails).cloneForWorker(0)); + assertTrue(cloneFails.freedClones.isEmpty()); + + // A clone past the end of a replayed sequence is an overrun, like any other draw. + FakeLibhegel exhausted = new FakeLibhegel(); + exhausted.cloneRc = Abi.E_STOP_TEST; + LiveDataSource ds0 = source(exhausted); + assertThrows(StopTest.class, () -> ds0.cloneForWorker(0)); + assertTrue(ds0.isAborted()); + + FakeLibhegel workerFails = new FakeLibhegel(); + workerFails.setWorkerFails = true; + assertThrows(HegelException.class, () -> source(workerFails).cloneForWorker(0)); + // The clone was made, so it is released before the failure surfaces. + assertEquals(List.of(FakeLibhegel.CLONE_BASE), workerFails.freedClones); + + // Cloning is a draw: it short-circuits once the case is aborted. + FakeLibhegel aborted = new FakeLibhegel(); + aborted.generateIntegerRc = Abi.E_STOP_TEST; + LiveDataSource ds = source(aborted); + assertThrows(StopTest.class, () -> ds.generateInteger(0, 1)); + assertThrows(StopTest.class, () -> ds.cloneForWorker(0)); + assertTrue(aborted.clonesMade.isEmpty()); + } + + @Test + void concurrentUseOfOneHandleIsExplained() { + FakeLibhegel fake = new FakeLibhegel(); + fake.poolAddRc = Abi.E_CONCURRENT_USE; + HegelException e = assertThrows(HegelException.class, () -> source(fake).poolAdd(3)); + assertTrue(e.getMessage().contains("two threads"), e.getMessage()); + assertTrue(e.getMessage().contains("ConcurrentPool"), e.getMessage()); + } + @Test void stringGeneratorConstructionAndOwnership() { FakeLibhegel fake = new FakeLibhegel(); @@ -112,10 +173,12 @@ void stopTestUnwindsAndAbortsEverything() { assertThrows(StopTest.class, () -> ds.newPool()); assertThrows(StopTest.class, () -> ds.poolAdd(1)); assertThrows(StopTest.class, () -> ds.poolGenerate(1, false)); - assertThrows(StopTest.class, () -> ds.newStateMachine(List.of("r"), null, List.of(), new boolean[0], 50)); + assertThrows( + StopTest.class, + () -> ds.newStateMachine(List.of("r"), new long[] {0}, null, List.of(), new boolean[0], 1, 1, 50)); assertThrows(StopTest.class, () -> ds.stateMachineNextGroup(1)); - assertThrows(StopTest.class, () -> ds.stateMachineNextRule(1)); - assertThrows(StopTest.class, () -> ds.stateMachineRuleRejected(1)); + assertThrows(StopTest.class, () -> ds.stateMachineNextRule(1, 0)); + assertThrows(StopTest.class, () -> ds.stateMachineRuleRejected(1, 0)); assertThrows(StopTest.class, () -> ds.stateMachineShouldCheckInvariant(1, 0)); // Freeing the machine is not a draw: it must still work once the case is aborted. ds.stateMachineFree(1); @@ -218,7 +281,8 @@ void poolAndStateMachineErrorsPropagate() { sm.newStateMachineRc = Abi.E_INVALID_ARG; assertThrows( IllegalArgumentException.class, - () -> source(sm).newStateMachine(List.of("r"), null, List.of(), new boolean[0], 50)); + () -> source(sm) + .newStateMachine(List.of("r"), new long[] {0}, null, List.of(), new boolean[0], 1, 1, 50)); FakeLibhegel group = new FakeLibhegel(); group.stateMachineNextGroupRc = Abi.E_STOP_TEST; @@ -226,11 +290,11 @@ void poolAndStateMachineErrorsPropagate() { FakeLibhegel next = new FakeLibhegel(); next.stateMachineNextRuleRc = Abi.E_STOP_TEST; - assertThrows(StopTest.class, () -> source(next).stateMachineNextRule(1)); + assertThrows(StopTest.class, () -> source(next).stateMachineNextRule(1, 0)); FakeLibhegel rejected = new FakeLibhegel(); rejected.stateMachineRuleRejectedRc = Abi.E_INVALID_ARG; - assertThrows(IllegalArgumentException.class, () -> source(rejected).stateMachineRuleRejected(1)); + assertThrows(IllegalArgumentException.class, () -> source(rejected).stateMachineRuleRejected(1, 0)); FakeLibhegel check = new FakeLibhegel(); check.stateMachineShouldCheckInvariantRc = Abi.E_STOP_TEST; diff --git a/shared/src/test/java/dev/hegel/ConcurrentCounterExample.java b/shared/src/test/java/dev/hegel/ConcurrentCounterExample.java new file mode 100644 index 0000000..bc39daa --- /dev/null +++ b/shared/src/test/java/dev/hegel/ConcurrentCounterExample.java @@ -0,0 +1,66 @@ +package dev.hegel; + +import static dev.hegel.Generators.text; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; + +/** + * The concurrent stateful example from the release notes and the {@link Stateful} Javadoc, as a + * runnable demonstration. It is meant to fail — the counter loses updates under contention — so it + * is skipped unless asked for: + * + * 
+ * mvn test -pl hegel -am -Dtest=ConcurrentCounterExample -Dhegel.examples=true \
+ *     -Dsurefire.failIfNoSpecifiedTests=false
+ *
+ * + * (or {@code -pl hegel-jna -am} for the JNA frontend). The report shows the round headers and the + * worker-stamped rule and draw lines that led to the lost update. + */ +@EnabledIfSystemProperty(named = "hegel.examples", matches = "true") +class ConcurrentCounterExample { + static final class Counter { + private final Map store = new ConcurrentHashMap<>(); + private final AtomicInteger increments = new AtomicInteger(); + private final ConcurrentPool keys; + + Counter(TestCase tc) { + keys = new ConcurrentPool<>(tc); + } + + @Rule(group = "ops") + void register(TestCase tc) { + String key = tc.draw(text().minSize(1).maxSize(3)); + store.putIfAbsent(key, 0); + keys.add(tc, key); + } + + @Rule(group = "ops", weight = 3) + void increment(TestCase tc) { + String key = tc.draw(keys.reusable()); + store.put(key, store.get(key) + 1); // racy: a lost update + increments.incrementAndGet(); + } + + @Rule(group = "audit") + void audit(TestCase tc) { + tc.note("store holds " + store.size() + " keys"); + } + + @Invariant + void noLostUpdates(TestCase tc) { + assertEquals( + increments.get(), + store.values().stream().mapToInt(Integer::intValue).sum()); + } + } + + @HegelTest(database = Database.DISABLED) + void counterUnderContention(TestCase tc) { + Stateful.run(new Counter(tc), tc, Stateful.options().maxConcurrency(4)); + } +} diff --git a/shared/src/test/java/dev/hegel/ConcurrentStatefulDriverTest.java b/shared/src/test/java/dev/hegel/ConcurrentStatefulDriverTest.java new file mode 100644 index 0000000..5c9f801 --- /dev/null +++ b/shared/src/test/java/dev/hegel/ConcurrentStatefulDriverTest.java @@ -0,0 +1,428 @@ +package dev.hegel; + +import static dev.hegel.Generators.integers; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.hegel.lowlevel.Abi; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; + +/** + * The concurrent stateful driver over the fake engine: per-worker rule queues make the round + * protocol, the join-point output layout and every outcome-resolution path deterministic. + */ +class ConcurrentStatefulDriverTest { + private static final Pattern STAMP = Pattern.compile("^\\[worker (\\d+) \\+\\d+\\.\\d{3}ms\\] "); + + /** A fake reporting {@code concurrency} workers and playing the given rounds. */ + private static FakeLibhegel concurrentFake(int concurrency, long[][]... rounds) { + FakeLibhegel fake = new FakeLibhegel(); + fake.stateMachineConcurrency = concurrency; + fake.concurrentRounds = rounds; + return fake; + } + + private static TestCase exploringCase(FakeLibhegel fake) { + return new TestCase(new LiveDataSource(fake, FakeLibhegel.TC), false, Reporter.silent()); + } + + private static TestCase capturedCase(FakeLibhegel fake) { + return new TestCase(new LiveDataSource(fake, FakeLibhegel.TC), true, Reporter.silent()); + } + + private static Stateful.Options twoWorkers() { + return Stateful.options().minConcurrency(2).maxConcurrency(2); + } + + /** Notes with the timing dropped from their worker stamps: {@code [worker N] ...}. */ + private static List unstamped(List lines) { + return lines.stream() + .map(line -> STAMP.matcher(line).replaceFirst("[worker $1] ")) + .toList(); + } + + private static List sorted(List values) { + List copy = new ArrayList<>(values); + Collections.sort(copy); + return copy; + } + + /** Rules alpha (anonymous), beta and gamma (group "ops"); beta always rejects, gamma draws. */ + static final class Recording { + final List applied = Collections.synchronizedList(new ArrayList<>()); + final Set threads = ConcurrentHashMap.newKeySet(); + final Set workers = ConcurrentHashMap.newKeySet(); + int sampledChecks; + int alwaysChecks; + + @Rule + void alpha(TestCase t) { + applied.add("alpha"); + threads.add(Thread.currentThread().getName()); + workers.add(t.worker()); + } + + @Rule(group = "ops") + void beta(TestCase t) { + applied.add("beta"); + t.assume(false); + } + + @Rule(group = "ops") + void gamma(TestCase t) { + applied.add("gamma"); + t.draw(integers().min(0).max(9), "x"); + } + + @Invariant + void sampled(TestCase t) { + sampledChecks++; + } + + @Invariant(alwaysRun = true) + void unsampled(TestCase t) { + alwaysChecks++; + } + } + + @Test + void driverFollowsTheRoundProtocolOnWorkerThreads() { + // Round 1: worker 0 runs alpha then gamma, worker 1 runs gamma. Round 2: only worker 1 runs alpha. + FakeLibhegel fake = concurrentFake(2, new long[][] {{0, 2}, {2}}, new long[][] {{}, {0}}); + Recording machine = new Recording(); + TestCase tc = capturedCase(fake); + Stateful.run(machine, tc, twoWorkers()); + + // Registration: groups numbered in first-appearance order over the name-sorted rules. + assertEquals(List.of("alpha", "beta", "gamma"), fake.stateMachineRules); + assertArrayEquals(new long[] {0, 1, 1}, fake.stateMachineRuleGroups); + assertEquals(2, fake.stateMachineMinConcurrency); + assertEquals(2, fake.stateMachineMaxConcurrency); + // Rules ran on worker threads, each worker pulling with its own index until its join point. + assertEquals(List.of("alpha", "alpha", "gamma", "gamma"), sorted(machine.applied)); + assertTrue(machine.threads.stream().allMatch(n -> n.startsWith("hegel-worker-")), machine.threads.toString()); + assertEquals(Set.of(0, 1), machine.workers); + assertEquals(-1, tc.worker()); + assertEquals(4, fake.nextRuleWorkers.stream().filter(w -> w == 0).count()); + assertEquals(4, fake.nextRuleWorkers.stream().filter(w -> w == 1).count()); + // One clone per worker per round, attributed to its worker, every one released; the machine too. + assertEquals(4, fake.clonesMade.size()); + for (int i = 0; i < 4; i++) { + assertEquals(Long.valueOf(i % 2), fake.workersSet.get(FakeLibhegel.CLONE_BASE + i)); + } + assertEquals( + List.of( + FakeLibhegel.CLONE_BASE, + FakeLibhegel.CLONE_BASE + 1, + FakeLibhegel.CLONE_BASE + 2, + FakeLibhegel.CLONE_BASE + 3), + fake.freedClones); + assertEquals(1, fake.freedStateMachines); + // Invariants: initial, one sampled join point per round (the fake samples everything in), final. + assertEquals(4, machine.sampledChecks); + assertEquals(4, machine.alwaysChecks); + assertEquals(List.of(0L, 1L, 0L, 1L), fake.invariantChecksAsked); + // Output: each round's worker lines follow its header, grouped by worker index. + assertEquals( + List.of( + "Concurrency level: 2", + "Checking invariants on the initial state.", + "---------------- Round 1: group \"\" ----------------", + "[worker 0] Rule: alpha", + "[worker 0] Rule: gamma", + "[worker 1] Rule: gamma", + "---------------- Round 2: group \"\" ----------------", + "[worker 1] Rule: alpha", + "Checking invariants on the final state."), + unstamped(tc.notes())); + // Draw names are unique across workers and recorded plain; the replayed line carries the stamp. + assertEquals(Set.of("x", "x_2"), tc.draws().keySet()); + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + tc.replayTo(Reporter.printing(new PrintStream(buf, true, StandardCharsets.UTF_8))); + List printed = + unstamped(List.of(buf.toString(StandardCharsets.UTF_8).split("\\R"))); + assertTrue( + printed.contains("[worker 0] x = 0;") || printed.contains("[worker 0] x_2 = 0;"), printed.toString()); + assertTrue( + printed.indexOf("[worker 0] Rule: gamma") < printed.indexOf("[worker 1] Rule: gamma"), + printed.toString()); + } + + @Test + void verboseRunsHearWorkerLinesAtTheJoinPoint() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{0}, {2}}); + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + TestCase tc = new TestCase( + new LiveDataSource(fake, FakeLibhegel.TC), + false, + true, + Reporter.printing(new PrintStream(buf, true, StandardCharsets.UTF_8))); + Stateful.run(new Recording(), tc, twoWorkers()); + List printed = + unstamped(List.of(buf.toString(StandardCharsets.UTF_8).split("\\R"))); + assertTrue(printed.contains("[worker 0] Rule: alpha"), printed.toString()); + assertTrue(printed.contains("[worker 1] Rule: gamma"), printed.toString()); + assertTrue(printed.contains("[worker 1] x = 0;"), printed.toString()); + // Not captured: nothing is kept for a report. + assertTrue(tc.notes().isEmpty()); + } + + @Test + void roundHeadersNameTheGroupAndRejectionsAreReportedPerWorker() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{1}, {}}); + fake.stateMachineGroupId = 1; // the "ops" group + TestCase tc = capturedCase(fake); + Stateful.run(new Recording(), tc, twoWorkers()); + List notes = unstamped(tc.notes()); + assertTrue(notes.contains("---------------- Round 1: group \"ops\" ----------------"), notes.toString()); + int rule = notes.indexOf("[worker 0] Rule: beta"); + assertTrue(rule >= 0, notes.toString()); + assertEquals("[worker 0] Rule stopped early due to violated assumption.", notes.get(rule + 1)); + assertEquals(List.of(0L), fake.rejectedWorkers); + } + + @Test + void anEngineLevelRejectionInsideARuleInvalidatesTheCase() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{2}, {}}); + fake.generateIntegerRc = Abi.E_ASSUME; + TestCase tc = exploringCase(fake); + assertThrows(AssumeRejected.class, () -> Stateful.run(new Recording(), tc, twoWorkers())); + assertEquals(0, fake.rejectedRules); + assertEquals(2, fake.freedClones.size()); + assertEquals(1, fake.freedStateMachines); + } + + /** Rules that end a round each in their own way: bad (usage error), crashes, draws, fails. */ + static final class Outcomes { + @Rule + void bad(TestCase t) { + throw new IllegalArgumentException("usage"); + } + + @Rule + void crashes(TestCase t) { + throw new IllegalStateException("crashed"); + } + + @Rule + void draws(TestCase t) { + t.draw(integers()); + } + + @Rule + void fails(TestCase t) { + throw new AssertionError("boom"); + } + } + + @Test + void anOverrunOutranksAFailureFoundAlongside() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{2}, {3}}); + fake.generateIntegerRc = Abi.E_STOP_TEST; + TestCase tc = capturedCase(fake); + assertThrows(StopTest.class, () -> Stateful.run(new Outcomes(), tc, twoWorkers())); + assertTrue( + tc.notes().contains("Dropped concurrent failure from worker 1: java.lang.AssertionError: boom"), + tc.notes().toString()); + } + + @Test + void anInvalidConclusionOutranksAFailureFoundAlongside() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{3}, {2}}); + fake.generateIntegerRc = Abi.E_ASSUME; + TestCase tc = capturedCase(fake); + assertThrows(AssumeRejected.class, () -> Stateful.run(new Outcomes(), tc, twoWorkers())); + assertTrue( + tc.notes().contains("Dropped concurrent failure from worker 0: java.lang.AssertionError: boom"), + tc.notes().toString()); + } + + @Test + void controlErrorsOutrankFailures() { + // A usage error in worker 1 beats a failure in worker 0. + FakeLibhegel usage = concurrentFake(2, new long[][] {{3}, {0}}); + IllegalArgumentException e = assertThrows( + IllegalArgumentException.class, () -> Stateful.run(new Outcomes(), exploringCase(usage), twoWorkers())); + assertEquals("usage", e.getMessage()); + + // A binding error inside a worker (here from rule_rejected) surfaces verbatim. + FakeLibhegel binding = concurrentFake(2, new long[][] {{1}, {}}); + binding.stateMachineRuleRejectedRc = Abi.E_BACKEND; + assertThrows(HegelException.class, () -> Stateful.run(new Recording(), exploringCase(binding), twoWorkers())); + assertEquals(2, binding.freedClones.size()); + } + + @Test + void theLowestWorkersFailureWinsAndTheRestAreNoted() { + FakeLibhegel errorFirst = concurrentFake(2, new long[][] {{3}, {1}}); + TestCase tc = capturedCase(errorFirst); + AssertionError error = assertThrows(AssertionError.class, () -> Stateful.run(new Outcomes(), tc, twoWorkers())); + assertEquals("boom", error.getMessage()); + assertTrue( + tc.notes() + .contains("Dropped concurrent failure from worker 1: java.lang.IllegalStateException: crashed"), + tc.notes().toString()); + + FakeLibhegel exceptionFirst = concurrentFake(2, new long[][] {{1}, {3}}); + TestCase tc2 = capturedCase(exceptionFirst); + IllegalStateException crashed = + assertThrows(IllegalStateException.class, () -> Stateful.run(new Outcomes(), tc2, twoWorkers())); + assertEquals("crashed", crashed.getMessage()); + assertTrue( + tc2.notes().contains("Dropped concurrent failure from worker 1: java.lang.AssertionError: boom"), + tc2.notes().toString()); + } + + @Test + void anOutOfRangeRuleIndexIsAnInternalError() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{9}, {}}); + HegelException e = assertThrows( + HegelException.class, () -> Stateful.run(new Recording(), exploringCase(fake), twoWorkers())); + assertTrue(e.getMessage().contains("out-of-range"), e.getMessage()); + } + + @Test + void cloneFailuresAbortTheCaseAndReleaseEverything() { + FakeLibhegel cloneFails = concurrentFake(2, new long[][] {{0}, {}}); + cloneFails.cloneRc = Abi.E_BACKEND; + assertThrows( + HegelException.class, () -> Stateful.run(new Recording(), exploringCase(cloneFails), twoWorkers())); + assertEquals(1, cloneFails.freedStateMachines); + + // On a replay of a shorter sequence the clone itself overruns; the case is an overrun. + FakeLibhegel exhausted = concurrentFake(2, new long[][] {{0}, {}}); + exhausted.cloneRc = Abi.E_STOP_TEST; + assertThrows(StopTest.class, () -> Stateful.run(new Recording(), exploringCase(exhausted), twoWorkers())); + assertEquals(1, exhausted.freedStateMachines); + + FakeLibhegel workerFails = concurrentFake(2, new long[][] {{0}, {}}); + workerFails.setWorkerFails = true; + assertThrows( + HegelException.class, () -> Stateful.run(new Recording(), exploringCase(workerFails), twoWorkers())); + assertEquals(List.of(FakeLibhegel.CLONE_BASE), workerFails.freedClones); + assertEquals(1, workerFails.freedStateMachines); + } + + /** A rule that sets its own thread's interrupt flag, so the worker exits between rounds. */ + static final class SelfInterrupting { + @Rule + void interrupt(TestCase t) { + Thread.currentThread().interrupt(); + } + } + + @Test + void aWorkerThatExitsWithoutReportingIsAnInternalError() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{0}, {}}, new long[][] {{0}, {}}); + HegelException e = assertThrows( + HegelException.class, () -> Stateful.run(new SelfInterrupting(), exploringCase(fake), twoWorkers())); + assertTrue(e.getMessage().contains("worker 0 exited without reporting"), e.getMessage()); + // Both rounds' clones were released before the round was judged. + assertEquals(4, fake.freedClones.size()); + assertEquals(1, fake.freedStateMachines); + } + + /** A rule that interrupts the driving thread while it waits at the join point. */ + static final class DriverInterrupting { + final Thread driver = Thread.currentThread(); + + @Rule + void interrupt(TestCase t) { + driver.interrupt(); + } + } + + @Test + void anInterruptedDriverStopsItsWorkersAndKeepsTheFlag() { + FakeLibhegel fake = concurrentFake(2, new long[][] {{0}, {}}); + HegelException e = assertThrows( + HegelException.class, () -> Stateful.run(new DriverInterrupting(), exploringCase(fake), twoWorkers())); + // The flag is restored for the caller; clear it so it does not leak into later tests. + assertTrue(Thread.interrupted(), "the interrupt flag was not restored"); + assertTrue(e.getMessage().contains("interrupted while waiting"), e.getMessage()); + // The round's handles were released once the workers had stopped. + assertEquals(2, fake.freedClones.size()); + assertEquals(1, fake.freedStateMachines); + } + + @Test + void optionsValidateAndExposeTheirValues() { + Stateful.Options options = + Stateful.options().stepCount(7).minConcurrency(2).maxConcurrency(5); + assertEquals(7, options.stepCount()); + assertEquals(2, options.minConcurrency()); + assertEquals(5, options.maxConcurrency()); + assertEquals(Stateful.DEFAULT_STEP_COUNT, Stateful.options().stepCount()); + assertEquals(1, Stateful.options().minConcurrency()); + assertEquals(1, Stateful.options().maxConcurrency()); + assertTrue(assertThrows( + IllegalArgumentException.class, () -> Stateful.options().stepCount(0)) + .getMessage() + .contains("stepCount")); + assertTrue(assertThrows( + IllegalArgumentException.class, () -> Stateful.options().minConcurrency(0)) + .getMessage() + .contains("minConcurrency")); + assertTrue(assertThrows( + IllegalArgumentException.class, () -> Stateful.options().maxConcurrency(0)) + .getMessage() + .contains("maxConcurrency")); + + FakeLibhegel fake = new FakeLibhegel(); + IllegalArgumentException e = assertThrows( + IllegalArgumentException.class, + () -> Stateful.run( + new Recording(), + exploringCase(fake), + Stateful.options().minConcurrency(3).maxConcurrency(2))); + assertTrue(e.getMessage().contains("must not exceed"), e.getMessage()); + // Rejected before anything reached the engine. + assertEquals(-1, fake.stateMachineStepCount); + } + + @Test + void maxConcurrencyOneRunsSequentiallyOnTheCallingThread() { + FakeLibhegel fake = new FakeLibhegel(); + fake.ruleSequence = new long[] {0}; + Recording machine = new Recording(); + TestCase tc = capturedCase(fake); + Stateful.run(machine, tc, Stateful.options().maxConcurrency(1).stepCount(9)); + assertEquals(Set.of(Thread.currentThread().getName()), machine.threads); + assertEquals(Set.of(-1), machine.workers); + assertTrue(fake.clonesMade.isEmpty()); + assertTrue(tc.notes().contains("Step 1: alpha"), tc.notes().toString()); + assertFalse(tc.notes().contains("Concurrency level: 1"), tc.notes().toString()); + assertEquals(9, fake.stateMachineStepCount); + // Groups are registered for sequential machines too; they have no effect at concurrency 1. + assertArrayEquals(new long[] {0, 1, 1}, fake.stateMachineRuleGroups); + } + + @Test + void concurrentPoolTracksValuesUnderItsLock() { + FakeLibhegel fake = new FakeLibhegel(); + TestCase tc = exploringCase(fake); + ConcurrentPool pool = new ConcurrentPool<>(tc); + assertTrue(pool.isEmpty()); + assertThrows(AssumeRejected.class, () -> tc.draw(pool.reusable())); + pool.add(tc, "a"); + assertFalse(pool.isEmpty()); + assertEquals(1, pool.size()); + assertEquals("a", tc.draw(pool.reusable())); + assertEquals(1, pool.size()); + assertEquals("a", tc.draw(pool.consuming())); + assertEquals(0, pool.size()); + assertThrows(AssumeRejected.class, () -> tc.draw(pool.consuming())); + } +} diff --git a/shared/src/test/java/dev/hegel/ConcurrentStatefulTest.java b/shared/src/test/java/dev/hegel/ConcurrentStatefulTest.java new file mode 100644 index 0000000..d285a1e --- /dev/null +++ b/shared/src/test/java/dev/hegel/ConcurrentStatefulTest.java @@ -0,0 +1,458 @@ +package dev.hegel; + +import static dev.hegel.Generators.integers; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; + +/** Concurrent stateful testing against the real engine. */ +class ConcurrentStatefulTest { + private static final Pattern WORKER_RULE_LINE = + Pattern.compile("\\[worker [0-9]+ \\+[0-9]+\\.[0-9]{3}ms\\] Rule: boom"); + private static final Pattern WORKER_DRAW_LINE = + Pattern.compile("\\[worker [0-9]+ \\+[0-9]+\\.[0-9]{3}ms\\] x(_[0-9]+)? = -?[0-9]+;"); + + private static Settings settings() { + return new Settings().database(Database.disabled()); + } + + private static Stateful.Options exactly(int workers) { + return Stateful.options().minConcurrency(workers).maxConcurrency(workers); + } + + private static boolean anyMatches(List lines, Pattern pattern) { + return lines.stream().anyMatch(line -> pattern.matcher(line).find()); + } + + /** A thread-safe counter whose decrement has a precondition, so rules are rejected at every level. */ + static final class ConcurrentCounter { + private final AtomicInteger n = new AtomicInteger(); + + @Rule + void increment(TestCase tc) { + n.incrementAndGet(); + } + + @Rule + void decrement(TestCase tc) { + tc.assume(n.get() > 0); + n.updateAndGet(v -> Math.max(0, v - 1)); + } + + @Invariant + void nonNegative(TestCase tc) { + assertTrue(n.get() >= 0); + } + } + + @HegelTest(database = Database.DISABLED) + void counterHoldsAtEveryConcurrencyLevel(TestCase tc) { + Stateful.run(new ConcurrentCounter(), tc, Stateful.options().maxConcurrency(3)); + } + + /** Named groups alongside the anonymous one, with a rule that rejects. */ + static final class Grouped { + private final AtomicInteger letters = new AtomicInteger(); + private final AtomicInteger numbers = new AtomicInteger(); + + @Rule(group = "letters") + void a(TestCase tc) { + letters.incrementAndGet(); + } + + @Rule(group = "letters", weight = 2) + void b(TestCase tc) { + letters.incrementAndGet(); + } + + @Rule(group = "numbers") + void one(TestCase tc) { + tc.assume(numbers.get() != 3); + numbers.incrementAndGet(); + } + + @Rule + void reset(TestCase tc) { + numbers.set(0); + } + + @Invariant + void counts(TestCase tc) { + assertTrue(letters.get() >= 0 && numbers.get() >= 0); + } + } + + @HegelTest(database = Database.DISABLED) + void groupedMachinePasses(TestCase tc) { + Stateful.run(new Grouped(), tc, Stateful.options().maxConcurrency(3)); + } + + /** Records which threads run its rules and whether two rules ever overlapped in time. */ + static final class Probe { + final Set threads = ConcurrentHashMap.newKeySet(); + private final AtomicInteger active = new AtomicInteger(); + volatile boolean overlapped; + volatile boolean invariantSawActiveRule; + + @Rule + void touch(TestCase tc) { + threads.add(Thread.currentThread().getId()); + if (active.incrementAndGet() > 1) { + overlapped = true; + } + LockSupport.parkNanos(20_000); + active.decrementAndGet(); + } + + @Invariant(alwaysRun = true) + void quiet(TestCase tc) { + if (active.get() != 0) { + invariantSawActiveRule = true; + } + } + } + + @Test + void severalWorkersRunRulesInParallelAndInvariantsRunBetweenRounds() { + long driver = Thread.currentThread().getId(); + Set workerCounts = ConcurrentHashMap.newKeySet(); + boolean[] overlapped = {false}; + boolean[] invariantOverlap = {false}; + Hegel.run( + tc -> { + Probe probe = new Probe(); + Stateful.run(probe, tc, exactly(3).stepCount(5)); + workerCounts.add(probe.threads.size()); + assertFalse(probe.threads.contains(driver), "a rule ran on the driving thread"); + overlapped[0] |= probe.overlapped; + invariantOverlap[0] |= probe.invariantSawActiveRule; + }, + settings().testCases(50), + Reporter.silent()) + .throwIfFailed(); + assertTrue(workerCounts.contains(3), "expected some case to use all three workers: " + workerCounts); + assertTrue(overlapped[0], "expected two rules to run at the same time in some case"); + assertFalse(invariantOverlap[0], "an invariant observed a rule in flight"); + } + + @Test + void defaultOptionsRunOnTheCallingThread() { + long driver = Thread.currentThread().getId(); + Hegel.run( + tc -> { + Probe probe = new Probe(); + Stateful.run(probe, tc); + assertEquals(Set.of(driver), probe.threads); + }, + settings().testCases(5), + Reporter.silent()) + .throwIfFailed(); + } + + /** Rules in three groups that fail if a rule of another group is in flight at the same time. */ + static final class Exclusive { + private final AtomicInteger[] active = {new AtomicInteger(), new AtomicInteger(), new AtomicInteger()}; + final Set groupsSeen = ConcurrentHashMap.newKeySet(); + volatile boolean overlapAcrossGroups; + + private void run(int group) { + groupsSeen.add(group); + active[group].incrementAndGet(); + for (int other = 0; other < active.length; other++) { + if (other != group && active[other].get() > 0) { + overlapAcrossGroups = true; + } + } + LockSupport.parkNanos(20_000); + active[group].decrementAndGet(); + } + + @Rule(group = "a") + void a1(TestCase tc) { + run(0); + } + + @Rule(group = "a") + void a2(TestCase tc) { + run(0); + } + + @Rule(group = "b") + void b(TestCase tc) { + run(1); + } + + @Rule + void anonymous(TestCase tc) { + run(2); + } + } + + @Test + void rulesOfDifferentGroupsNeverOverlap() { + Set groupsSeen = ConcurrentHashMap.newKeySet(); + boolean[] overlap = {false}; + Hegel.run( + tc -> { + Exclusive machine = new Exclusive(); + Stateful.run(machine, tc, exactly(4).stepCount(10)); + groupsSeen.addAll(machine.groupsSeen); + overlap[0] |= machine.overlapAcrossGroups; + }, + settings().testCases(50), + Reporter.silent()) + .throwIfFailed(); + assertFalse(overlap[0], "rules from two groups ran at the same time"); + assertEquals(Set.of(0, 1, 2), groupsSeen); + } + + /** A rule that always fails, after a labelled draw so the report has a worker-stamped draw line. */ + static final class Boom { + private final boolean armed; + + Boom(boolean armed) { + this.armed = armed; + } + + @Rule + void boom(TestCase tc) { + int x = tc.draw(integers().min(0).max(1000), "x"); + if (armed) { + throw new IllegalStateException("concurrent boom " + x); + } + } + } + + @Test + void aWorkerFailureIsReportedWithItsRoundAndWorkerLines() { + ByteArrayOutputStream buf = new ByteArrayOutputStream(); + RunReport report = Hegel.run( + tc -> Stateful.run(new Boom(true), tc, exactly(2)), + settings().seed(3), + Reporter.printing(new PrintStream(buf, true, StandardCharsets.UTF_8))); + assertEquals(RunStatus.FAILED, report.status()); + Failure failure = report.failures().get(0); + IllegalStateException e = assertInstanceOf(IllegalStateException.class, failure.exception()); + assertTrue(e.getMessage().startsWith("concurrent boom"), e.getMessage()); + List notes = failure.notes(); + assertTrue(notes.contains("Concurrency level: 2"), notes.toString()); + assertTrue( + notes.contains("---------------- Round 1: group \"" + Stateful.ANONYMOUS_GROUP + "\" ----------------"), + notes.toString()); + assertTrue(anyMatches(notes, WORKER_RULE_LINE), notes.toString()); + // Draws are recorded under their plain names; the stamp is only on the printed line. + for (String name : failure.draws().keySet()) { + assertTrue(name.matches("x(_[0-9]+)?"), name); + } + String printed = buf.toString(StandardCharsets.UTF_8); + assertTrue(anyMatches(List.of(printed.split("\\R")), WORKER_DRAW_LINE), printed); + // The failure is the body's own exception, rethrown as-is by test(). + assertThrows(IllegalStateException.class, report::throwIfFailed); + } + + @Test + void aConcurrentFailureBlobReplaysAndGoesStaleOnceFixed() { + RunReport report = Hegel.run( + tc -> Stateful.run(new Boom(true), tc, exactly(2)), settings().seed(3), Reporter.silent()); + assertEquals(RunStatus.FAILED, report.status()); + String blob = report.failures().get(0).reproduceBlob().orElseThrow(); + + RunReport replayed = Hegel.run( + tc -> Stateful.run(new Boom(true), tc, exactly(2)), + settings().reproduceFailure(blob), + Reporter.silent()); + assertEquals(RunStatus.FAILED, replayed.status()); + assertInstanceOf(IllegalStateException.class, replayed.failures().get(0).exception()); + + HegelException stale = assertThrows( + HegelException.class, + () -> Hegel.run( + tc -> Stateful.run(new Boom(false), tc, exactly(2)), + settings().reproduceFailure(blob), + Reporter.silent())); + assertEquals(Runner.STALE_BLOB, stale.getMessage()); + } + + /** A key-value store whose increment reads, yields, then writes: a lost-update race. */ + static final class RacyStore { + private final Map store = new ConcurrentHashMap<>(); + private final AtomicInteger increments = new AtomicInteger(); + private final ConcurrentPool keys; + + RacyStore(TestCase tc) { + keys = new ConcurrentPool<>(tc); + } + + @Rule(group = "ops") + void register(TestCase tc) { + int key = tc.draw(integers().min(0).max(3)); + if (store.putIfAbsent(key, 0) == null) { + keys.add(tc, key); + } + } + + @Rule(group = "ops", weight = 3) + void increment(TestCase tc) { + int key = tc.draw(keys.reusable()); + int value = store.get(key); + LockSupport.parkNanos(10_000); + store.put(key, value + 1); + increments.incrementAndGet(); + } + + @Rule(group = "audit") + void audit(TestCase tc) { + tc.note("store holds " + store.size() + " keys"); + } + + @Invariant(alwaysRun = true) + void noLostUpdates(TestCase tc) { + int total = store.values().stream().mapToInt(Integer::intValue).sum(); + assertEquals(increments.get(), total, "lost update"); + } + } + + @Test + void aLostUpdateRaceIsFound() { + RunReport report = Hegel.run( + tc -> Stateful.run( + new RacyStore(tc), + tc, + Stateful.options().maxConcurrency(4).stepCount(10)), + settings().testCases(100), + Reporter.silent()); + assertEquals(RunStatus.FAILED, report.status(), String.valueOf(report.error())); + Failure failure = report.failures().get(0); + AssertionError e = assertInstanceOf(AssertionError.class, failure.exception()); + assertTrue(e.getMessage().contains("lost update"), e.getMessage()); + } + + /** Rules that add and consume values through a pool shared by every worker. */ + static final class PoolMachine { + private final Set live = ConcurrentHashMap.newKeySet(); + private final AtomicInteger next = new AtomicInteger(); + private final ConcurrentPool pool; + + PoolMachine(TestCase tc) { + pool = new ConcurrentPool<>(tc); + } + + @Rule + void create(TestCase tc) { + int v = next.getAndIncrement(); + live.add(v); + pool.add(tc, v); + } + + @Rule + void reuse(TestCase tc) { + int v = tc.draw(pool.reusable()); + // Another worker may consume v between the draw and this check, so only ask whether + // it was ever created. + assertTrue(v >= 0 && v < next.get(), "reused a value never created: " + v); + } + + @Rule + void consume(TestCase tc) { + int v = tc.draw(pool.consuming()); + assertTrue(live.remove(v), "consumed a value twice or never created: " + v); + } + + @Invariant(alwaysRun = true) + void nothingLost(TestCase tc) { + assertEquals(live.size(), pool.size()); + } + } + + @HegelTest(database = Database.DISABLED) + void concurrentPoolsHandOutEachValueToOneConsumer(TestCase tc) { + Stateful.run(new PoolMachine(tc), tc, Stateful.options().maxConcurrency(4)); + } + + /** A rule that exhausts the engine's choice budget once, in the run's first case. */ + static final class Greedy { + private final AtomicBoolean drained; + + Greedy(AtomicBoolean drained) { + this.drained = drained; + } + + @Rule + void drain(TestCase tc) { + if (drained.getAndSet(true)) { + return; + } + for (int i = 0; i < 2_000_000; i++) { + tc.draw(integers()); + } + } + } + + @Test + void anOverrunningWorkerConcludesTheCaseAsAnOverrun() { + // Draining the budget costs about a million engine calls, so only the first case does it; + // the engine then counts that case as an overrun and moves on to a valid one. + AtomicBoolean drained = new AtomicBoolean(); + RunReport report = Hegel.run( + tc -> Stateful.run(new Greedy(drained), tc, exactly(2)), + settings().testCases(1), + Reporter.silent()); + assertEquals(RunStatus.PASSED, report.status(), String.valueOf(report.error())); + assertEquals(1, report.statistics().overrun()); + assertEquals(1, report.statistics().valid()); + } + + /** An invariant whose assumption fails at a join point invalidates the case. */ + static final class Picky { + @Rule + void step(TestCase tc) {} + + @Invariant(alwaysRun = true) + void never(TestCase tc) { + tc.assume(false); + } + } + + @Test + void anInvariantAssumptionFailureInvalidatesTheCase() { + RunReport report = Hegel.run( + tc -> Stateful.run(new Picky(), tc, exactly(2)), settings().testCases(5), Reporter.silent()); + assertEquals(RunStatus.ERROR, report.status()); + assertTrue(report.healthCheckFailed(), String.valueOf(report.error())); + } + + static final class NoRules { + @Invariant + void lonely(TestCase tc) {} + } + + @HegelTest(database = Database.DISABLED, testCases = 1) + void invalidOptionsAndMachinesAreRejectedBeforeReachingTheEngine(TestCase tc) { + IllegalArgumentException bounds = assertThrows( + IllegalArgumentException.class, + () -> Stateful.run( + new ConcurrentCounter(), + tc, + Stateful.options().minConcurrency(2).maxConcurrency(1))); + assertTrue(bounds.getMessage().contains("minConcurrency"), bounds.getMessage()); + assertThrows(IllegalArgumentException.class, () -> Stateful.options().minConcurrency(0)); + assertThrows(IllegalArgumentException.class, () -> Stateful.options().maxConcurrency(0)); + assertThrows(IllegalArgumentException.class, () -> Stateful.options().stepCount(0)); + assertThrows(IllegalArgumentException.class, () -> Stateful.run(new NoRules(), tc, exactly(2))); + } +} diff --git a/shared/src/test/java/dev/hegel/FakeLibhegel.java b/shared/src/test/java/dev/hegel/FakeLibhegel.java index 1a78aeb..030b0f3 100644 --- a/shared/src/test/java/dev/hegel/FakeLibhegel.java +++ b/shared/src/test/java/dev/hegel/FakeLibhegel.java @@ -6,7 +6,9 @@ import java.time.LocalDateTime; import java.time.LocalTime; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.function.Consumer; /** @@ -22,6 +24,18 @@ final class FakeLibhegel implements Libhegel { static final long TC = 0x300; static final long RESULT = 0x400; static final long STRING_GEN = 0x600; + /** Clone handles are {@code CLONE_BASE + n} for the n-th clone made, so they are distinct from {@link #TC}. */ + static final long CLONE_BASE = 0x1000; + + // Test-case clones (concurrent stateful workers). + int cloneRc = Abi.OK; + boolean setWorkerFails; + /** The source handle of each clone made, in order. */ + final List clonesMade = new ArrayList<>(); + /** Clone handles released through testCaseFree, in order. */ + final List freedClones = new ArrayList<>(); + /** The worker index each handle was attributed to. */ + final Map workersSet = new LinkedHashMap<>(); String lastError = "fake error"; String version = "0.0.0-fake"; @@ -127,9 +141,13 @@ final class FakeLibhegel implements Libhegel { private long nextVariableId; int poolGenerateRc = Abi.OK; Long poolGenerateValue; // null = the first added variable id (0) - // State machines. The fake plays the engine's sequential protocol: each round (next_group) - // hands out exactly one rule from `ruleSequence` (next_rule), then the join point; once the - // sequence is exhausted next_group reports DONE. + // State machines. The fake plays the engine's round protocol. Sequentially, each round + // (next_group) hands out exactly one rule from `ruleSequence` (next_rule), then the join + // point; once the sequence is exhausted next_group reports DONE. Concurrently, set + // `concurrentRounds`: round r hands worker w the rules `concurrentRounds[r][w]` (a missing or + // empty worker entry means that worker's round is over at once), and `stateMachineConcurrency` + // says how many workers new_state_machine reports. The protocol methods are synchronized + // because concurrent workers call them at the same time. int newStateMachineRc = Abi.OK; long stateMachineId = 5; List stateMachineRules; @@ -146,9 +164,18 @@ final class FakeLibhegel implements Libhegel { int stateMachineNextRuleRc = Abi.OK; long[] ruleSequence = {}; // the rule index handed out each round private int ruleIndex; - private boolean roundOpen; + long[][][] concurrentRounds; // null = the sequential protocol over ruleSequence + private int roundsServed; + private long[][] currentRound; // null = no round open + private int[] positions; // per worker, how far into its queue for the current round + /** The worker index of every next_rule call, in call order. */ + final List nextRuleWorkers = new ArrayList<>(); + int stateMachineRuleRejectedRc = Abi.OK; int rejectedRules; + /** The worker index of every rule_rejected call, in call order. */ + final List rejectedWorkers = new ArrayList<>(); + int stateMachineShouldCheckInvariantRc = Abi.OK; boolean shouldCheckInvariant = true; // the sampling decision for every invariant final List invariantChecksAsked = new ArrayList<>(); @@ -308,6 +335,26 @@ public int testCaseFromBlob(long settings, String blob, Consumer output, @Override public void testCaseFree(long tc) { freedTestCases++; + if (tc != TC) { + freedClones.add(tc); + } + } + + @Override + public int testCaseClone(long tc, long[] out) { + if (cloneRc == Abi.OK) { + out[0] = CLONE_BASE + clonesMade.size(); + clonesMade.add(tc); + } + return cloneRc; + } + + @Override + public void testCaseSetWorker(long tc, long workerIndex) { + if (setWorkerFails) { + throw new HegelException("hegel_test_case_set_worker failed: " + lastError); + } + workersSet.put(tc, workerIndex); } @Override @@ -566,27 +613,40 @@ public int newStateMachine( } @Override - public int stateMachineNextGroup(long tc, long stateMachineId, long[] outGroupId) { + public synchronized int stateMachineNextGroup(long tc, long stateMachineId, long[] outGroupId) { if (stateMachineNextGroupRc == Abi.OK) { - roundOpen = ruleIndex < ruleSequence.length; - outGroupId[0] = roundOpen ? stateMachineGroupId : Abi.STATE_MACHINE_DONE; + if (concurrentRounds != null) { + currentRound = roundsServed < concurrentRounds.length ? concurrentRounds[roundsServed++] : null; + } else { + // Sequential: one rule for worker 0 per round. + currentRound = ruleIndex < ruleSequence.length ? new long[][] {{ruleSequence[ruleIndex++]}} : null; + } + positions = new int[(int) stateMachineConcurrency]; + outGroupId[0] = currentRound != null ? stateMachineGroupId : Abi.STATE_MACHINE_DONE; } return stateMachineNextGroupRc; } @Override - public int stateMachineNextRule(long tc, long stateMachineId, long workerIndex, long[] outRuleIndex) { + public synchronized int stateMachineNextRule(long tc, long stateMachineId, long workerIndex, long[] outRuleIndex) { if (stateMachineNextRuleRc == Abi.OK) { - outRuleIndex[0] = roundOpen ? ruleSequence[ruleIndex++] : Abi.STATE_MACHINE_DONE; - roundOpen = false; + nextRuleWorkers.add(workerIndex); + int w = (int) workerIndex; + long[] queue = currentRound != null && w < currentRound.length ? currentRound[w] : new long[0]; + if (positions[w] < queue.length) { + outRuleIndex[0] = queue[positions[w]++]; + } else { + outRuleIndex[0] = Abi.STATE_MACHINE_DONE; + } } return stateMachineNextRuleRc; } @Override - public int stateMachineRuleRejected(long tc, long stateMachineId, long workerIndex) { + public synchronized int stateMachineRuleRejected(long tc, long stateMachineId, long workerIndex) { if (stateMachineRuleRejectedRc == Abi.OK) { rejectedRules++; + rejectedWorkers.add(workerIndex); } return stateMachineRuleRejectedRc; } diff --git a/shared/src/test/java/dev/hegel/StatefulTest.java b/shared/src/test/java/dev/hegel/StatefulTest.java index bc4fa45..994b76a 100644 --- a/shared/src/test/java/dev/hegel/StatefulTest.java +++ b/shared/src/test/java/dev/hegel/StatefulTest.java @@ -161,7 +161,15 @@ void engineRejectsInvalidWeightsItIsHanded() { () -> Runner.run( Engine.get(), new Settings().database(Database.disabled()).testCases(1), - tc -> tc.newStateMachine(List.of("r"), new double[] {0}, List.of(), new boolean[0], 50), + tc -> tc.newStateMachine( + List.of("r"), + new long[] {0}, + new double[] {0}, + List.of(), + new boolean[0], + 1, + 1, + 50), Reporter.silent()) .throwIfFailed()); assertTrue(e.getMessage().toLowerCase().contains("weight"), e.getMessage());