Note

No IDE or local setup required. This repository is optimized for fully AI-assisted development using Claude Code. No local toolchain, no IDE, nothing to install — everything works completely through Claude.

AI:

Build:

Coverage:

Quality:

Security:

Package:

License:

Community:

streambuffer

StreamBuffer is a single-class Java library that bridges an OutputStream and an InputStream through a dynamic, unbounded FIFO queue — solving the fixed-buffer and potential deadlock limitations of Java's built-in PipedInputStream/PipedOutputStream. Several threads may write and several may read at the same time; with fixed-size records it also works as a work queue that spreads the records over several readers (see Fixed-Size Records with Several Readers).

• GitHub: https://github.com/bernardladenthin/streambuffer

Getting Started

Pull it from the central Maven repositories:

<dependency>
  <groupId>net.ladenthin</groupId>
  <artifactId>streambuffer</artifactId>
  <version>1.2.0</version>
</dependency>

Basic Usage

StreamBuffer sb = new StreamBuffer();
OutputStream os = sb.getOutputStream();
InputStream  is = sb.getInputStream();

os.write(new byte[]{1, 2, 3});

byte[] buf = new byte[3];
is.read(buf); // reads [1, 2, 3]

sb.close();

Closing via sb.close(), os.close(), or is.close() all close the entire buffer.

Motivation

Since JDK 1.0, PipedInputStream/PipedOutputStream has been available for connecting streams across threads, but it has notable limitations:

• Write operations block when the fixed circular buffer is full, potentially deadlocking the writing thread.
• The circular buffer has a fixed, 32-bit maximum size and does not grow or shrink dynamically.
• Every written byte array is copied into the circular buffer, doubling memory usage in the worst case.
• The buffer does not shrink after reads — a large circular buffer stays large.
• A pipe is considered broken if the thread that provided data is no longer alive, even if data remains in the buffer.

It is also not recommended to use both ends of a pipe from a single thread, as this may deadlock.

Solution

StreamBuffer holds references to written byte arrays directly in a Deque<byte[]> (FIFO), without copying data into a secondary circular buffer. This avoids the fixed-size constraint and eliminates write-blocking caused by buffer saturation.

Similar Solutions

• Gradle: StreamByteBuffer

Features

Dynamic FIFO Buffer

Instead of a fixed circular buffer, StreamBuffer uses a Deque<byte[]> that grows as data is written and shrinks as data is read. There is no upper bound on buffered data size at the FIFO level.

Full InputStream/OutputStream Compatibility

StreamBuffer exposes a standard InputStream (via getInputStream()) and a standard OutputStream (via getOutputStream()). Both are full subclass implementations and can be used anywhere those types are accepted. One deliberate difference: read(byte[], off, len) waits for all len bytes (or close()) and then takes them in one step, see Read.

Thread Safety

The InputStream and OutputStream can be used concurrently from different threads without additional synchronization, by several writers and several readers at the same time:

• Each write is atomic: its bytes enter the buffer together and in order.
• Each read(b, off, len) is atomic too: it takes its len bytes in one step once all of them are buffered, so they are contiguous in the stream (see Fixed-Size Records with Several Readers).
• Each buffered byte is delivered to exactly one reader, in the order it was written. No reader blocks while bytes are buffered, and after close() every waiting reader returns. read() throws no unchecked exception when another reader took the bytes first (bug-hunt finding F278; before 2.0.0 the buffer was correct for one reader only).
• There is no fairness between readers: which reader gets the next bytes is up to the scheduler. On a stream that is closed after finitely many bytes, every call still returns, even with threads that call forever and without fair locks, as long as the scheduler eventually runs every thread that can go on (weak fairness per thread, writer and closer): model-checked in bounded models, and proved with TLAPS for the counter abstraction of the calls and any number of threads. Separately, while a writer writes forever and nobody closes, a reader can lose every race for the bytes (model-checked). See src/test/tla/.
• All Deque accesses are guarded by a bufferLock object.
• State fields (streamClosed, safeWrite, availableBytes, positionAtCurrentBufferEntry, maxBufferElements, maxAllocationSize, isTrimRunning, maxObservedBytes) are volatile; the cumulative counters totalBytesWritten and totalBytesRead are final AtomicLong (atomic read-modify-write under concurrent I/O).
• close() takes the same lock, so it waits for a write or trim in progress on another thread: a write racing the close is either fully buffered or fails with "Stream closed", never appended after a reader saw the end of the stream. A trim copies every buffered byte, so with a large buffer that wait can be noticeable.
• A waiting reader blocks on a Condition of a ReentrantLock (waitLock, separate from bufferLock), which every write and the close signal to all waiters, avoiding busy-waiting. While no reader waits, a write skips that lock (a count of waiting readers), and a read with enough buffered bytes does not take it either. A java.util.concurrent lock does not pin a virtual thread's carrier while it waits. External semaphores can be registered via addSignal for thread-decoupled notification.
• A read interrupted while it waits throws java.io.InterruptedIOException (cause: the InterruptedException, interrupt flag set again). It has taken no byte (bytesTransferred is 0), since a read takes its bytes only once its wait is over. A thread that is already interrupted and would have to wait throws at once; with enough buffered bytes, or on a closed stream, it reads without an exception.

