From c6e5a8cf534a0473eada203a239db9293a965ccf Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 19 Oct 2026 15:21:50 -0700 Subject: [PATCH] review round 9: destination too small invalidates MT frames (test), --rm keeps a renamed or replaced source, multithreaded CLI tests, MSan step (checkpoint) --- .github/workflows/sanitizers.yml | 7 ++ NEWS | 4 +- build/VS2022/_test.bat | 13 +++ build/cmake/CMakeLists.txt | 2 + build/meson/meson/lib/meson.build | 3 +- lib/lz4frame.c | 24 +++--- programs/lz4.1 | 4 +- programs/lz4.1.md | 6 +- programs/lz4io.c | 31 ++++++-- programs/lz4io.h | 5 +- tests/Makefile | 6 +- tests/frameMT.c | 109 ++++++++++++++++++++++++- tests/test-lz4-multithread.sh | 128 ++++++++++++++++++++++++++++++ 13 files changed, 312 insertions(+), 30 deletions(-) create mode 100755 tests/test-lz4-multithread.sh diff --git a/.github/workflows/sanitizers.yml b/.github/workflows/sanitizers.yml index 53976a0b..dc31a1b0 100644 --- a/.github/workflows/sanitizers.yml +++ b/.github/workflows/sanitizers.yml @@ -86,6 +86,13 @@ jobs: make clean CC=clang CFLAGS=-fsanitize=memory LDFLAGS=-fsanitize=memory make -j -C tests test-frametest V=1 + - name: MSan - multi-threaded frame compression + # build with multi-threading support only : the build without it runs the same code serially + run: | + make clean + CC=clang CFLAGS=-fsanitize=memory LDFLAGS=-fsanitize=memory make -j -C tests frameMT V=1 + cd tests && ./frameMT -q + - name: MSan - fuzzer test run: | make clean diff --git a/NEWS b/NEWS index 1bf04b4b..04f1e84b 100644 --- a/NEWS +++ b/NEWS @@ -38,7 +38,7 @@ api : new static functions `LZ4F_cctx_size()` and `LZ4F_dctx_size()`, by @felix api : new static function `LZ4F_resumeFrame()` : resume decoding from any block start, for random access within frames api : `lz4file` : new static functions `LZ4F_seek()` and `LZ4F_tell()`, for random access within compressed files api : `LZ4_ACCELERATION_DEFAULT` and `LZ4_ACCELERATION_MAX` are defined in `lz4.h` (static section) -api : new `LZ4F_preferences_t.nbWorkers` (experimental) : multithreaded lz4frame compression, same output whatever the nb of threads +api : new `LZ4F_preferences_t.nbWorkers` (experimental) : multithreaded lz4frame compression, same output for any `nbWorkers` >= 1 api : `LZ4F_compressOptions_t.stableSrc` : new value `LZ4F_srcStableUntilFlush`, letting worker threads read input in place api : reserved fields of `LZ4F_preferences_t` and `LZ4F_compressOptions_t` are now scalars : flatten positional initializers `{ 0, 0, 0 }` cli : decodes Firefox's mozlz4 files (`.jsonlz4`, `.mozlz4`), proposed by @UnitedMarsupials @@ -46,7 +46,7 @@ cli : decodes legacy frames ended by four zero bytes (Linux kernel end marker), cli : compressed output no longer depends on multithreading support (`lz4` and `lz4-nomt` produce the same frames, also with `-BD`) cli : custom block sizes (`-B#`, # >= 32) are honored, and compressed single-threaded cli : `-T#` : nb of worker threads, beside the main thread ; automatic nb limited by a memory budget (`--memory=#`, `LZ4_MEMORY`) and by file size -cli : a source file modified during its compression is detected, and kept with `--rm` ; error 54 when the output can't be verified +cli : a source file modified, renamed or replaced during its compression is detected, and kept with `--rm` ; error 54 when the output can't be verified cli : `lz4 -l` could hang on Windows with an empty input, as could decoding the resulting legacy stream cli : `--rm` keeps the source file after a decompression failure, by @Furan666 cli : `lz4 -t -m` no longer writes decoded content to stdout, reported by @tansy diff --git a/build/VS2022/_test.bat b/build/VS2022/_test.bat index 8021aaab..5bcbd701 100644 --- a/build/VS2022/_test.bat +++ b/build/VS2022/_test.bat @@ -30,6 +30,19 @@ echo !BIN!\lz4 -d -f tmp-empty.lz4 tmp-empty.out !BIN!\lz4 -d -f tmp-empty.lz4 tmp-empty.out || goto :ERROR del tmp-empty tmp-empty.lz4 tmp-empty.out +@rem multithreaded compression (liblz4 worker threads) : same output whatever the nb of threads +echo !BIN!\datagen -g13M ^> tmp-mt + !BIN!\datagen -g13M > tmp-mt || goto :ERROR +for %%O in ("-1" "-1 -BD" "-9 -BD -BX") do ( + echo !BIN!\lz4 %%~O -T1 / -T4 tmp-mt + !BIN!\lz4 -q -f %%~O -T1 tmp-mt tmp-mt-T1.lz4 || goto :ERROR + !BIN!\lz4 -q -f %%~O -T4 tmp-mt tmp-mt-T4.lz4 || goto :ERROR + fc /b tmp-mt-T1.lz4 tmp-mt-T4.lz4 > nul || goto :ERROR + !BIN!\lz4 -q -d -f tmp-mt-T4.lz4 tmp-mt.out || goto :ERROR + fc /b tmp-mt tmp-mt.out > nul || goto :ERROR +) +del tmp-mt tmp-mt-T1.lz4 tmp-mt-T4.lz4 tmp-mt.out + echo fullbench !BIN!\fullbench.exe --no-prompt -i1 %TEST_FILES% || goto :ERROR diff --git a/build/cmake/CMakeLists.txt b/build/cmake/CMakeLists.txt index cee3df72..e848a313 100644 --- a/build/cmake/CMakeLists.txt +++ b/build/cmake/CMakeLists.txt @@ -280,6 +280,8 @@ if(LZ4_PROGRAMS_BUILT) foreach(prog ${LZ4_PROGRAMS_BUILT}) target_link_libraries(${prog} PRIVATE Threads::Threads) endforeach() + else() + message(WARNING "pthread not found : lz4 programs are built without multithreading support") endif() endif() endif() diff --git a/build/meson/meson/lib/meson.build b/build/meson/meson/lib/meson.build index 136d7baf..dd176523 100644 --- a/build/meson/meson/lib/meson.build +++ b/build/meson/meson/lib/meson.build @@ -133,7 +133,8 @@ if internal_lib endif # liblz4 is the shared library when both are built : -# the threads dependency of a multi-threaded static library must be listed explicitly (pkg-config --static) +# the threads dependency of a multi-threaded static library must then be listed explicitly (pkg-config --static). +# A static-only build already lists it (Libs: -pthread). pkgconfig.generate( liblz4, name: 'lz4', diff --git a/lib/lz4frame.c b/lib/lz4frame.c index c666816e..36278e32 100644 --- a/lib/lz4frame.c +++ b/lib/lz4frame.c @@ -1753,23 +1753,25 @@ static BYTE* LZ4F_mt_acquireInBuff(LZ4F_mtCtx_t* mt, LZ4F_CustomMem cmem) } } -/* emits oldest submitted job into @dst + @pos, waiting for it if needed (see LZ4F_pool_waitDone()) */ +/* emits oldest submitted job into @dst + @pos, waiting for it if needed (see LZ4F_pool_waitDone()). + * On error, the job remains submitted : callers invalidate the frame, discarding its jobs (FORWARD_IF_ERROR_MT) */ static size_t LZ4F_mt_emitOldest(LZ4F_mtCtx_t* mt, BYTE* dst, size_t dstCapacity, size_t pos, unsigned minQueue) { LZ4F_mtJob_t* const job = mt->jobs + mt->oldest; assert(mt->nbSubmitted > 0); assert(pos <= dstCapacity); LZ4F_pool_waitDone(mt->pool, &job->done, minQueue); - LZ4F_mt_releaseInBuff(mt, job); - mt->oldest = (mt->oldest + 1) % mt->nbJobSlots; - mt->nbSubmitted--; FORWARD_IF_ERROR(job->cSize); RETURN_ERROR_IF(job->cSize > dstCapacity - pos, dstCapacity_tooSmall); memcpy(dst + pos, job->outBuff, job->cSize); + LZ4F_mt_releaseInBuff(mt, job); + mt->oldest = (mt->oldest + 1) % mt->nbJobSlots; + mt->nbSubmitted--; return job->cSize; } -/* emits submitted jobs in order : all of them (@all), or only the already completed ones */ +/* emits submitted jobs in order : all of them (@all), or only the already completed ones. + * On error, callers invalidate the frame (FORWARD_IF_ERROR_MT) */ static size_t LZ4F_mt_emit(LZ4F_mtCtx_t* mt, BYTE* dst, size_t dstCapacity, size_t pos, int all) { size_t written = 0; @@ -1799,7 +1801,9 @@ static void LZ4F_mt_abort(LZ4F_mtCtx_t* mt) mt->uncTailSize = 0; } -/* a failed operation invalidates the frame, and discards its jobs */ +/* a failed operation invalidates the frame, and discards its jobs : + * next operations fail with compressionState_uninitialized, until LZ4F_compressBegin(). + * Idempotent : also employed to forward errors of functions which already invalidated the frame */ #define FORWARD_IF_ERROR_MT(r, cctx) do { \ size_t const err_code = (r); \ if (LZ4F_isError(err_code)) { \ @@ -2038,7 +2042,7 @@ static size_t LZ4F_compressUpdate_MT(LZ4F_cctx_t* cctx, DEBUGLOG(4, "LZ4F_compressUpdate_MT (srcSize=%u)", (unsigned)srcSize); RETURN_ERROR_IF(cctx->cStage != 1, compressionState_uninitialized); { size_t const r = LZ4F_mt_emitUncTail(cctx, dst, dstCapacity, pos); /* switch from uncompressed blocks */ - FORWARD_IF_ERROR(r); + FORWARD_IF_ERROR_MT(r, cctx); pos += r; } @@ -2097,7 +2101,7 @@ static size_t LZ4F_compressUpdate_MT(LZ4F_cctx_t* cctx, if (cctx->prefs.autoFlush) { size_t const r = LZ4F_mt_flush(cctx, dst, dstCapacity, pos); - FORWARD_IF_ERROR(r); + FORWARD_IF_ERROR_MT(r, cctx); return pos + r; } { size_t const r = LZ4F_mt_emit(mt, dst, dstCapacity, pos, 0); @@ -2123,7 +2127,7 @@ static size_t LZ4F_uncompressedUpdate_MT(LZ4F_cctx_t* cctx, RETURN_ERROR_IF(cctx->prefs.frameInfo.blockMode != LZ4F_blockIndependent, blockMode_invalid); if (mt->nbSubmitted || mt->filling) { /* switch from compressed blocks */ pos = LZ4F_mt_flush(cctx, dst, dstCapacity, 0); - FORWARD_IF_ERROR(pos); + FORWARD_IF_ERROR_MT(pos, cctx); } mt->stableEnd = NULL; /* next input doesn't follow previous stable input anymore */ if (cctx->prefs.frameInfo.contentChecksumFlag == LZ4F_contentChecksumEnabled) @@ -2139,7 +2143,7 @@ static size_t LZ4F_uncompressedUpdate_MT(LZ4F_cctx_t* cctx, srcSize -= toCopy; if (mt->uncTailSize == blockSize) { size_t const r = LZ4F_mt_emitUncTail(cctx, dst, dstCapacity, pos); - FORWARD_IF_ERROR(r); + FORWARD_IF_ERROR_MT(r, cctx); pos += r; } } diff --git a/programs/lz4.1 b/programs/lz4.1 index f4081148..7c24f6bf 100644 --- a/programs/lz4.1 +++ b/programs/lz4.1 @@ -210,7 +210,7 @@ Memory\-map regular files to compress, so that compression threads read them in Note : a file truncated while being compressed, or an I/O error while reading it, then interrupts \fBlz4\fR with error 53 (instead of compressing the content read up to that point) ; use \fB\-\-no\-mmap\fR when sources may be truncated or rewritten concurrently\. . .br -Note : a source file modified while being compressed (size or times changed) is detected after its compression, and is kept, even with \fB\-\-rm\fR\. When the output is valid, \fBlz4\fR only issues a warning : with \fB\-\-no\-mmap\fR, input is read once, so the output holds the content read (ex: up to the end of a log file growing meanwhile) ; memory\-mapped input is read several times, so its content is verified unchanged, using the frame checksum (ex: log file growing meanwhile : the output holds the file as it was when opened)\. Otherwise (content rewritten, \fB\-\-no\-frame\-crc\fR, or file shrunk), the output could be inconsistent : \fBlz4\fR reports error 54 and removes it\. +Note : a source file modified while being compressed (size or times changed), or renamed or replaced (ex: log rotation), is detected after its compression, and is kept, even with \fB\-\-rm\fR\. When the output is valid, \fBlz4\fR only issues a warning : with \fB\-\-no\-mmap\fR, input is read once, so the output holds the content read (ex: up to the end of a log file growing meanwhile) ; memory\-mapped input is read several times, so its content is verified unchanged, using the frame checksum (ex: log file growing meanwhile : the output holds the file as it was when opened)\. Otherwise (content rewritten, \fB\-\-no\-frame\-crc\fR, or file shrunk), the output could be inconsistent : \fBlz4\fR reports error 54 and removes it\. . .TP \fB\-l\fR @@ -243,7 +243,7 @@ Preserve source files (default behavior) . .TP \fB\-\-rm\fR -Delete source files on successful compression or decompression\. A source file modified during its compression is kept (see \fB\-\-[no\-]mmap\fR)\. +Delete source files on successful compression or decompression\. A source file modified, renamed or replaced during its compression is kept (see \fB\-\-[no\-]mmap\fR)\. . .TP \fB\-\-\fR diff --git a/programs/lz4.1.md b/programs/lz4.1.md index 11cd6bdc..9875a7bb 100644 --- a/programs/lz4.1.md +++ b/programs/lz4.1.md @@ -226,8 +226,8 @@ only the latest one will be applied. Note : a file truncated while being compressed, or an I/O error while reading it, then interrupts `lz4` with error 53 (instead of compressing the content read up to that point) ; use `--no-mmap` when sources may be truncated or rewritten concurrently.
- Note : a source file modified while being compressed (size or times changed) is detected after its compression, - and is kept, even with `--rm`. When the output is valid, `lz4` only issues a warning : + Note : a source file modified while being compressed (size or times changed), or renamed or replaced (ex: log rotation), + is detected after its compression, and is kept, even with `--rm`. When the output is valid, `lz4` only issues a warning : with `--no-mmap`, input is read once, so the output holds the content read (ex: up to the end of a log file growing meanwhile) ; memory-mapped input is read several times, so its content is verified unchanged, using the frame checksum (ex: log file growing meanwhile : the output holds the file as it was when opened). @@ -258,7 +258,7 @@ only the latest one will be applied. * `--rm` : Delete source files on successful compression or decompression. - A source file modified during its compression is kept (see `--[no-]mmap`). + A source file modified, renamed or replaced during its compression is kept (see `--[no-]mmap`). * `--` : Treat all subsequent arguments as files diff --git a/programs/lz4io.c b/programs/lz4io.c index 23b10ac5..02c1d517 100644 --- a/programs/lz4io.c +++ b/programs/lz4io.c @@ -1420,6 +1420,23 @@ static int LZ4IO_srcModified(const LZ4IO_srcState_t* before, FILE* srcFile) return 0; } +/* @return : 1 if @srcFileName doesn't designate the compressed file anymore (ex: log rotation), + * so that --rm would remove another file */ +static int LZ4IO_srcReplaced(const LZ4IO_srcState_t* s, const char* srcFileName) +{ +#if defined(_WIN32) + /* no inode numbers, and st_dev differs between _stat() and _fstat() : + * but an open file can't be renamed nor removed there */ + (void)s; (void)srcFileName; + return 0; +#else + stat_t st; + if (!s->valid || LZ4IO_isStdin(srcFileName)) return 0; + if (!UTIL_getFileStat(srcFileName, &st)) return 1; /* removed, or not a regular file anymore */ + return (st.st_dev != s->st.st_dev) || (st.st_ino != s->st.st_ino); +#endif +} + /* @return : 1 if the mapped content still matches the frame's content checksum (stored in its last 4 bytes), * 0 if it differs, or can't be verified (no checksum, or file shrunk : mapped content can't be read) */ static int LZ4IO_mappedContentUnchanged(const LZ4IO_cInput_t* in, FILE* srcFile, @@ -1466,7 +1483,7 @@ LZ4IO_compressFilename_extRess(unsigned long long* inStreamSize, /* Init */ FILE* const srcFile = LZ4IO_openSrcFile(srcFileName); LZ4IO_srcState_t srcState; - int srcModified; + int srcModified, srcReplaced; const void* frameEnd = NULL; /* end of frame, within dstBuffer (content checksum) */ if (srcFile == NULL) return 1; srcState = LZ4IO_getSrcState(srcFile); @@ -1570,20 +1587,23 @@ LZ4IO_compressFilename_extRess(unsigned long long* inStreamSize, /* Release file handlers */ srcModified = LZ4IO_srcModified(&srcState, srcFile); + srcReplaced = LZ4IO_srcReplaced(&srcState, srcFileName); { int const inconsistent = srcModified && (input.map != NULL) && !LZ4IO_mappedContentUnchanged(&input, srcFile, prefs.frameInfo.contentChecksumFlag == LZ4F_contentChecksumEnabled, frameEnd); LZ4IO_unmapInput(&input); fclose (srcFile); if (!LZ4IO_isStdout(dstFileName)) fclose(dstFile); /* do not close stdout */ - if (srcModified) { + if (srcModified || srcReplaced) { /* source is preserved, even with --rm */ if (inconsistent) { /* inconsistent output is removed */ if (!LZ4IO_isStdout(dstFileName) && !LZ4IO_isDevNull(dstFileName)) (void)remove(dstFileName); END_PROCESS(54, "%s : file modified during compression (memory-mapped) : output removed ; see --no-mmap", srcFileName); } - DISPLAYLEVEL(2, "\rWarning : %s : file modified during compression : output holds the content read%s \n", - srcFileName, io_prefs->removeSrcFile ? " ; source kept" : ""); + /* --rm not honored : also displayed in modes lowering notifications (stdout, multiple files) */ + DISPLAYLEVEL((io_prefs->removeSrcFile ? 1 : 2), "\rWarning : %s : file %s during compression : output holds the content read%s \n", + srcFileName, srcReplaced ? "renamed or replaced" : "modified", + io_prefs->removeSrcFile ? " ; source kept" : ""); } } /* Copy owner, file permissions and modification time */ @@ -1591,11 +1611,12 @@ LZ4IO_compressFilename_extRess(unsigned long long* inStreamSize, if (!LZ4IO_isStdin(srcFileName) && !LZ4IO_isStdout(dstFileName) && !LZ4IO_isDevNull(dstFileName) + && !srcReplaced && UTIL_getFileStat(srcFileName, &statbuf)) { UTIL_setFileStat(dstFileName, &statbuf); } } - if (io_prefs->removeSrcFile && !srcModified) { /* remove source file : --rm */ + if (io_prefs->removeSrcFile && !srcModified && !srcReplaced) { /* remove source file : --rm */ if (remove(srcFileName)) END_PROCESS(50, "Remove error : %s: %s", srcFileName, strerror(errno)); } diff --git a/programs/lz4io.h b/programs/lz4io.h index 7d030c9c..a0a4c650 100644 --- a/programs/lz4io.h +++ b/programs/lz4io.h @@ -79,12 +79,12 @@ int LZ4IO_decompressMultipleFilenames(const char** inFileNamesTable, int ifntSiz * and no more than the nb of jobs of each file (when its size is known) ; * >= 1 : explicit (within memory budget when it's set, or on 32-bit systems) */ int LZ4IO_setNbWorkers(LZ4IO_prefs_t* const prefs, int nbWorkers); +int LZ4IO_defaultNbWorkers(void); /* memory budget of multi-threaded compression, in bytes : limits the automatic nb of workers, * and, when set (> 0), explicit ones too ; 0 : default budget (LZ4IO_defaultMemoryBudget()) */ unsigned long long LZ4IO_setMemoryBudget(LZ4IO_prefs_t* const prefs, unsigned long long budget); unsigned long long LZ4IO_defaultMemoryBudget(void); /* 4 GB, 1 GB on 32-bit systems */ -int LZ4IO_defaultNbWorkers(void); int LZ4IO_setDictionaryFilename(LZ4IO_prefs_t* const prefs, const char* dictionaryFilename); @@ -127,7 +127,8 @@ int LZ4IO_setSparseFile(LZ4IO_prefs_t* const prefs, int enable); /* Default setting : 1 == regular files are memory-mapped for compression, when supported. * Note : a file truncated during its compression, or unreadable (I/O error), raises SIGBUS : * a handler, installed while the file is mapped, reports it and terminates the process (error 53). - * A file modified during its compression is also detected : input read once only triggers a warning ; + * A file modified during its compression is also detected (kept with --rm, as a file renamed or replaced) : + * input read once only triggers a warning ; * memory-mapped input is read several times : when its content can't be verified unchanged * (using the frame checksum, ex: file only grown), the output is removed (error 54). */ int LZ4IO_setMmap(LZ4IO_prefs_t* const prefs, int enable); diff --git a/tests/Makefile b/tests/Makefile index 385271da..d4f0f0b4 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -276,6 +276,10 @@ test-lz4-blocksize: lz4 datagen checkFrame @echo "\n ---- test lz4 with custom block sizes ----" ./test-lz4-blocksize.sh +test-lz4-multithread: lz4 datagen + @echo "\n ---- test lz4 multithreaded compression, memory budget, sources modified meanwhile ----" + ./test-lz4-multithread.sh + test-lz4hc-hugefile: lz4 datagen @echo "\n ---- test HC compression/decompression of huge files ----" ./test-lz4hc-hugefile.sh @@ -297,7 +301,7 @@ test-lz4-opt-parser: lz4 datagen test-lz4-essentials : lz4 datagen test-datagen-invalid-input test-lz4-basic test-lz4-multiple \ test-lz4-multiple-legacy test-lz4-frame-concatenation \ test-lz4-testmode test-lz4-contentSize test-lz4-dict \ - test-lz4-blocksize test-lz4-mozlz4 + test-lz4-blocksize test-lz4-mozlz4 test-lz4-multithread test-lz4: lz4 datagen test-lz4-essentials test-lz4-opt-parser \ test-lz4-sparse test-lz4-hugefile test-lz4-dict \ diff --git a/tests/frameMT.c b/tests/frameMT.c index 31838430..20b84299 100644 --- a/tests/frameMT.c +++ b/tests/frameMT.c @@ -35,6 +35,7 @@ * abandoned frames (jobs in flight) are handled * - with independent blocks, output is identical to nbWorkers == 0 * - serial processing (dstCapacity too small for parallel jobs) produces the same output + * - destination too small : the frame is invalidated (no truncated frame), and the context remains usable * - all allocations happen in the calling thread (custom allocator is not thread-safe : see TSAN) * - prints a digest of produced frames, to compare builds with and without LZ4F_MULTITHREAD * Bench mode : frameMT -b FILE [-l#] [-B#] [-D] [-c] [-T#] @@ -697,11 +698,105 @@ static int validateAllocFailures(void) return 0; } +/* begin, update (whole input), flush, end : from operation @smallFrom onwards (0:update, 1:flush, 2:end ; 3:never), + * each operation only receives @capacity, instead of all remaining capacity. @smallFrom == 2 : no flush. + * @return : frame size, or the first error, *failedOp being then the failed operation */ +static size_t streamOps(LZ4F_cctx* cctx, const LZ4F_preferences_t* prefs, + char* dst, size_t dstCapacity, const char* src, size_t srcSize, + int smallFrom, size_t capacity, int* failedOp) +{ + size_t pos = LZ4F_compressBegin(cctx, dst, dstCapacity, prefs); + int op; + *failedOp = -1; + if (LZ4F_isError(pos)) return pos; + for (op = 0; op < 3; op++) { + size_t const c = (op >= smallFrom) ? capacity : dstCapacity - pos; + size_t r; + if ((op == 1) && (smallFrom == 2)) continue; + if (op == 0) r = LZ4F_compressUpdate(cctx, dst + pos, c, src, srcSize, NULL); + else if (op == 1) r = LZ4F_flush(cctx, dst + pos, c, NULL); + else r = LZ4F_compressEnd(cctx, dst + pos, c, NULL); + if (LZ4F_isError(r)) { *failedOp = op; return r; } + pos += r; + } + return pos; +} + +/* Destination too small : the operation fails with dstCapacity_tooSmall, and invalidates the frame : + * next operations fail with compressionState_uninitialized (in particular, no end mark : no truncated frame), + * until LZ4F_compressBegin() starts a new frame, identical to the reference. Same contract as nbWorkers == 0. + * Without autoFlush, the failing operation depends on scheduling : the first one given too little capacity, or a later one. */ +static int validateDstTooSmall(void) +{ + static const unsigned nbWorkersList[] = { 0, 1, 3 }; /* 1 : job slots fill up within LZ4F_compressUpdate() */ + size_t const srcSize = 13 MB + 50 KB; /* incomplete last job, and last block */ + size_t const smallCapacity = 1 KB; + size_t const dstCapacity = 2 * LZ4F_compressFrameBound(srcSize, NULL); + char* const src = (char*)malloc(srcSize); + char* const ref = (char*)malloc(dstCapacity); + char* const out = (char*)malloc(dstCapacity); + LZ4F_CustomMem const countingMem = { countingAlloc, NULL, countingFree, NULL }; + unsigned nbConfigs = 0, nbFailures = 0; + size_t wi; + int linked, autoFlush, smallFrom; + + g_step = "dstCapacity too small"; + CHECK(src && ref && out, "allocation failed"); + RDG_genBuffer(src, srcSize, 0.5, 0.0, 19); + for (wi = 0; wi < sizeof(nbWorkersList)/sizeof(nbWorkersList[0]); wi++) + for (linked = 0; linked <= 1; linked++) + for (autoFlush = 0; autoFlush <= 1; autoFlush++) + for (smallFrom = 0; smallFrom <= 2; smallFrom++) { + LZ4F_preferences_t prefs = LZ4F_INIT_PREFERENCES; + int const mustFail = !autoFlush || (smallFrom == 0); /* otherwise, only the end mark remains to be written */ + LZ4F_cctx* cctx; + size_t refSize, r; + int failedOp; + if (g_quick && (smallFrom == 1)) continue; + cctx = LZ4F_createCompressionContext_advanced(countingMem, LZ4F_VERSION); + CHECK(cctx != NULL, "allocation failed"); + prefs.compressionLevel = 1; + prefs.nbWorkers = nbWorkersList[wi]; + prefs.autoFlush = (unsigned)autoFlush; + prefs.frameInfo.blockMode = linked ? LZ4F_blockLinked : LZ4F_blockIndependent; + prefs.frameInfo.contentChecksumFlag = LZ4F_contentChecksumEnabled; + refSize = streamOps(cctx, &prefs, ref, dstCapacity, src, srcSize, 3, 0, &failedOp); + + CHECK_LZ4F(refSize); + + r = streamOps(cctx, &prefs, out, dstCapacity, src, srcSize, smallFrom, smallCapacity, &failedOp); + if (!LZ4F_isError(r)) { + CHECK(!mustFail, "%u workers, smallFrom=%i : should have failed", prefs.nbWorkers, smallFrom); + CHECK(r == refSize && !memcmp(out, ref, r), "%u workers : output differs", prefs.nbWorkers); + } else { + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_dstCapacity_tooSmall, + "%u workers : unexpected error %s", prefs.nbWorkers, LZ4F_getErrorName(r)); + CHECK(failedOp >= smallFrom, "%u workers : failed with enough capacity", prefs.nbWorkers); + nbFailures++; + /* frame invalidated */ + r = LZ4F_compressUpdate(cctx, out, dstCapacity, src, 1 KB, NULL); + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_compressionState_uninitialized, + "%u workers : update after failure : %s", prefs.nbWorkers, LZ4F_getErrorName(r)); + r = LZ4F_flush(cctx, out, dstCapacity, NULL); + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_compressionState_uninitialized, + "%u workers : flush after failure : %s", prefs.nbWorkers, LZ4F_getErrorName(r)); + r = LZ4F_compressEnd(cctx, out, dstCapacity, NULL); + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_compressionState_uninitialized, + "%u workers : end after failure : %s", prefs.nbWorkers, LZ4F_getErrorName(r)); + } + /* context remains usable */ + r = streamOps(cctx, &prefs, out, dstCapacity, src, srcSize, 3, 0, &failedOp); + CHECK(r == refSize && !memcmp(out, ref, r), "%u workers : next frame differs", prefs.nbWorkers); + LZ4F_freeCompressionContext(cctx); + CHECK(g_nbAllocs == g_nbFrees, "memory leak (%u allocations, %u frees)", (unsigned)g_nbAllocs, (unsigned)g_nbFrees); + nbConfigs++; + } + DISPLAY("dstCapacity too small : %u configurations validated, %u failures \n", nbConfigs, nbFailures); + free(src); free(ref); free(out); + return 0; +} + -/*-************************************ -* File API (lz4file.h) -**************************************/ -/* LZ4F_write() streams through LZ4F_compressUpdate() : same frame as one-shot compression */ /* A failure while creating worker threads (ex: transient allocation failure) is absorbed : * jobs are processed serially, producing the same output. Worker threads are created again for next frames. */ static int validatePoolRetry(void) @@ -759,6 +854,11 @@ static int validatePoolRetry(void) return 0; } + +/*-************************************ +* File API (lz4file.h) +**************************************/ +/* LZ4F_write() streams through LZ4F_compressUpdate() : same frame as one-shot compression */ static int validateFileAPI(void) { size_t const srcSize = 9 MB + 5; @@ -1027,6 +1127,7 @@ int main(int argc, const char** argv) validateStreaming(&digest); validateFlushContract(); validateAllocFailures(); + validateDstTooSmall(); validatePoolRetry(); validateFileAPI(); printf("digest : %016llx \n", (unsigned long long)XXH64_digest(&digest)); diff --git a/tests/test-lz4-multithread.sh b/tests/test-lz4-multithread.sh new file mode 100755 index 00000000..acc72ec8 --- /dev/null +++ b/tests/test-lz4-multithread.sh @@ -0,0 +1,128 @@ +#!/bin/sh + +FPREFIX="tmp-tmt" + +set -e + +remove () { + rm -rf $FPREFIX* +} + +trap remove EXIT + +set -x + +datagen -g13M > $FPREFIX-src # several jobs, last one incomplete + +# standard block sizes : frames are cut into jobs (custom block sizes : see test-lz4-blocksize.sh) +# same output whatever the nb of threads, memory-mapped input or not, file or stdin, memory budget +for opts in "-1" "-1 -BD" "-9 -BD -BX" "-3 -B4 -BD" "-1 -B5" "--fast=3 -BD --no-frame-crc" +do + lz4 -f -T1 $opts $FPREFIX-src $FPREFIX-T1.lz4 + lz4 -f -T4 $opts $FPREFIX-src $FPREFIX.lz4 + cmp $FPREFIX-T1.lz4 $FPREFIX.lz4 + lz4 -f -T0 $opts $FPREFIX-src $FPREFIX.lz4 + cmp $FPREFIX-T1.lz4 $FPREFIX.lz4 + lz4 -f -T3 --no-mmap $opts $FPREFIX-src $FPREFIX.lz4 + cmp $FPREFIX-T1.lz4 $FPREFIX.lz4 + lz4 -T2 $opts < $FPREFIX-src > $FPREFIX.lz4 + cmp $FPREFIX-T1.lz4 $FPREFIX.lz4 + lz4 -f --memory=50M $opts $FPREFIX-src $FPREFIX.lz4 + cmp $FPREFIX-T1.lz4 $FPREFIX.lz4 + lz4 -dc $FPREFIX-T1.lz4 | cmp - $FPREFIX-src +done + +# memory budget : --memory=# and LZ4_MEMORY limit the nb of threads, not the output +lz4 -f --memory=abc $FPREFIX-src $FPREFIX.lz4 && exit 1 # must fail +lz4 -f --memory=0 $FPREFIX-src $FPREFIX.lz4 && exit 1 # must fail +lz4 -f --memory=18446744073709551616 $FPREFIX-src $FPREFIX.lz4 && exit 1 # must fail (overflow) +lz4 -f --memory=17179869184G $FPREFIX-src $FPREFIX.lz4 && exit 1 # must fail (overflow) +lz4 -f -1 $FPREFIX-src $FPREFIX-ref.lz4 +MT=0; if lz4 -V 2>&1 | grep -q "multithread"; then MT=1; fi # otherwise, -T# and memory budget are ignored +lz4 -v -f -T64 --memory=50M $FPREFIX-src $FPREFIX.lz4 2> $FPREFIX.err +cmp $FPREFIX-ref.lz4 $FPREFIX.lz4 +if [ $MT = 1 ]; then grep "Memory budget (50 MB)" $FPREFIX.err; fi +LZ4_MEMORY=50MB lz4 -v -f -T64 $FPREFIX-src $FPREFIX.lz4 2> $FPREFIX.err +if [ $MT = 1 ]; then grep "Memory budget (50 MB)" $FPREFIX.err; fi +LZ4_MEMORY=2GiB lz4 -v -f -T64 --memory 50M $FPREFIX-src $FPREFIX.lz4 2> $FPREFIX.err # command line wins +if [ $MT = 1 ]; then grep "Memory budget (50 MB)" $FPREFIX.err; fi +LZ4_MEMORY=abc lz4 -f $FPREFIX-src $FPREFIX.lz4 2>&1 | grep "LZ4_MEMORY=abc" # ignored, with a warning +cmp $FPREFIX-ref.lz4 $FPREFIX.lz4 +lz4 -f --memory=1M $FPREFIX-src $FPREFIX.lz4 2> $FPREFIX.err # warning only +cmp $FPREFIX-ref.lz4 $FPREFIX.lz4 +if [ $MT = 1 ]; then grep "too small" $FPREFIX.err; fi + + +# Source modified during its compression (see --[no-]mmap) : +# output goes into a fifo, read only once the source is modified : +# lz4 is then blocked writing its first output, after reading its first 4 MB chunk. +# -T1 : 3 job slots, so lz4 is blocked before reaching the end of the source. +case "$(uname)" in MINGW*|MSYS*|CYGWIN*) exit 0 ;; esac +command -v mkfifo > /dev/null || exit 0 +MMAP=0; if lz4 -V 2>&1 | grep -q "64-bit"; then MMAP=1; fi # memory-mapped input : POSIX 64-bit systems + +rewriteStart() { printf 'modified' | dd of=$FPREFIX-mod bs=8 count=1 conv=notrunc 2> /dev/null; } +appendData() { echo "appended" >> $FPREFIX-mod; } +replaceSrc() { mv $FPREFIX-mod $FPREFIX-mod.old; echo "new file" > $FPREFIX-mod; } + +# compressModified OPTIONS MODIFICATION : compresses $FPREFIX-mod into $FPREFIX.out, sets rc +compressModified() { + rm -f $FPREFIX-mod* $FPREFIX.fifo $FPREFIX.out + cp $FPREFIX-src $FPREFIX-mod + touch -t 200001010000 $FPREFIX-mod # modification detected even with 1-second timestamps + mkfifo $FPREFIX.fifo + lz4 -f -T1 $1 $FPREFIX-mod $FPREFIX.fifo 2> $FPREFIX.err & + pid=$! + { dd bs=4 count=1 2> /dev/null; $2 < /dev/null; cat; } < $FPREFIX.fifo > $FPREFIX.out + rc=0; wait $pid || rc=$? + cat $FPREFIX.err +} + +# content already read is rewritten : memory-mapped input, read again, would be inconsistent (error 54) ; +# except on macOS, where a private mapping keeps the content as opened : then the output is valid +compressModified "" rewriteStart +if [ $MMAP = 1 ] && [ "$(uname)" != Darwin ]; then + test $rc = 54 + grep -q "output removed" $FPREFIX.err + test ! -e $FPREFIX.fifo +else + test $rc = 0 + lz4 -dc $FPREFIX.out | cmp - $FPREFIX-src +fi +# input read once : the output holds the content read +compressModified "--no-mmap" rewriteStart +test $rc = 0 +grep -q "Warning" $FPREFIX.err +lz4 -dc $FPREFIX.out | cmp - $FPREFIX-src +# --rm not honored : warning also displayed when notifications are lowered (stdout) +rm -f $FPREFIX-mod $FPREFIX.fifo; cp $FPREFIX-src $FPREFIX-mod; touch -t 200001010000 $FPREFIX-mod; mkfifo $FPREFIX.fifo +lz4 -c -T1 --no-mmap --rm $FPREFIX-mod > $FPREFIX.fifo 2> $FPREFIX.err & +pid=$! +{ dd bs=4 count=1 2> /dev/null; rewriteStart < /dev/null; cat; } < $FPREFIX.fifo > $FPREFIX.out +wait $pid +grep -q "source kept" $FPREFIX.err +test -f $FPREFIX-mod + +# growing file : memory-mapped input holds the file as it was when opened, verified with the frame checksum ; +# input read once also holds the appended content (unless end of file was already met : libc dependent) ; --rm keeps the source +compressModified "--rm" appendData +test $rc = 0 +grep -q "source kept" $FPREFIX.err +test -f $FPREFIX-mod +if [ $MMAP = 1 ]; then lz4 -dc $FPREFIX.out | cmp - $FPREFIX-src; fi +compressModified "--rm --no-mmap" appendData +test $rc = 0 +test -f $FPREFIX-mod +lz4 -dc $FPREFIX.out > $FPREFIX.dec +cmp $FPREFIX.dec $FPREFIX-mod || cmp $FPREFIX.dec $FPREFIX-src +# without content checksum, memory-mapped input can't be verified +compressModified "--no-frame-crc" appendData +if [ $MMAP = 1 ]; then test $rc = 54; else test $rc = 0; fi + +# source renamed and replaced (ex: log rotation) : --rm must not remove the new file +compressModified "--rm" replaceSrc +test $rc = 0 +grep -q "renamed or replaced" $FPREFIX.err +test "$(cat $FPREFIX-mod)" = "new file" +test -f $FPREFIX-mod.old +lz4 -dc $FPREFIX.out | cmp - $FPREFIX-src