feat: expose NB_WORKERS parameter setter on streaming compressors (#82)

ZstdCompressStream and ZstdOutputStream now expose the same
parameter(ZstdCompressParameter, int) surface as ZstdCompressContext,
so NB_WORKERS (plus JOB_SIZE/OVERLAP_LOG) is reachable from streaming
compression - previously MT was one-shot-only, even though unbounded
streaming input is where it pays off most. Set right after
construction, before the first compress()/write() call.

Also renames Mt* identifiers to MultiThread* for consistency with
ZstdMultithreadTest (MtCompressBenchmark -> MultiThreadCompressBenchmark,
Smoke.mtRoundTrip -> multiThreadRoundTrip, and the new/existing
zstd-jni interop test names).

Tests: unit round-trips for both streaming classes under NB_WORKERS,
plus a new zstd-jni interop test proving a Java MT ZstdOutputStream
frame decodes in zstd-jni (the reverse direction was already covered).

ADR 0015 updated to record the follow-up as resolved.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: dfa1

.github/smoke/Smoke.java

@@ -68,7 +68,7 @@ public static void main(String[] args) throws IOException {
         compressContextAdvancedParameters();
         decompressContextAdvanced();
         prefixCompression();
-        mtRoundTrip();
+        multiThreadRoundTrip();
 
         List<byte[]> samples = jsonSamples();
         ZstdDictionary dict = ZstdDictionary.train(samples, 8 * 1024);
@@ -258,7 +258,7 @@ private static void prefixCompression() {
         }
     }
 
-    private static void mtRoundTrip() {
+    private static void multiThreadRoundTrip() {
         // Native worker-thread spawning is exactly the platform-specific
         // behavior this smoke suite exists for: pthreads on glibc/musl,
         // Win32 _beginthreadex on Windows. 2 MiB clears zstd's 512 KiB

adr/0015-enable-zstd-multithread.md

@@ -82,9 +82,14 @@ compression is only for dedicated, caller-owned contexts —
 ### Risks to manage
 - PR CI compiles all six classifiers but executes tests only on linux-x86_64;
   thread spawning on Windows/macOS/musl is proven by the release-smoke matrix
-  (`Smoke.mtRoundTrip`), not by PR CI.
-- Streaming classes (`ZstdCompressStream`, `ZstdOutputStream`) have no
-  parameter setter, so MT is one-shot-only today; tracked as a follow-up.
+  (`Smoke.multiThreadRoundTrip`), not by PR CI.
+- ~~Streaming classes (`ZstdCompressStream`, `ZstdOutputStream`) have no
+  parameter setter, so MT is one-shot-only today; tracked as a follow-up.~~
+  Resolved by [issue #82](https://github.com/dfa1/zstd-java/issues/82): both
+  classes now expose `parameter(ZstdCompressParameter, int)`, set once right
+  after construction and before the first `compress`/`write` call — the same
+  lifecycle constraint above applies, since the streams already own their
+  context exclusively for their lifetime and free it on `close()`.
 
 ## Alternatives considered
 

benchmark/README.md

@@ -21,7 +21,7 @@ Plus a multithreading suite (kept separate so the single-threaded numbers stay
 longitudinally comparable, and because MT is a no-op below zstd's 512 KiB job
 minimum):
 
-- `MtCompressBenchmark` — 1 MiB / 64 MiB × `nbWorkers` 0 / 2 / 4 at level 3,
+- `MultiThreadCompressBenchmark` — 1 MiB / 64 MiB × `nbWorkers` 0 / 2 / 4 at level 3,
   zero-copy `MemorySegment` path only. The 1 MiB legs measure MT overhead on a
   single-job input (no speedup expected by design — at level 3 the default job
   size is ~8 MiB); the 64 MiB legs span multiple jobs and show the scaling.

…dfa1/zstd/bench/MtCompressBenchmark.java …/bench/MultiThreadCompressBenchmark.javabenchmark/src/main/java/io/github/dfa1/zstd/bench/MtCompressBenchmark.java renamed to benchmark/src/main/java/io/github/dfa1/zstd/bench/MultiThreadCompressBenchmark.java benchmark/src/main/java/io/github/dfa1/zstd/bench/MtCompressBenchmark.java renamed to benchmark/src/main/java/io/github/dfa1/zstd/bench/MultiThreadCompressBenchmark.java

@@ -37,7 +37,7 @@
 @Fork(value = 1, jvmArgsAppend = "--enable-native-access=ALL-UNNAMED")
 @Warmup(iterations = 3)
 @Measurement(iterations = 5)
-public class MtCompressBenchmark {
+public class MultiThreadCompressBenchmark {
 
     @Param({"1048576", "67108864"})
     private int size;

integration-tests/src/test/java/io/github/dfa1/zstd/it/ZstdJniInteropTest.java

@@ -102,12 +102,12 @@ void jniStreamToJavaStream() throws Exception {
     class Multithreaded {
 
         /// Above zstd's 512 KiB minimum job size, so workers actually engage.
-        private static final int MT_PAYLOAD_SIZE = 2 * 1024 * 1024;
+        private static final int MULTI_THREAD_PAYLOAD_SIZE = 2 * 1024 * 1024;
 
         @ParameterizedTest
         @ValueSource(ints = {1, 2})
-        void javaMtCompressJniDecompress(int nbWorkers) {
-            byte[] data = compressible(new Random(0x5EED), MT_PAYLOAD_SIZE);
+        void javaMultiThreadCompressJniDecompress(int nbWorkers) {
+            byte[] data = compressible(new Random(0x5EED), MULTI_THREAD_PAYLOAD_SIZE);
 
             byte[] frame;
             try (ZstdCompressContext ctx = new ZstdCompressContext()) {
@@ -118,8 +118,8 @@ void javaMtCompressJniDecompress(int nbWorkers) {
         }
 
         @Test
-        void jniMtCompressJavaDecompress() throws Exception {
-            byte[] data = compressible(new Random(0xFEED), MT_PAYLOAD_SIZE);
+        void jniMultiThreadCompressJavaDecompress() throws Exception {
+            byte[] data = compressible(new Random(0xFEED), MULTI_THREAD_PAYLOAD_SIZE);
 
             ByteArrayOutputStream sink = new ByteArrayOutputStream();
             try (var zout = new com.github.luben.zstd.ZstdOutputStream(sink)) {
@@ -128,6 +128,18 @@ void jniMtCompressJavaDecompress() throws Exception {
             }
             assertThat(Zstd.decompress(sink.toByteArray(), data.length)).isEqualTo(data);
         }
+
+        @Test
+        void javaMultiThreadStreamCompressJniDecompress() throws Exception {
+            byte[] data = compressible(new Random(0xC0FFEE), MULTI_THREAD_PAYLOAD_SIZE);
+
+            ByteArrayOutputStream sink = new ByteArrayOutputStream();
+            try (ZstdOutputStream zout = new ZstdOutputStream(sink)) {
+                zout.parameter(ZstdCompressParameter.NB_WORKERS, 2);
+                zout.write(data);
+            }
+            assertThat(com.github.luben.zstd.Zstd.decompress(sink.toByteArray(), data.length)).isEqualTo(data);
+        }
     }
 
     @Nested

zstd/src/main/java/io/github/dfa1/zstd/ZstdCompressStream.java

@@ -82,6 +82,21 @@ private void loadDictionary(ZstdDictionary dictionary) {
         }
     }
 
+    /// Sets an advanced compression parameter — e.g.
+    /// [ZstdCompressParameter#NB_WORKERS] for multithreaded streaming
+    /// compression. Set it right after construction, before the first
+    /// [#compress] call; changing it mid-frame is not supported.
+    ///
+    /// @param parameter the parameter to set
+    /// @param value     the value, validated natively against the parameter's bounds
+    /// @return `this`, for chaining
+    /// @throws ZstdException if the value is out of range for the parameter
+    public ZstdCompressStream parameter(ZstdCompressParameter parameter, int value) {
+        NativeCall.checkReturnValue(() -> (long) Bindings.CCTX_SET_PARAMETER.invokeExact(
+                ptr(), parameter.value(), value));
+        return this;
+    }
+
     /// Compresses as much of `src` as fits into `dst` in one step.
     ///
     /// Advance the source by [ZstdStreamResult#bytesConsumed()] and write out

zstd/src/main/java/io/github/dfa1/zstd/ZstdOutputStream.java

@@ -27,9 +27,8 @@
 /// Not thread-safe: confine an instance to a single thread.
 public final class ZstdOutputStream extends OutputStream {
 
-    // ZSTD_cParameter / ZSTD_EndDirective values from zstd.h — see
+    // ZSTD_EndDirective values from zstd.h — see
     // https://facebook.github.io/zstd/doc/api_manual_latest.html
-    private static final int ZSTD_C_COMPRESSION_LEVEL = 100;
     private static final int ZSTD_E_CONTINUE = 0;
     private static final int ZSTD_E_FLUSH = 1;
     private static final int ZSTD_E_END = 2;
@@ -105,7 +104,7 @@ public ZstdOutputStream(OutputStream out, int level, ZstdDictionary dictionary)
             c = NativeCall.createOrThrow("ZSTD_createCCtx", () -> (MemorySegment) Bindings.CREATE_CCTX.invokeExact());
             this.cctx = c;
             NativeCall.checkReturnValue(() -> (long) Bindings.CCTX_SET_PARAMETER.invokeExact(
-                    cctx, ZSTD_C_COMPRESSION_LEVEL, level));
+                    cctx, ZstdCompressParameter.COMPRESSION_LEVEL.value(), level));
             if (dictionary != null) {
                 loadDictionary(dictionary);
             }
@@ -142,6 +141,21 @@ private void loadDictionary(ZstdDictionary dictionary) {
         }
     }
 
+    /// Sets an advanced compression parameter — e.g.
+    /// [ZstdCompressParameter#NB_WORKERS] for multithreaded compression of a
+    /// large or unbounded [#write] stream. Set it right after construction,
+    /// before the first `write`; changing it mid-frame is not supported.
+    ///
+    /// @param parameter the parameter to set
+    /// @param value     the value, validated natively against the parameter's bounds
+    /// @return `this`, for chaining
+    /// @throws ZstdException if the value is out of range for the parameter
+    public ZstdOutputStream parameter(ZstdCompressParameter parameter, int value) {
+        NativeCall.checkReturnValue(() -> (long) Bindings.CCTX_SET_PARAMETER.invokeExact(
+                cctx, parameter.value(), value));
+        return this;
+    }
+
     @Override
     public void write(int b) throws IOException {
         single[0] = (byte) b;

zstd/src/test/java/io/github/dfa1/zstd/ZstdMultithreadTest.java

@@ -5,8 +5,14 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
 import java.nio.charset.StandardCharsets;
 
+import static io.github.dfa1.zstd.ZstdTestSupport.bytesOf;
+import static io.github.dfa1.zstd.ZstdTestSupport.segmentOf;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /// Multithreaded compression via `NB_WORKERS`. zstd engages workers only when
@@ -77,6 +83,50 @@ void honorsJobSizeAndOverlapLog() {
         }
     }
 
+    @Nested
+    class SegmentStreamRoundTrip {
+
+        @ParameterizedTest
+        @ValueSource(ints = {1, 2})
+        void roundTripsWithWorkers(int nbWorkers) {
+            // Given a segment stream configured with worker threads
+            byte[] frame;
+            try (Arena arena = Arena.ofConfined();
+                 ZstdCompressStream sut = new ZstdCompressStream()) {
+                sut.parameter(ZstdCompressParameter.NB_WORKERS, nbWorkers);
+
+                // When compressing a payload large enough to engage the workers
+                MemorySegment src = segmentOf(arena, LARGE_PAYLOAD);
+                MemorySegment dst = arena.allocate(Zstd.compressBound(LARGE_PAYLOAD.length));
+                ZstdStreamResult r = sut.compress(dst, src, ZstdEndDirective.END);
+                frame = bytesOf(dst, r.bytesProduced());
+            }
+
+            // Then the frame decompresses back to the original
+            assertThat(Zstd.decompress(frame)).isEqualTo(LARGE_PAYLOAD);
+        }
+    }
+
+    @Nested
+    class OutputStreamRoundTrip {
+
+        @ParameterizedTest
+        @ValueSource(ints = {1, 2})
+        void roundTripsWithWorkers(int nbWorkers) throws IOException {
+            // Given an output stream configured with worker threads
+            ByteArrayOutputStream sink = new ByteArrayOutputStream();
+            try (ZstdOutputStream sut = new ZstdOutputStream(sink)) {
+                sut.parameter(ZstdCompressParameter.NB_WORKERS, nbWorkers);
+
+                // When writing a payload large enough to engage the workers
+                sut.write(LARGE_PAYLOAD);
+            }
+
+            // Then the frame decompresses back to the original
+            assertThat(Zstd.decompress(sink.toByteArray(), LARGE_PAYLOAD.length)).isEqualTo(LARGE_PAYLOAD);
+        }
+    }
+
     @Nested
     class Lifecycle {