No Write Deadlock

In contrast to PipedOutputStream, write operations on StreamBuffer never block or deadlock — the FIFO grows as needed.

Fixed-Size Records with Several Readers (Work Queue)

Because read(b, off, len) takes its len bytes in one step, several readers that all read the same record size each get whole records. Every record goes to exactly one reader — the competing consumers pattern of a work queue, the way a load balancer spreads requests over workers. It is not a broadcast (publish/subscribe, fan-out): a record that one reader took is gone for the others.

static final int RECORD = 1429;

// any number of readers, each in its own thread
byte[] record = new byte[RECORD];
int n;
while ((n = is.read(record, 0, RECORD)) != -1) {
    // n == RECORD: a whole record; n < RECORD only for the last read after close()
    process(record, n);
}

What it relies on:

• The stream is a sequence of whole records. With one writer, the writes may have any size, like TCP segments: a record may be split across writes, and a write may hold the end of one record and the start of the next. With several writers, each write must hold whole records (one or several), because the writes of different writers interleave.
• Every reader reads the record size (or the same multiple of it); a reader with another size would shift the record boundaries for everyone.
• No order between readers. Each reader gets the records of each writer in the order they were written, but which reader gets which record is up to the scheduler. If the consumers must restore one order, number the records.
• Variable-length records (a length prefix followed by the payload) need two reads, and another reader can come in between. Use one reader for them, or pad them to a fixed size.

Like TCP, StreamBuffer is a byte stream and keeps no message boundaries of its own; the fixed record size provides them. For messages of arbitrary size, a BlockingQueue<byte[]> is the more direct tool. StreamBufferConcurrentReadersTest checks the record cases with 1429-byte records carrying writer id, sequence number and a CRC32: one writer with writes of random size and four readers, and three writers of whole records with four readers.

Smart Partial Reads

StreamBuffer tracks a positionAtCurrentBufferEntry index within the head byte array. Partial reads consume bytes from the current head entry without copying the remaining data. The trim operation accounts for this index and only copies the unread remainder.

Safe Write (Immutable Byte Arrays)

By default, StreamBuffer stores a direct reference to the written byte array (safeWrite = false). If the caller modifies the array after writing, the buffered content may be affected.

Enable safe write mode to clone every written byte array before buffering:

sb.setSafeWrite(true);

When writing with an offset (e.g., write(b, off, len)), a new byte array of exactly len bytes is always created, regardless of the safeWrite setting, since only the relevant portion is stored.

Buffer Trimming

When the Deque grows beyond maxBufferElements entries (default: 100), the next write operation consolidates the buffered data. This bounds the number of FIFO elements and can improve read performance for large accumulated buffers.

A trim merges geometrically: walking from the newest entry back, it takes every array the caller wrote since the last trim, and an older chunk of its own only while that chunk is not longer than what it has collected so far. Each byte is therefore copied a logarithmic number of times rather than once per trim, and the FIFO holds a few chunks that grow towards the head instead of one array. Would that still leave more than half of maxBufferElements entries, the trim merges everything, as before. A writer that stays ahead of its readers no longer pays for its backlog again on every trim: with 1429-byte records, the contended benchmark went from about 40 ms to about 4 ms per 11 MiB at the default threshold (details in TODO.md, A8 and A14).

