From 353dd08e3b3e36abdf6aa022128bbee177c11786 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 13:02:44 +0000 Subject: [PATCH 1/2] Add jqwik property tests and jcstress concurrency tests Introduces two complementary test frameworks alongside the existing JUnit 5 suite to cover gaps that example-based tests cannot reach: - jqwik (property-based, JUnit 5 engine) verifies round-trip, chunk-size independence, and counter accounting over arbitrary inputs. - jcstress (Mode.Termination) verifies that a reader blocked in read() is correctly unblocked by both close() and write() signals, exercising the signalModification semaphore and streamClosed volatile. jcstress is wired into the default test phase via exec-maven-plugin (exec goal, forked JVM, -m quick) so mvn test runs it always-on. --- .gitignore | 5 + CLAUDE.md | 4 + pom.xml | 33 ++++++ .../streambuffer/StreamBufferProperties.java | 100 ++++++++++++++++++ .../jcstress/CloseDuringReadRace.java | 42 ++++++++ .../jcstress/WriteUnblocksReadRace.java | 44 ++++++++ 6 files changed, 228 insertions(+) create mode 100644 src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java create mode 100644 src/test/java/net/ladenthin/streambuffer/jcstress/CloseDuringReadRace.java create mode 100644 src/test/java/net/ladenthin/streambuffer/jcstress/WriteUnblocksReadRace.java diff --git a/.gitignore b/.gitignore index 8536208..8915662 100644 --- a/.gitignore +++ b/.gitignore @@ -29,6 +29,11 @@ hs_err_pid* /target/ /bin/ +# jcstress / jqwik test outputs (generated in repo root) +/results/ +/jcstress-results-*.bin.gz +/.jqwik-database + # IDEA files /.idea/ *.iml diff --git a/CLAUDE.md b/CLAUDE.md index 7ff9ac6..bf9af4b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -21,6 +21,10 @@ mvn test -Dtest=StreamBufferTest#testSimpleRoundTrip mvn org.pitest:pitest-maven:mutationCoverage ``` +`mvn test` also runs: +- **jqwik properties** (`StreamBufferProperties`) — picked up by Surefire as a JUnit 5 engine. +- **jcstress** tests under `net.ladenthin.streambuffer.jcstress` — executed in a forked JVM via `exec-maven-plugin` in the `test` phase (`-m quick` mode). + ## Architecture `StreamBuffer` is a single-class Java library (`net.ladenthin.streambuffer`) that connects an `OutputStream` and `InputStream` through a dynamic FIFO queue — solving the fixed-buffer and cross-thread-deadlock limitations of Java's `PipedInputStream`/`PipedOutputStream`. diff --git a/pom.xml b/pom.xml index c316dd5..821d579 100644 --- a/pom.xml +++ b/pom.xml @@ -21,6 +21,8 @@ SPDX-License-Identifier: Apache-2.0 UTF-8 1.8 21 + 0.16 + 1.9.2 ${git.commit.time} 2014 @@ -95,6 +97,18 @@ SPDX-License-Identifier: Apache-2.0 1.37 test + + net.jqwik + jqwik + ${jqwik.version} + test + + + org.openjdk.jcstress + jcstress-core + ${jcstress.version} + test + @@ -224,6 +238,25 @@ SPDX-License-Identifier: Apache-2.0 org.openjdk.jmh.Main test + + + jcstress + test + exec + + ${java.home}/bin/java + test + + -classpath + + org.openjdk.jcstress.Main + -v + -m + quick + + + + org.pitest diff --git a/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java b/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java new file mode 100644 index 0000000..ac0016d --- /dev/null +++ b/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java @@ -0,0 +1,100 @@ +// SPDX-FileCopyrightText: 2014-2026 Bernard Ladenthin +// +// SPDX-License-Identifier: Apache-2.0 +package net.ladenthin.streambuffer; + +import net.jqwik.api.ForAll; +import net.jqwik.api.Property; +import net.jqwik.api.constraints.IntRange; +import net.jqwik.api.constraints.Size; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.List; + +public class StreamBufferProperties { + + @Property + boolean writeAllThenReadAvailableYieldsConcatenation( + @ForAll @Size(min = 0, max = 32) List<@Size(min = 0, max = 256) byte[]> chunks + ) throws IOException { + StreamBuffer sb = new StreamBuffer(); + OutputStream os = sb.getOutputStream(); + InputStream is = sb.getInputStream(); + + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + for (byte[] chunk : chunks) { + os.write(chunk); + expected.write(chunk); + } + + int available = is.available(); + if (available != expected.size()) return false; + if (available == 0) return true; + + byte[] read = new byte[available]; + int n = is.read(read, 0, available); + if (n != available) return false; + + return java.util.Arrays.equals(read, expected.toByteArray()); + } + + @Property + boolean readChunkSizeDoesNotAffectContent( + @ForAll @Size(min = 1, max = 1024) byte[] payload, + @ForAll @IntRange(min = 1, max = 64) int writeChunk, + @ForAll @IntRange(min = 1, max = 64) int readChunk + ) throws IOException { + StreamBuffer sb = new StreamBuffer(); + OutputStream os = sb.getOutputStream(); + InputStream is = sb.getInputStream(); + + for (int off = 0; off < payload.length; off += writeChunk) { + int len = Math.min(writeChunk, payload.length - off); + os.write(payload, off, len); + } + + byte[] out = new byte[payload.length]; + int read = 0; + while (read < payload.length) { + int len = Math.min(readChunk, payload.length - read); + int n = is.read(out, read, len); + if (n <= 0) return false; + read += n; + } + + return java.util.Arrays.equals(payload, out); + } + + @Property + boolean counterAccountingIsConsistent( + @ForAll @Size(min = 0, max = 16) List<@Size(min = 0, max = 128) byte[]> writes, + @ForAll @IntRange(min = 0, max = 16) int readChunk + ) throws IOException { + StreamBuffer sb = new StreamBuffer(); + OutputStream os = sb.getOutputStream(); + InputStream is = sb.getInputStream(); + + long expectedWritten = 0; + for (byte[] w : writes) { + os.write(w); + expectedWritten += w.length; + } + + long expectedRead = 0; + if (readChunk > 0) { + byte[] buf = new byte[readChunk]; + while (is.available() > 0) { + int n = is.read(buf, 0, Math.min(readChunk, is.available())); + if (n <= 0) break; + expectedRead += n; + } + } + + if (sb.getTotalBytesWritten() != expectedWritten) return false; + if (sb.getTotalBytesRead() != expectedRead) return false; + return sb.getTotalBytesWritten() == sb.getTotalBytesRead() + is.available(); + } +} diff --git a/src/test/java/net/ladenthin/streambuffer/jcstress/CloseDuringReadRace.java b/src/test/java/net/ladenthin/streambuffer/jcstress/CloseDuringReadRace.java new file mode 100644 index 0000000..94d755f --- /dev/null +++ b/src/test/java/net/ladenthin/streambuffer/jcstress/CloseDuringReadRace.java @@ -0,0 +1,42 @@ +// SPDX-FileCopyrightText: 2014-2026 Bernard Ladenthin +// +// SPDX-License-Identifier: Apache-2.0 +package net.ladenthin.streambuffer.jcstress; + +import net.ladenthin.streambuffer.StreamBuffer; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Description; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Mode; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.Signal; +import org.openjdk.jcstress.annotations.State; + +import java.io.IOException; +import java.io.InputStream; + +@JCStressTest(Mode.Termination) +@Description("A reader blocked in read() must be unblocked when close() is invoked.") +@Outcome(id = "TERMINATED", expect = Expect.ACCEPTABLE, desc = "close() unblocked the reader") +@Outcome(id = "STALE", expect = Expect.FORBIDDEN, desc = "Reader stuck after close()") +@State +public class CloseDuringReadRace { + + private final StreamBuffer sb = new StreamBuffer(); + private final InputStream is = sb.getInputStream(); + + @Actor + public void reader() { + try { + is.read(); + } catch (IOException ignored) { + // acceptable: close races may surface as IOException in some paths + } + } + + @Signal + public void closer() throws IOException { + sb.close(); + } +} diff --git a/src/test/java/net/ladenthin/streambuffer/jcstress/WriteUnblocksReadRace.java b/src/test/java/net/ladenthin/streambuffer/jcstress/WriteUnblocksReadRace.java new file mode 100644 index 0000000..0dcc3cc --- /dev/null +++ b/src/test/java/net/ladenthin/streambuffer/jcstress/WriteUnblocksReadRace.java @@ -0,0 +1,44 @@ +// SPDX-FileCopyrightText: 2014-2026 Bernard Ladenthin +// +// SPDX-License-Identifier: Apache-2.0 +package net.ladenthin.streambuffer.jcstress; + +import net.ladenthin.streambuffer.StreamBuffer; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Description; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Mode; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.Signal; +import org.openjdk.jcstress.annotations.State; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +@JCStressTest(Mode.Termination) +@Description("A reader blocked in read() must be unblocked when a writer publishes a byte.") +@Outcome(id = "TERMINATED", expect = Expect.ACCEPTABLE, desc = "write() unblocked the reader") +@Outcome(id = "STALE", expect = Expect.FORBIDDEN, desc = "Reader stuck after write()") +@State +public class WriteUnblocksReadRace { + + private final StreamBuffer sb = new StreamBuffer(); + private final InputStream is = sb.getInputStream(); + private final OutputStream os = sb.getOutputStream(); + + @Actor + public void reader() { + try { + is.read(); + } catch (IOException ignored) { + // not expected on this path, but tolerated + } + } + + @Signal + public void writer() throws IOException { + os.write(0x42); + } +} From 82c0f56b3f7e32b1ac88584078c8549b9ff5b98d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 13:58:51 +0000 Subject: [PATCH 2/2] Grow concurrency seed: jcstress default mode + concurrent-write race + 2 properties - Bump jcstress mode from -m quick to -m default (5 forks vs 1) for higher per-test sample counts; build time goes from ~55s to ~3min. - Add ConcurrentWriteRace: two writers race; arbiter asserts both payloads appear contiguously (no interleaving) in one of two valid orders. - Add safeWriteIsolatesFromExternalMutation property: with safeWrite=true, mutating the source array after write() must not affect read output. - Add trimPreservesContentAcrossMaxBufferElements property: round-trip content must be preserved for arbitrary maxBufferElements (covers trim correctness). --- pom.xml | 2 +- .../streambuffer/StreamBufferProperties.java | 46 +++++++++++++++ .../jcstress/ConcurrentWriteRace.java | 58 +++++++++++++++++++ 3 files changed, 105 insertions(+), 1 deletion(-) create mode 100644 src/test/java/net/ladenthin/streambuffer/jcstress/ConcurrentWriteRace.java diff --git a/pom.xml b/pom.xml index 821d579..1994f38 100644 --- a/pom.xml +++ b/pom.xml @@ -252,7 +252,7 @@ SPDX-License-Identifier: Apache-2.0 org.openjdk.jcstress.Main -v -m - quick + default diff --git a/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java b/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java index ac0016d..e36aeba 100644 --- a/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java +++ b/src/test/java/net/ladenthin/streambuffer/StreamBufferProperties.java @@ -97,4 +97,50 @@ boolean counterAccountingIsConsistent( if (sb.getTotalBytesRead() != expectedRead) return false; return sb.getTotalBytesWritten() == sb.getTotalBytesRead() + is.available(); } + + @Property + boolean safeWriteIsolatesFromExternalMutation( + @ForAll @Size(min = 1, max = 256) byte[] payload + ) throws IOException { + StreamBuffer sb = new StreamBuffer(); + sb.setSafeWrite(true); + OutputStream os = sb.getOutputStream(); + InputStream is = sb.getInputStream(); + + byte[] expected = payload.clone(); + os.write(payload); + // Mutate the source after writing; safeWrite=true must have cloned on write. + java.util.Arrays.fill(payload, (byte) 0x7F); + + byte[] read = new byte[expected.length]; + int n = is.read(read, 0, expected.length); + if (n != expected.length) return false; + return java.util.Arrays.equals(read, expected); + } + + @Property + boolean trimPreservesContentAcrossMaxBufferElements( + @ForAll @Size(min = 1, max = 64) List<@Size(min = 0, max = 64) byte[]> chunks, + @ForAll @IntRange(min = 0, max = 16) int maxBufferElements + ) throws IOException { + StreamBuffer sb = new StreamBuffer(); + sb.setMaxBufferElements(maxBufferElements); + OutputStream os = sb.getOutputStream(); + InputStream is = sb.getInputStream(); + + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + for (byte[] chunk : chunks) { + os.write(chunk); + expected.write(chunk); + } + + int available = is.available(); + if (available != expected.size()) return false; + if (available == 0) return true; + + byte[] read = new byte[available]; + int n = is.read(read, 0, available); + if (n != available) return false; + return java.util.Arrays.equals(read, expected.toByteArray()); + } } diff --git a/src/test/java/net/ladenthin/streambuffer/jcstress/ConcurrentWriteRace.java b/src/test/java/net/ladenthin/streambuffer/jcstress/ConcurrentWriteRace.java new file mode 100644 index 0000000..b801f4d --- /dev/null +++ b/src/test/java/net/ladenthin/streambuffer/jcstress/ConcurrentWriteRace.java @@ -0,0 +1,58 @@ +// SPDX-FileCopyrightText: 2014-2026 Bernard Ladenthin +// +// SPDX-License-Identifier: Apache-2.0 +package net.ladenthin.streambuffer.jcstress; + +import net.ladenthin.streambuffer.StreamBuffer; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Description; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.Z_Result; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.Arrays; + +@JCStressTest +@Description("Two concurrent writers must each appear contiguously in the FIFO; bytes must not interleave.") +@Outcome(id = "true", expect = Expect.ACCEPTABLE, desc = "Both payloads intact in some order") +@Outcome(id = "false", expect = Expect.FORBIDDEN, desc = "Torn / interleaved write") +@State +public class ConcurrentWriteRace { + + private static final byte[] A = new byte[]{1, 2}; + private static final byte[] B = new byte[]{3, 4}; + + private final StreamBuffer sb = new StreamBuffer(); + private final OutputStream os = sb.getOutputStream(); + private final InputStream is = sb.getInputStream(); + + @Actor + public void writerA() { + try { os.write(A); } catch (IOException ignored) { } + } + + @Actor + public void writerB() { + try { os.write(B); } catch (IOException ignored) { } + } + + @Arbiter + public void check(Z_Result r) { + try { + int available = is.available(); + byte[] all = new byte[available]; + int n = (available == 0) ? 0 : is.read(all, 0, available); + byte[] got = Arrays.copyOf(all, Math.max(n, 0)); + r.r1 = Arrays.equals(got, new byte[]{1, 2, 3, 4}) + || Arrays.equals(got, new byte[]{3, 4, 1, 2}); + } catch (IOException e) { + r.r1 = false; + } + } +}