From d38d9bffe9ffbda81b6cd4ec27f54864f628cd73 Mon Sep 17 00:00:00 2001 From: Corey Tasz Date: Mon, 28 Sep 2026 14:04:44 +0000 Subject: [PATCH] Add ZSTD_decompressMultiThreaded(): frame-level parallel decompression Decompresses concatenated frames concurrently, one frame per worker thread, addressing #2245. Frame boundaries are scanned up front with ZSTD_findFrameCompressedSize(); each frame is then decompressed by its own job with its own decompression context (blocks within a frame are left serial, since later blocks may reference earlier output). - nbThreads <= 1 and single-frame inputs take the plain serial path. - Frames with stored content sizes decompress straight into disjoint dst slices; each job gets exactly its frame's content size as dstCapacity, since the decoder may use spare dst room as scratch. - Frames without a stored content size use per-job temp buffers that grow until the frame fits, then copy into dst in frame order. - Strict all-or-nothing input validation: bad magic -> prefix_unknown, truncation -> srcSize_wrong, short dst -> dstSize_tooSmall. - Without ZSTD_MULTITHREAD the pool runs jobs synchronously, so the result stays correct (just serial). Tests: tests/mtDecompressTest covers thread counts 1-16, the unknown-content-size path, skippable frames, and exact error codes. --- lib/decompress/zstd_decompress.c | 211 ++++++++++++++++++++++++++ lib/zstd.h | 32 ++++ tests/.gitignore | 1 + tests/Makefile | 6 + tests/mtDecompressTest.c | 248 +++++++++++++++++++++++++++++++ 5 files changed, 498 insertions(+) create mode 100644 tests/mtDecompressTest.c diff --git a/lib/decompress/zstd_decompress.c b/lib/decompress/zstd_decompress.c index 325f0b54dfc..5ef342542a8 100644 --- a/lib/decompress/zstd_decompress.c +++ b/lib/decompress/zstd_decompress.c @@ -61,6 +61,7 @@ #include "../common/error_private.h" #include "../common/zstd_internal.h" /* blockProperties_t */ #include "../common/mem.h" /* low level memory routines */ +#include "../common/pool.h" /* thread pool, for ZSTD_decompressMultiThreaded() */ #include "../common/bits.h" /* ZSTD_highbit32 */ #define FSE_STATIC_LINKING_ONLY #include "../common/fse.h" @@ -1216,6 +1217,216 @@ size_t ZSTD_decompress(void* dst, size_t dstCapacity, const void* src, size_t sr } +/*-************************************** +* Multi-threaded decompression +* Decompresses concatenated frames in parallel, one frame per job. +* Threading is provided by the common thread pool (POOL_*). +* Without ZSTD_MULTITHREAD the pool executes jobs synchronously, +* so the result stays correct, just serial. +****************************************/ + +typedef struct { + size_t srcOffset; + size_t srcSize; + unsigned long long contentSize; /* ZSTD_CONTENTSIZE_UNKNOWN or ZSTD_CONTENTSIZE_ERROR when not known */ +} ZSTD_mtFrameInfo; + +typedef struct { + const void* src; + size_t srcSize; + void* dst; /* direct mode: output slice; temp mode: NULL */ + size_t dstCapacity; /* direct mode: capacity of the output slice */ + size_t initCap; /* temp mode: initial buffer size (0 = derive from srcSize) */ + size_t maxCap; /* temp mode: temp buffer never grows past this */ + void* tmpBuf; /* temp mode: decompressed data (owned by caller) */ + size_t result; /* bytes decompressed, or error code */ +} ZSTD_mtFrameJob; + +static void ZSTD_mtDecompressDirect(void* opaque) +{ + ZSTD_mtFrameJob* const job = (ZSTD_mtFrameJob*)opaque; + /* one-shot decompression creates its own context: no shared state */ + job->result = ZSTD_decompress(job->dst, job->dstCapacity, job->src, job->srcSize); +} + +/* temp mode: content size wasn't stored in the frame header, + * so grow a buffer until the frame fits (bounded by maxCap). */ +static void ZSTD_mtDecompressTemp(void* opaque) +{ + ZSTD_mtFrameJob* const job = (ZSTD_mtFrameJob*)opaque; + size_t cap = job->initCap; + if (cap == 0) { + cap = job->srcSize * 4; /* guess: 4x compressed size */ + if (cap < job->srcSize) cap = job->maxCap; /* multiplication overflowed */ + } + if (cap > job->maxCap) cap = job->maxCap; + if (cap == 0) cap = 1; /* malloc(0) may return NULL */ + for (;;) { + void* buf = ZSTD_malloc(cap); + size_t rr; + if (buf == NULL) { job->result = ERROR(memory_allocation); return; } + rr = ZSTD_decompress(buf, cap, job->src, job->srcSize); + if (!ZSTD_isError(rr)) { job->tmpBuf = buf; job->result = rr; return; } + ZSTD_free(buf); + if (ZSTD_getErrorCode(rr) != ZSTD_error_dstSize_tooSmall) { job->result = rr; return; } + /* a frame larger than maxCap can never fit into the caller's dst */ + if (cap >= job->maxCap) { job->result = ERROR(dstSize_tooSmall); return; } + { size_t const newCap = cap * 2; + cap = (newCap > job->maxCap || newCap < cap) ? job->maxCap : newCap; + } + } +} + +size_t ZSTD_decompressMultiThreaded(void* dst, size_t dstCapacity, + const void* src, size_t srcSize, + int nbThreads) +{ + ZSTD_mtFrameInfo* frames = NULL; + ZSTD_mtFrameJob* jobs = NULL; + POOL_ctx* pool; + size_t nbFrames = 0, framesCap = 0; + size_t offset = 0; + size_t f, g; + size_t nbWorkers; + int allKnown; + unsigned long long totalKnown = 0; + size_t totalOut = 0; + + if (srcSize == 0) return ERROR(srcSize_wrong); + if (nbThreads <= 1) + return ZSTD_decompress(dst, dstCapacity, src, srcSize); + + /* 1. scan all frame boundaries up front; strict all-or-nothing: + * bad magic -> prefix_unknown, truncation -> srcSize_wrong */ + while (offset < srcSize) { + size_t const fcs = ZSTD_findFrameCompressedSize((const char*)src + offset, srcSize - offset); + if (ZSTD_isError(fcs)) { ZSTD_free(frames); return fcs; } + if (nbFrames == framesCap) { + size_t const newCap = framesCap ? framesCap * 2 : 16; + ZSTD_mtFrameInfo* const newFrames = (ZSTD_mtFrameInfo*)ZSTD_malloc(newCap * sizeof(*newFrames)); + if (newFrames == NULL) { ZSTD_free(frames); return ERROR(memory_allocation); } + ZSTD_memcpy(newFrames, frames, nbFrames * sizeof(*frames)); + ZSTD_free(frames); + frames = newFrames; + framesCap = newCap; + } + frames[nbFrames].srcOffset = offset; + frames[nbFrames].srcSize = fcs; + nbFrames++; + offset += fcs; + } + + /* 2. a single frame takes the plain serial path: no thread overhead */ + if (nbFrames == 1) { + size_t const r = ZSTD_decompress(dst, dstCapacity, src, srcSize); + ZSTD_free(frames); + return r; + } + + /* 3. read each frame's content size; overflow-safe running total */ + allKnown = 1; + for (f = 0; f < nbFrames; f++) { + unsigned long long const cs = + ZSTD_getFrameContentSize((const char*)src + frames[f].srcOffset, + frames[f].srcSize); + frames[f].contentSize = cs; + if (cs == ZSTD_CONTENTSIZE_UNKNOWN || cs == ZSTD_CONTENTSIZE_ERROR) { + allKnown = 0; + } else { + if (cs > dstCapacity || totalKnown > (unsigned long long)dstCapacity - cs) { + ZSTD_free(frames); + return ERROR(dstSize_tooSmall); + } + totalKnown += cs; + } + } + + /* 4. one job per frame; never more workers than frames */ + nbWorkers = (size_t)nbThreads < nbFrames ? (size_t)nbThreads : nbFrames; + pool = POOL_create(nbWorkers, 0); + if (pool == NULL) { ZSTD_free(frames); return ERROR(memory_allocation); } + jobs = (ZSTD_mtFrameJob*)ZSTD_malloc(nbFrames * sizeof(*jobs)); + if (jobs == NULL) { POOL_free(pool); ZSTD_free(frames); return ERROR(memory_allocation); } + + if (allKnown) { + /* every size known and total fits: decompress straight into dst slices. + * NOTE: each job's dstCapacity is exactly its frame's content size, not + * the remaining tail of dst. ZSTD_decompress() is allowed to use spare + * dstCapacity past the frame's end as scratch space (it parks the + * block's literals there, see ZSTD_allocateLiteralsBuffer()), so + * handing a job the whole tail would let it scribble into the next + * frame's slice while that frame decompresses concurrently. */ + size_t dstOffset = 0; + for (f = 0; f < nbFrames; f++) { + jobs[f].src = (const char*)src + frames[f].srcOffset; + jobs[f].srcSize = frames[f].srcSize; + jobs[f].dst = (char*)dst + dstOffset; + jobs[f].dstCapacity = (size_t)frames[f].contentSize; + jobs[f].tmpBuf = NULL; + jobs[f].result = 0; + dstOffset += (size_t)frames[f].contentSize; + POOL_add(pool, ZSTD_mtDecompressDirect, &jobs[f]); + } + } else { + /* some sizes unknown: temp buffer per frame, gather into dst in order */ + for (f = 0; f < nbFrames; f++) { + unsigned long long const cs = frames[f].contentSize; + int const known = (cs != ZSTD_CONTENTSIZE_UNKNOWN) && (cs != ZSTD_CONTENTSIZE_ERROR); + jobs[f].src = (const char*)src + frames[f].srcOffset; + jobs[f].srcSize = frames[f].srcSize; + jobs[f].dst = NULL; + jobs[f].initCap = (known && cs > 0) + ? (cs > (unsigned long long)dstCapacity ? dstCapacity : (size_t)cs) + : 0; + jobs[f].maxCap = dstCapacity; + jobs[f].tmpBuf = NULL; + jobs[f].result = 0; + if (known && cs == 0) + continue; /* empty frame (incl. skippable): nothing to do */ + POOL_add(pool, ZSTD_mtDecompressTemp, &jobs[f]); + } + } + ZSTD_free(frames); frames = NULL; + + POOL_joinJobs(pool); + POOL_free(pool); + + /* 5. collect: first error wins */ + for (f = 0; f < nbFrames; f++) { + if (ZSTD_isError(jobs[f].result)) { + size_t const err = jobs[f].result; + for (g = 0; g < nbFrames; g++) ZSTD_free(jobs[g].tmpBuf); + ZSTD_free(jobs); + return err; + } + } + if (allKnown) { + totalOut = (size_t)totalKnown; + } else { + unsigned long long total = 0; + char* op = (char*)dst; + for (f = 0; f < nbFrames; f++) { + /* each frame fit in maxCap == dstCapacity, but the sum may not */ + if (jobs[f].result > dstCapacity + || total > (unsigned long long)dstCapacity - jobs[f].result) { + for (g = 0; g < nbFrames; g++) ZSTD_free(jobs[g].tmpBuf); + ZSTD_free(jobs); + return ERROR(dstSize_tooSmall); + } + total += jobs[f].result; + } + for (f = 0; f < nbFrames; f++) { + if (jobs[f].result) ZSTD_memcpy(op, jobs[f].tmpBuf, jobs[f].result); + op += jobs[f].result; + ZSTD_free(jobs[f].tmpBuf); + } + totalOut = (size_t)total; + } + ZSTD_free(jobs); + return totalOut; +} + + /*-************************************** * Advanced Streaming Decompression API * Bufferless and synchronous diff --git a/lib/zstd.h b/lib/zstd.h index 97fef316fd6..cd14659c521 100644 --- a/lib/zstd.h +++ b/lib/zstd.h @@ -173,6 +173,38 @@ ZSTDLIB_API size_t ZSTD_compress( void* dst, size_t dstCapacity, ZSTDLIB_API size_t ZSTD_decompress( void* dst, size_t dstCapacity, const void* src, size_t compressedSize); +/*! ZSTD_decompressMultiThreaded() : + * Same as ZSTD_decompress(), but decompresses concatenated frames in parallel, + * using `nbThreads` worker threads. + * + * Frame boundaries are discovered up front, without decompressing, and every + * frame is then decompressed by its own job with its own decompression + * context. This is the recommended way to parallelize decompression: + * blocks _within_ a single frame cannot be decompressed independently + * (later blocks may reference earlier output), but independent _frames_ can. + * + * `src` must contain an integer number of complete frames. + * Standard, skippable and legacy frames are all accepted and may be mixed. + * The whole input is validated strictly before any output is produced: + * an error in any frame aborts the entire operation. + * `nbThreads` : number of worker threads. Values <= 1 select the plain serial + * ZSTD_decompress() path (no thread is created, no overhead). + * When `nbThreads` exceeds the number of frames, it is clamped down to it. + * Note: actual parallelism requires the library to be built with + * ZSTD_MULTITHREAD; otherwise jobs run one after another (still correct). + * @return : total number of decompressed bytes written into `dst` + * (<= `dstCapacity`), or an error code (test with ZSTD_isError()). + * Error codes match ZSTD_decompress(), plus: + * - "srcSize_wrong" : `src` is empty, or a frame is truncated. + * - "prefix_unknown" : a frame has an invalid magic number. + * - "dstSize_tooSmall": the decompressed data does not fit into `dst`. + * Note: frames compressed without a stored content size (e.g. via streaming) + * are decompressed into temporary buffers first, then copied into `dst` + * in frame order. */ +ZSTDLIB_API size_t ZSTD_decompressMultiThreaded(void* dst, size_t dstCapacity, + const void* src, size_t srcSize, + int nbThreads); + /*====== Decompression helper functions ======*/ diff --git a/tests/.gitignore b/tests/.gitignore index 0f155764b96..bb38800dc37 100644 --- a/tests/.gitignore +++ b/tests/.gitignore @@ -6,6 +6,7 @@ fuzzer fuzzer32 fuzzer-dll largeDictionary +mtDecompressTest zbufftest zbufftest32 zbufftest-dll diff --git a/tests/Makefile b/tests/Makefile index e268ddeb9e4..d4d47aaee88 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -208,6 +208,12 @@ zstreamtest_ubsan : CFLAGS += -fsanitize=undefined zstreamtest_ubsan : $(ZSTREAMFILES) $(LINK.c) $(MULTITHREAD) $^ -o $@$(EXT) +CLEAN += mtDecompressTest +mtDecompressTest : CPPFLAGS += $(MULTITHREAD_CPP) +mtDecompressTest : LDFLAGS += $(MULTITHREAD_LD) +mtDecompressTest : $(ZSTDMT_OBJECTS) mtDecompressTest.c + $(LINK.c) $^ -o $@$(EXT) + # note : broken : requires symbols unavailable from dynamic library zstreamtest-dll : $(LIB_SRCDIR)/common/xxhash.c # xxh symbols not exposed from dll zstreamtest-dll : $(ZSTREAM_LOCAL_FILES) diff --git a/tests/mtDecompressTest.c b/tests/mtDecompressTest.c new file mode 100644 index 00000000000..6bf1b42ead0 --- /dev/null +++ b/tests/mtDecompressTest.c @@ -0,0 +1,248 @@ +/* + * mtDecompressTest.c — test driver for ZSTD_decompressMultiThreaded() + * + * Builds multi-frame buffers (with and without stored content sizes, + * mixed with skippable frames) and checks that multi-threaded + * decompression is byte-identical to serial ZSTD_decompress(), + * plus a battery of edge cases. + * + * Returns 0 on success, 1 on any failure. + */ + +#include +#include +#include + +#define ZSTD_STATIC_LINKING_ONLY +#include "../lib/zstd.h" + +static int failures = 0; +#define CHECK(cond, ...) do { \ + if (!(cond)) { printf("FAIL: "); printf(__VA_ARGS__); printf("\n"); failures++; } \ +} while (0) + +/* simple deterministic PRNG (xorshift64) */ +static unsigned long long rng_state = 0x123456789abcdefULL; +static unsigned char rng_next(void) { + unsigned long long x = rng_state; + x ^= x << 13; x ^= x >> 7; x ^= x << 17; + rng_state = x; + return (unsigned char)(x >> 33); +} + +/* compressible-ish data: mostly repeating patterns with noise */ +static void gen_data(unsigned char* buf, size_t size, int seed) { + size_t i; + rng_state = 0x9e3779b97f4a7c15ULL + (unsigned long long)seed; + for (i = 0; i < size; i++) { + unsigned char r = rng_next(); + buf[i] = (r < 200) ? (unsigned char)(i & 0xFF) : r; /* ~78% pattern */ + } +} + +/* one compressed frame; contentSizeFlag 1 = store content size, 0 = streaming-like */ +static size_t make_frame(unsigned char* dst, size_t dstCap, + const unsigned char* src, size_t srcSize, + int contentSizeFlag, int level) { + ZSTD_CCtx* cctx = ZSTD_createCCtx(); + size_t r; + ZSTD_CCtx_setParameter(cctx, ZSTD_c_contentSizeFlag, contentSizeFlag); + ZSTD_CCtx_setParameter(cctx, ZSTD_c_compressionLevel, level); + r = ZSTD_compress2(cctx, dst, dstCap, src, srcSize); + ZSTD_freeCCtx(cctx); + return r; +} + +static size_t make_skippable(unsigned char* dst, size_t dstCap, + const unsigned char* payload, size_t payloadSize) { + return ZSTD_writeSkippableFrame(dst, dstCap, payload, payloadSize, 7); +} + +int main(void) { + /* ---- build an 8-frame buffer: varied sizes 1KB..1MB, 2 skippable mixed in ---- */ + static const size_t frameSizes[8] = { + 1024, 65536, 300000, 1048576, 4096, 262144, 77777, 1048576 + }; + unsigned char* originals[8]; + unsigned char* frames[8]; + size_t frameCSizes[8]; + int isSkippable[8] = { 0, 0, 1, 0, 0, 1, 0, 0 }; + unsigned char* multi = NULL; + size_t multiCap = 0, multiSize = 0; + unsigned char* serialOut = NULL; + size_t totalOrig = 0; + int i, t; + static const int threadCounts[] = { 1, 2, 4, 8, 16 }; + unsigned char skipPayload[64]; + + printf("ZSTD version %s — mtDecompressTest\n", ZSTD_versionString()); + + for (i = 0; i < 64; i++) skipPayload[i] = (unsigned char)i; + for (i = 0; i < 8; i++) { + size_t bound; + originals[i] = (unsigned char*)malloc(frameSizes[i]); + gen_data(originals[i], frameSizes[i], i); + bound = ZSTD_compressBound(frameSizes[i]) + 64; + frames[i] = (unsigned char*)malloc(bound); + if (isSkippable[i]) { + frameCSizes[i] = make_skippable(frames[i], bound, skipPayload, sizeof(skipPayload)); + } else { + frameCSizes[i] = make_frame(frames[i], bound, originals[i], frameSizes[i], 1, 3); + } + CHECK(!ZSTD_isError(frameCSizes[i]), "frame %d compress failed: %s", i, ZSTD_getErrorName(frameCSizes[i])); + multiCap += frameCSizes[i]; + totalOrig += isSkippable[i] ? 0 : frameSizes[i]; + } + multi = (unsigned char*)malloc(multiCap); + for (i = 0; i < 8; i++) { + memcpy(multi + multiSize, frames[i], frameCSizes[i]); + multiSize += frameCSizes[i]; + } + serialOut = (unsigned char*)malloc(totalOrig ? totalOrig : 1); + + /* serial reference */ + { + size_t r = ZSTD_decompress(serialOut, totalOrig, multi, multiSize); + CHECK(!ZSTD_isError(r), "serial reference failed: %s", ZSTD_getErrorName(r)); + CHECK(r == totalOrig, "serial size %u != %u", (unsigned)r, (unsigned)totalOrig); + } + /* concatenated originals reference (skip skippable frames) */ + { + unsigned char* cat = (unsigned char*)malloc(totalOrig ? totalOrig : 1); + size_t pos = 0; + for (i = 0; i < 8; i++) { + if (!isSkippable[i]) { memcpy(cat + pos, originals[i], frameSizes[i]); pos += frameSizes[i]; } + } + CHECK(memcmp(serialOut, cat, totalOrig) == 0, "serial != concatenated originals"); + free(cat); + } + + /* ---- main test: thread counts 1,2,4,8,16 ---- */ + for (t = 0; t < 5; t++) { + int nbT = threadCounts[t]; + unsigned char* out = (unsigned char*)malloc(totalOrig ? totalOrig : 1); + size_t r = ZSTD_decompressMultiThreaded(out, totalOrig, multi, multiSize, nbT); + CHECK(!ZSTD_isError(r), "nbThreads=%d failed: %s", nbT, ZSTD_getErrorName(r)); + CHECK(r == totalOrig, "nbThreads=%d size %u != %u", nbT, (unsigned)r, (unsigned)totalOrig); + CHECK(memcmp(out, serialOut, totalOrig) == 0, "nbThreads=%d output != serial", nbT); + free(out); + printf(" threads=%2d OK (%u bytes)\n", nbT, (unsigned)totalOrig); + } + + /* ---- unknown content-size path (contentSizeFlag=0): temp-buffer fallback ---- */ + { + unsigned char* multi2 = (unsigned char*)malloc(multiCap); + size_t multi2Size = 0, total2 = 0; + for (i = 0; i < 8; i++) { + size_t bound = ZSTD_compressBound(frameSizes[i]) + 64; + unsigned char* f = (unsigned char*)malloc(bound); + size_t cs; + if (isSkippable[i]) { + cs = make_skippable(f, bound, skipPayload, sizeof(skipPayload)); + } else { + cs = make_frame(f, bound, originals[i], frameSizes[i], 0, 3); + /* confirm the flag took effect */ + CHECK(ZSTD_getFrameContentSize(f, cs) == ZSTD_CONTENTSIZE_UNKNOWN, + "frame %d unexpectedly stores content size", i); + total2 += frameSizes[i]; + } + CHECK(!ZSTD_isError(cs), "frame %d (noSize) compress failed", i); + memcpy(multi2 + multi2Size, f, cs); + multi2Size += cs; + free(f); + } + for (t = 0; t < 5; t++) { + int nbT = threadCounts[t]; + unsigned char* out = (unsigned char*)malloc(total2 ? total2 : 1); + size_t r = ZSTD_decompressMultiThreaded(out, total2, multi2, multi2Size, nbT); + CHECK(!ZSTD_isError(r), "noSize nbThreads=%d failed: %s", nbT, ZSTD_getErrorName(r)); + CHECK(r == total2, "noSize nbThreads=%d size %u != %u", nbT, (unsigned)r, (unsigned)total2); + CHECK(memcmp(out, serialOut, total2) == 0, "noSize nbThreads=%d output != serial", nbT); + free(out); + } + printf(" unknown-content-size path OK (threads 1,2,4,8,16)\n"); + free(multi2); + } + + /* ---- edge cases ---- */ + { + unsigned char* out = (unsigned char*)malloc(totalOrig ? totalOrig : 1); + unsigned char* bad = (unsigned char*)malloc(multiSize); + size_t r; + + /* empty input -> error */ + r = ZSTD_decompressMultiThreaded(out, totalOrig, multi, 0, 4); + CHECK(ZSTD_isError(r), "empty input should error"); + CHECK(ZSTD_getErrorCode(r) == ZSTD_error_srcSize_wrong, + "empty input: expected srcSize_wrong, got %s", ZSTD_getErrorName(r)); + + /* truncated final frame -> error */ + memcpy(bad, multi, multiSize); + r = ZSTD_decompressMultiThreaded(out, totalOrig, bad, multiSize - 10, 4); + CHECK(ZSTD_isError(r), "truncated input should error (got %u)", (unsigned)r); + CHECK(ZSTD_getErrorCode(r) == ZSTD_error_srcSize_wrong, + "truncated input: expected srcSize_wrong, got %s", ZSTD_getErrorName(r)); + + /* corrupt magic -> error */ + memcpy(bad, multi, multiSize); + bad[0] ^= 0xFF; + r = ZSTD_decompressMultiThreaded(out, totalOrig, bad, multiSize, 4); + CHECK(ZSTD_isError(r), "corrupt magic should error"); + CHECK(ZSTD_getErrorCode(r) == ZSTD_error_prefix_unknown, + "corrupt magic: expected prefix_unknown, got %s", ZSTD_getErrorName(r)); + + /* dst one byte short -> dstSize_tooSmall */ + r = ZSTD_decompressMultiThreaded(out, totalOrig - 1, multi, multiSize, 4); + CHECK(ZSTD_getErrorCode(r) == ZSTD_error_dstSize_tooSmall, + "short dst: expected dstSize_tooSmall, got %s", ZSTD_getErrorName(r)); + + /* nbThreads=0 and negative -> serial behavior */ + r = ZSTD_decompressMultiThreaded(out, totalOrig, multi, multiSize, 0); + CHECK(!ZSTD_isError(r) && r == totalOrig && memcmp(out, serialOut, totalOrig) == 0, + "nbThreads=0 should behave serial"); + r = ZSTD_decompressMultiThreaded(out, totalOrig, multi, multiSize, -3); + CHECK(!ZSTD_isError(r) && r == totalOrig && memcmp(out, serialOut, totalOrig) == 0, + "nbThreads=-3 should behave serial"); + + /* nbThreads >> nframes */ + r = ZSTD_decompressMultiThreaded(out, totalOrig, multi, multiSize, 64); + CHECK(!ZSTD_isError(r) && r == totalOrig && memcmp(out, serialOut, totalOrig) == 0, + "nbThreads=64 should work"); + + /* single frame -> serial path, identical result */ + r = ZSTD_decompressMultiThreaded(out, frameSizes[3], frames[3], frameCSizes[3], 8); + { + size_t rs = ZSTD_decompress(serialOut, frameSizes[3], frames[3], frameCSizes[3]); + CHECK(!ZSTD_isError(r) && r == rs && memcmp(out, serialOut, r) == 0, + "single frame should equal serial"); + } + + /* skippable-only input -> 0 bytes, no error */ + { + unsigned char sk[128]; + size_t sks = make_skippable(sk, sizeof(sk), skipPayload, sizeof(skipPayload)); + r = ZSTD_decompressMultiThreaded(out, 16, sk, sks, 4); + CHECK(!ZSTD_isError(r) && r == 0, "skippable-only should return 0 (got %s)", + ZSTD_isError(r) ? ZSTD_getErrorName(r) : "ok"); + } + + /* garbage input -> error */ + for (i = 0; i < 64; i++) bad[i] = rng_next(); + r = ZSTD_decompressMultiThreaded(out, totalOrig, bad, 64, 4); + CHECK(ZSTD_isError(r), "garbage input should error"); + CHECK(ZSTD_getErrorCode(r) == ZSTD_error_prefix_unknown, + "garbage input: expected prefix_unknown, got %s", ZSTD_getErrorName(r)); + + free(bad); + free(out); + printf(" edge cases OK\n"); + } + + for (i = 0; i < 8; i++) { free(originals[i]); free(frames[i]); } + free(multi); + free(serialOut); + + if (failures == 0) printf("PASS: all mtDecompressTest checks passed\n"); + else printf("FAIL: %d check(s) failed\n", failures); + return failures != 0; +}