sb.setMaxBufferElements(50);  // trim when more than 50 elements are queued
sb.setMaxBufferElements(0);   // disable trimming entirely

Trimming is triggered by writes, not by setMaxBufferElements. The trim copies the buffered arrays directly into the consolidated ones, without going through the streams, so safeWrite does not apply (the arrays it produces are not reachable from outside the buffer) and the buffered byte count stays constant while it runs: a concurrent waitForAtLeast or available() never sees a partly drained buffer. Every array the caller may still hold is copied by every trim, so after a trim no buffered array is reachable from outside: a caller writing without safeWrite may overwrite its arrays once a trim has run (a trim end signal tells when), and only the arrays written after that trim are still buffered by reference.

If maxAllocationSize is set (see below), trim may produce multiple smaller chunks instead of one. Trim is also skipped automatically when consolidation would not reduce the chunk count below maxBufferElements — for example, when maxAllocationSize is small enough that the resulting chunk count would still exceed the threshold. This prevents repeated no-op trims on every write.

Use isTrimRunning() to observe whether a trim is currently executing. This value is volatile and can change at any time in concurrent scenarios.

Large Buffer Support

available() returns Integer.MAX_VALUE when the number of buffered bytes exceeds Integer.MAX_VALUE, correctly handling buffers larger than 2 GB.

Signal/Slot Notifications

Register external Semaphore objects to receive thread-decoupled notifications when the buffer is modified (data written or stream closed). Each registered semaphore is released using the "max 1 permit" pattern (a permit only while it has none), enabling observers to block in their own threads and wake up when something changes:

Semaphore mySignal = new Semaphore(0);
sb.addSignal(mySignal);

// Observer's own thread:
while (!done) {
    mySignal.acquire();  // blocks until writer signals
    // process in MY thread — fully decoupled from writer
    if (sb.isClosed()) {
        done = true;
    }
}

API:

Method	Description
addSignal(Semaphore)	Registers an external semaphore; throws NullPointerException if null
removeSignal(Semaphore)	Removes a semaphore; returns false if not found or null

For trim lifecycle events specifically, see Trim Observer Signals below.

Signals are stored in a CopyOnWriteArrayList for thread-safe iteration. The "max 1 permit" pattern means rapid writes collapse into a single wake-up — the observer should check buffer state (e.g., isClosed(), available()) after waking to determine what changed. A semaphore holds at most one permit also when several threads write or close at the same time (bug-hunt finding F280, with a TLAPS proof in src/test/tla/), as long as only this buffer releases it: a semaphore registered with two buffers, or released by its observer, can get more. Its availablePermits() is called under a short internal lock and must return promptly.

Statistics Tracking

StreamBuffer tracks cumulative I/O statistics for the bytes flowing through the user-facing API. The trim operation is excluded: it consolidates the buffered arrays directly, without reading or writing through the streams, so the statistics always reflect user I/O only.

StreamBuffer sb = new StreamBuffer();
OutputStream os = sb.getOutputStream();
InputStream  is = sb.getInputStream();

os.write(new byte[]{1, 2, 3});        // totalBytesWritten = 3

byte[] buf = new byte[2];
is.read(buf);                          // totalBytesRead = 2

System.out.println(sb.getTotalBytesWritten()); // 3
System.out.println(sb.getTotalBytesRead());    // 2
System.out.println(sb.getMaxObservedBytes());  // 3 (peak bytes in buffer)

API:

Method	Description
getTotalBytesWritten()	Cumulative bytes written by user I/O operations (excludes internal trim)
getTotalBytesRead()	Cumulative bytes consumed by user reads (excludes internal trim)
getMaxObservedBytes()	Peak value of availableBytes ever observed

Configurable Trim Allocation Size

maxAllocationSize limits how large each individual byte array can be during consolidation. When the bytes a trim merges exceed maxAllocationSize, it produces several smaller chunks:

sb.setMaxAllocationSize(1024 * 1024);  // cap each consolidated chunk at 1 MiB
long current = sb.getMaxAllocationSize();

• Default: Integer.MAX_VALUE - 8, the largest array the JVM reliably allocates (effectively one chunk per trim pass); larger values passed to setMaxAllocationSize are capped to it.
• Throws IllegalArgumentException if the value is ≤ 0.
• Trim is skipped automatically when the resulting chunk count would not be smaller than the current Deque size — this prevents repeated no-op trims when maxAllocationSize is very small.
• Every chunk of a trim pass is allocated before any buffered entry is consolidated. If an allocation fails with OutOfMemoryError, the write that triggered the trim throws it and the buffer is left exactly as that write left it (entries, order and byte count), so every byte stays readable and the next write trims again.

Trim Observer Signals

Register Semaphore objects to be notified when a trim cycle starts or ends. This uses the same "max 1 permit" semaphore pattern as the general modification signals:

Semaphore trimStarted = new Semaphore(0);
Semaphore trimEnded   = new Semaphore(0);
sb.addTrimStartSignal(trimStarted);
sb.addTrimEndSignal(trimEnded);

// Observer thread:
trimStarted.acquire();  // blocks until trim begins
// ... trim is running ...
trimEnded.acquire();    // blocks until trim finishes

API:

Method	Description
addTrimStartSignal(Semaphore)	Registers a semaphore released when trim starts; throws NullPointerException if null
removeTrimStartSignal(Semaphore)	Removes a trim-start semaphore; returns false if not found or null
addTrimEndSignal(Semaphore)	Registers a semaphore released when trim ends; throws NullPointerException if null
removeTrimEndSignal(Semaphore)	Removes a trim-end semaphore; returns false if not found or null

API Reference

StreamBuffer

public class StreamBuffer implements Closeable

Method	Description
StreamBuffer()	Constructs a new buffer
getInputStream()	Returns the InputStream end
getOutputStream()	Returns the OutputStream end
close()	Closes both ends of the buffer
isClosed()	Returns true if the buffer is closed
isSafeWrite()	Returns the current safeWrite flag
setSafeWrite(boolean)	Enables or disables safe write (byte array cloning)
getMaxBufferElements()	Returns the current trim threshold
setMaxBufferElements(int)	Sets the trim threshold; <= 0 disables trimming
getBufferElementCount()	Returns the current number of byte arrays in the internal queue (synchronized)
getAvailableBytesExact()	Returns the exact number of buffered bytes as a long (unclamped, unlike available())
getTotalBytesWritten()	Cumulative bytes written by user I/O (excludes internal trim)
getTotalBytesRead()	Cumulative bytes consumed by user reads (excludes internal trim)
getMaxObservedBytes()	Peak value of available bytes ever observed
getMaxAllocationSize()	Returns the maximum byte-array size used during trim consolidation
setMaxAllocationSize(long)	Sets the maximum allocation size; throws IllegalArgumentException if ≤ 0
isTrimRunning()	true while trim is executing; volatile, may change at any time
addSignal(Semaphore)	Registers an external semaphore for thread-decoupled notification
removeSignal(Semaphore)	Removes a registered semaphore
addTrimStartSignal(Semaphore)	Registers a semaphore released when trim starts
removeTrimStartSignal(Semaphore)	Removes a trim-start semaphore
addTrimEndSignal(Semaphore)	Registers a semaphore released when trim ends
removeTrimEndSignal(Semaphore)	Removes a trim-end semaphore
waitForAtLeast(long)	Blocks until at least the given number of bytes are available (or the stream closes); returns fewer only for a closed stream
waitForAtLeast(long, long, TimeUnit)	Like waitForAtLeast(long), but waits at most the given time; returns the available bytes, fewer on timeout or close (isClosed() tells which)
waitForAnyData()	Blocks until at least one byte is available (or the stream closes)

Static Validation Methods

public static boolean validateOffsetAndLengthToRead(byte[] b, int off, int len)
public static boolean validateOffsetAndLengthToWrite(byte[] b, int off, int len)

Both methods mirror the parameter validation performed by InputStream.read(byte[], int, int) and OutputStream.write(byte[], int, int). They throw NullPointerException for null arrays, IndexOutOfBoundsException for invalid offsets or lengths (including integer overflow: off + len < 0), and return false for zero-length operations.

Signal/Slot Pattern

External observers register java.util.concurrent.Semaphore objects via addSignal(Semaphore). When the buffer is modified (write or close), each registered semaphore is released using the "max 1 permit" pattern — a permit is released only if the semaphore currently has zero permits. The observer blocks on semaphore.acquire() in its own thread and wakes up when data is available or the stream is closed. This provides full thread decoupling between writer and observer.

Deadlock Behavior

Read

If no data is available and the stream is not closed, read() blocks the calling thread. read(byte[], off, len) goes further: it waits until all len bytes are buffered, then takes them in one step, and returns fewer only when the stream is closed. It takes no byte while it waits. That is stricter than the InputStream contract (block until at least one byte) and unlike PipedInputStream; waiting for the full length has been the design since the first version, taking the bytes in one step since 2.0.0 (before, the first byte was taken before the wait for the rest).

To avoid blocking, only read as many bytes as available() reports. The waitForAnyData() and waitForAtLeast(long) methods can be used to wait for data before reading:

sb.waitForAnyData();                               // returns once at least one byte is buffered (or closed)
int n = is.read(buf, 0, Math.min(buf.length, Math.max(1, is.available())));

To give up after a deadline instead of only on an interrupt, use the timed variant:

long available = sb.waitForAtLeast(RECORD, 5, TimeUnit.SECONDS);
if (available < RECORD && !sb.isClosed()) {
    // timed out: nothing is taken, the bytes stay buffered
}

This matters for wrappers that read in blocks, such as InputStreamReader/BufferedReader (8 KiB): over an open buffer in a request/response protocol they wait until a whole block has arrived.

Write

Write operations never block, regardless of how much data is already buffered.

Runtime dependencies

streambuffer ships with zero transitive runtime dependencies. A downstream consumer that declares

<dependency>
    <groupId>net.ladenthin</groupId>
    <artifactId>streambuffer</artifactId>
    <version>1.2.0</version>
</dependency>

gets exactly that one artifact on its runtime classpath. mvn dependency:tree on the consumer side will show no transitive deps from streambuffer.

The two annotation libraries we use at compile time are marked <optional>true</optional> in our pom.xml:

Dep	Why it's used	What's in the bytecode	Runtime classpath impact
org.jspecify:jspecify	@NullMarked on the package, future @Nullable markers	annotation references, @Retention(CLASS)	none — JVM never loads these classes
com.google.errorprone:error_prone_annotations	@GuardedBy on internal lock-protected fields	annotation references, @Retention(CLASS)	none — JVM never loads these classes

Both annotation types have @Retention(CLASS): the references are written into streambuffer's .class files (so static analyzers like NullAway, IntelliJ, or the Checker Framework on a consumer's project can read the contract), but the JVM never loads the annotation classes at runtime. The <optional>true</optional> scope keeps the JARs off downstream runtime classpaths.

If you run static analysis on your own project and want IntelliJ / NullAway to see streambuffer's nullness contract, declare org.jspecify:jspecify yourself explicitly — it is not provided transitively.

Build

Requires Java 8 and Maven 3.9.0+.

mvn compile          # Compile
mvn test             # Run all tests with coverage
mvn package          # Build JAR
mvn install -Dgpg.skip=true  # Install locally without GPG signing

Run a single test:

mvn test -Dtest=StreamBufferTest#testSimpleRoundTrip

mvn verify also runs an API compatibility gate (japicmp): it compares the jar with the last release on Maven Central (japicmp.baseline.version in pom.xml) and fails when the version step does not match the change under semantic versioning: a binary-incompatible change needs a new major version, a new feature a new minor one. The report lands in target/japicmp/.

Run mutation tests (the test-compile prefix is required so PIT inherits JaCoCo's agent):

mvn test-compile org.pitest:pitest-maven:mutationCoverage

Testing

Tests are in StreamBufferTest using JUnit 6 (JUnit Jupiter); property-based tests live in StreamBufferProperties and use jqwik 1.9.3 (pinned — see the contributors note at the end of this section). Most behavioral tests are parameterized across three write strategies:

WriteMethod	Description
ByteArray	os.write(byte[])
Int	os.write(int)
ByteArrayWithParameter	os.write(byte[], int, int)

Test coverage includes:

• Simple and parameterized round-trip reads/writes
• Unsigned byte values (0–255), including high-byte values 128–255 and the & 0xff mask
• Partial reads with offset tracking
• Safe write with and without trim interaction
• Trim with boundary conditions (empty buffer, single entry, maxBufferElements = 0)
• Buffer trimming and byte-order preservation
• Close via sb.close(), os.close(), and is.close() — all paths tested
• available() behavior before and after close, including with buffered data remaining
• Thread interruption during blocked reads (InterruptedIOException with the InterruptedException as cause, interrupt flag kept, no byte taken), and an already interrupted thread with and without buffered bytes
• Several readers on one stream (StreamBufferConcurrentReadersTest): two parked readers and two writes or a close, three readers at the lock for two bytes (also closed meanwhile), read(byte[]) racing read(), a reader parked next to a larger waitForAtLeast, three writers and three readers
• Fixed-size records with several readers (1429-byte records with writer id, sequence number and CRC32): one writer with writes of random size and four readers, three writers of whole records and four readers — every record whole, exactly once, in writer order per reader
• Virtual threads (StreamBufferVirtualThreadsTest): 64 virtual readers and 4 virtual writers of records, also with trim and a trim signal; a JFR jdk.VirtualThreadPinned check (none inside StreamBuffer on JDK 24+, reported on JDK 21 by a scheduled job)
• waitForAtLeast(bytes, timeout, unit): returns at once, after the timeout, early when bytes arrive, on close; no wait for a timeout of zero; interrupts; two timed waiters woken by one write
• vmlens interleavings for several readers (StreamBufferSeveralReadersInterleavingTest): two readers and a close, two record readers, a trim with a trim start signal, safeWrite with a concurrent single-byte write — vmlens also fails the build on any data race
• jcstress with two readers: TwoReadersCloseRace, TwoReadersTwoWritesRace (Mode.Termination) and RecordReadersRace (torn records forbidden)
• Lincheck including reads (one at a time, only what is buffered) and a variant with maxBufferElements 2, so writes trim
• Concurrent read/write stress tests
• Parallel close without deadlock
• Deadlock-freedom of every blocking entry point, stress-checked with jcstress Mode.Termination (-Pjcstress): a thread parked in read(), in read(byte[], off, len) waiting for the rest of its bytes, or in waitForAtLeast(...) is always released by both a concurrent write and a concurrent close
• Exhaustive thread-interleaving model check of the read/write/close protocol with Java PathFinder (src/test/jpf/, scheduled CI job + local Docker recipe) — the erschöpfende counterpart to the jcstress stress tests; seven harnesses: writer and reader, writer racing a closer, two readers, two readers of fixed-size records, a waitForAtLeast caller next to a reader, the same next to a 2-byte write (what waitForAtLeast returns), trim re-entrancy through a trim signal
• Signal/slot notification via external semaphores on write and all close paths
• removeSignal(null) returning false without throwing
• addSignal(null) throwing NullPointerException
• Thread-decoupled signal barrier — observer wakes in its own thread
• validateOffsetAndLengthToRead and validateOffsetAndLengthToWrite — all branches including integer overflow
• getBufferElementCount() and getAvailableBytesExact() on an empty buffer
• waitForAnyData() / waitForAtLeast(long) with data written before and after the call
• Statistics tracking: getTotalBytesWritten, getTotalBytesRead, getMaxObservedBytes — user I/O only, excluding internal trim operations
• setMaxAllocationSize / getMaxAllocationSize — boundary values, trim with chunked allocation
• isTrimRunning() flag transitions during concurrent trim execution
• Trim observer signals (addTrimStartSignal, addTrimEndSignal) — semaphore released at correct lifecycle points
• Configuration changes during active trim (setMaxBufferElements, setMaxAllocationSize) — verified not to affect running trim
• Concurrent close during active trim — no exceptions or deadlock
• decideTrimExecution pure function — comprehensive table-driven tests covering all boundary conditions and the smart-skip edge case

Formal Verification

StreamBuffer carries machine-checked JML specifications (src/main/jml/net/ladenthin/streambuffer/StreamBuffer.jml), verified in a separate CI pipeline (formal-verification.yml) in two layers:

• Deductive proof — OpenJML ESC (Z3, and again with cvc5). The complete static, sequential core is proven against full functional contracts: both offset/length validators — including the classic (off + len) < 0 integer-overflow guard, proven equivalent to the clean bounds condition under Java wrap semantics — and the thirteen static trim-decision/arithmetic/timeout helpers, among them the ceiling division in calculateResultingChunks (proven against its multiplicative characterization m·(⌈a/m⌉−1) < a ≤ m·⌈a/m⌉, including the wrap-around boundary a = Long.MAX_VALUE − m + 1) and the full decision tree of decideTrimExecution.

• Runtime assertion checking — OpenJML RAC. The entire JUnit suite additionally runs against RAC-instrumented classes (mvn -P jml-rac); every JML contract — the proven static ones plus runtime contracts on the instance API and the two stream adapters — is then checked at runtime and a violation throws instead of scrolling by. That includes four class invariants (0 <= availableBytes, 0 <= positionAtCurrentBufferEntry, 1 <= maxAllocationSize, 0 <= maxObservedBytes) and the read/write guarantees: a read returns fewer bytes than asked for, or the end of the stream, only on a closed buffer that is then empty; a write fails only on a closed buffer; every read and write is counted in the statistics. Under concurrency these contracts stay sound because they only use monotone state (closed stays closed, a closed buffer only drains, the counters only grow).

• Model checking of the protocol — TLA+ / TLC. A TLA+ model of how threads write, close, wait and take data (src/test/tla/) is checked for safety and liveness (every waiting reader returns) with up to three readers, for read(), for fixed-size records with read(b, off, len) (also across the entries of multi-byte writes and next to a read() reader) and for waitForAtLeast. The same model with the older code fails exactly as that code did: before bug-hunt finding F278 an empty take and a reader waiting forever after the close or next to a waiter, before #212 a torn record, before F279 a count from the middle of a read. Inductive invariants of counter abstractions remove the bounds for read(), for fixed-size records with read(b, off, len) and for waitForAtLeast next to readers:

  • Apalache checks them for fixed reader counts, with any number of bytes, record size and waitForAtLeast request.
  • TLAPS proofs show them for any number of readers, and of writers for records and waitForAtLeast.
  • The link: TLC checks that the detailed model refines each abstraction, in its bounded configurations.
  • Older code: each older version reaches a violation in a bounded search.
  • Trace validation: recorded runs of the code (an instrumented copy of the class in the tests, every access to shared state logged in order) are checked against the model with TLC; 990 runs were accepted, and models of older code, a mutated model and corrupted runs are rejected.
  • Threads that call forever: with a finite stream and a closer every call returns, with fair scheduling only and no lock fairness: TLC in bounded models of the abstraction and the detailed model, TLAPS for the abstraction and any number of threads (assuming fairness by role, which follows from fairness per thread and is checked by TLC in bounded models). With a writer that writes forever and no close, TLC finds a reader that starves even with fair locks.

Concurrency itself is outside the deductive scope by JML doctrine (JML specifies sequential behavior; OpenJML does not model volatile or monitors) and is covered empirically by the Lincheck, jcstress and vmlens suites listed above, plus an exhaustive Java PathFinder interleaving model check of the read/write/close protocol (src/test/jpf/); the lock-holding discipline on the FIFO deque is additionally gated by Error Prone's @GuardedBy check in the ordinary build. The scope rationale — with citations to the published case studies that made the same split (LinkedList/TACAS 2020, IdentityHashMap/iFM 2022) — lives in the header of the specification file.

Data-Race Freedom

Why the concurrency checks above say something about real executions under the Java memory model, and not only about the interleavings they enumerate:

1. The theorem. JLS §17.4.5: a program is correctly synchronized if no sequentially consistent execution has a data race, and then all of its executions appear sequentially consistent. It is machine-checked (Aspinall & Ševčík, TPHOLs 2007; Lochbihler, JinjaThreads, TOPLAS 2014). The memory model's known problems (out-of-thin-air values, causality) concern racy programs only.
2. Every shared field is safe by construction — final, volatile, atomic, or only touched under one lock (table below).
3. vmlens checks it. It runs the scenarios through their sequentially consistent interleavings and reports a data race by the memory model's happens-before; the vmlens job of the Publish pipeline fails the build on one (There are N data races). Scenarios: a reader against a writer, two readers and a close, two record readers, a trim with a trim start signal, safeWrite with a concurrent single-byte write, two writers and an observer of a registered signal (F280). All but one run out of interleavings within the iteration limit (all of them explored); two readers and a close stops at vmlens's limit, and JPF explores that case exhaustively (TwoReaders.jpf, WaitAndRead.jpf).

Hence, for those scenarios, StreamBuffer is free of data races, its executions are sequentially consistent, and the results of the interleaving-based checks (JPF, vmlens, Lincheck, the jcstress outcomes) carry over to the Java memory model. Limits: it covers the executed scenarios, not every program; the JDK classes it builds on (ReentrantLock/Condition/AQS, Semaphore, ArrayDeque, CopyOnWriteArrayList, AtomicInteger/AtomicLong) and the JVM including its JIT are trusted.

Field	Kind	Written	Read
bufferLock, waitLock, modified, is, os	final	constructor	anywhere
buffer (ArrayDeque<byte[]>)	final, @GuardedBy("bufferLock")	under bufferLock	under bufferLock
availableBytes	volatile long	under bufferLock	under bufferLock, and lock-free by available(), getAvailableBytesExact() and the wait (fast path and under waitLock)
positionAtCurrentBufferEntry	volatile int	under bufferLock	under bufferLock
streamClosed	volatile boolean	under bufferLock (close())	under bufferLock (write), lock-free (wait, isClosed())
isTrimRunning	volatile boolean	under bufferLock (trim())	under bufferLock, lock-free by isTrimRunning()
maxObservedBytes	volatile long	under bufferLock	lock-free by the getter
safeWrite, maxBufferElements, maxAllocationSize	volatile	by their setters, any thread	once per write or trim
waitingReaders	final AtomicInteger	under waitLock	lock-free by the writer's signal
totalBytesWritten, totalBytesRead	final AtomicLong	under bufferLock (addAndGet)	lock-free by the getters
signals, trimStartSignals, trimEndSignals	final CopyOnWriteArrayList	by add/remove, any thread	iterated over a snapshot
signalLock	final	constructor	anywhere (the lock of the two deques below; taken after bufferLock, never before it)
releasingSignals, signaledWhileReleasing (ArrayDeque<Semaphore>)	final, @GuardedBy("signalLock")	under signalLock	under signalLock (the claims and notes of F280)

The byte[] contents are published through bufferLock: the writer adds the array under the lock, a reader copies from it under the lock, so the monitor's happens-before orders the two. With safeWrite off the buffer keeps the caller's array; a caller that changes it after the write races with the readers itself — that is the documented contract of Safe Write. The arrays trim() produces are allocated and filled under the lock. The skip of the wait lock when no reader waits (waitingReaders) relies on the total synchronization order of volatile accesses (JLS §17.4.4): the writer stores availableBytes before it reads the count, a waiter increments the count before it reads availableBytes, so one of them sees the other.

Contributors: do not upgrade jqwik past 1.9.3

⚠️ DO NOT UPGRADE jqwik past 1.9.3. jqwik 1.10.0 added an anti-AI prompt-injection string to test stdout; the 1.10.1 user guide states the library "is not meant to be used by any 'AI' coding agents at all." 1.9.3 is the last pre-disclosure release and is the pinned version. See CLAUDE.md section "jqwik prompt-injection in test output" for the full context. Dependabot is configured to ignore all net.jqwik updates (every version, including patches) — see the ignore rule in .github/dependabot.yml.

License

Code is under the Apache Licence v2.