From f9524cf366af289d2c24e673360c8b05f70b6a83 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 13:48:41 -0700 Subject: [PATCH 01/13] added --memtest to fuzzer --- tests/fuzzer.c | 193 +++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 179 insertions(+), 14 deletions(-) diff --git a/tests/fuzzer.c b/tests/fuzzer.c index c9461ee90..6586fc12e 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -51,14 +51,14 @@ static const U32 nbTestsDefault = 30000; /*-************************************ * Display Macros **************************************/ -#define DISPLAY(...) fprintf(stderr, __VA_ARGS__) +#define DISPLAY(...) fprintf(stdout, __VA_ARGS__) #define DISPLAYLEVEL(l, ...) if (g_displayLevel>=l) { DISPLAY(__VA_ARGS__); } static U32 g_displayLevel = 2; #define DISPLAYUPDATE(l, ...) if (g_displayLevel>=l) { \ if ((FUZ_clockSpan(g_displayClock) > g_refreshRate) || (g_displayLevel>=4)) \ { g_displayClock = clock(); DISPLAY(__VA_ARGS__); \ - if (g_displayLevel>=4) fflush(stderr); } } + if (g_displayLevel>=4) fflush(stdout); } } static const clock_t g_refreshRate = CLOCKS_PER_SEC / 6; static clock_t g_displayClock = 0; @@ -97,7 +97,161 @@ static unsigned FUZ_highbit32(U32 v32) /*============================================= -* Basic Unit tests +* Memory Tests +=============================================*/ +#if defined(__APPLE__) && defined(__MACH__) + +#include /* malloc_size */ + +typedef struct { + unsigned long long totalMalloc; + size_t peakMalloc; + unsigned nbMalloc; + unsigned nbFree; +} mallocCounter_t; + +static const mallocCounter_t INIT_MALLOC_COUNTER = { 0, 0, 0, 0 }; + +static void* FUZ_mallocDebug(void* counter, size_t size) +{ + mallocCounter_t* const mcPtr = (mallocCounter_t*)counter; + void* const ptr = malloc(size); + if (ptr==NULL) return NULL; + mcPtr->totalMalloc += size; + mcPtr->peakMalloc += size; + mcPtr->nbMalloc += 1; + return ptr; +} + +static void FUZ_freeDebug(void* counter, void* address) +{ + mallocCounter_t* const mcPtr = (mallocCounter_t*)counter; + free(address); + mcPtr->nbFree += 1; + mcPtr->peakMalloc -= malloc_size(address); /* OS-X specific */ +} + +static void FUZ_displayMallocStats(mallocCounter_t count) +{ + DISPLAYLEVEL(3, "peak:%u KB, nbMallocs:%u, total:%u KB \n", + (U32)(count.peakMalloc >> 10), + count.nbMalloc, + (U32)(count.totalMalloc >> 10)); +} + +static int FUZ_mallocTests(unsigned seed, double compressibility) +{ + size_t const inSize = 64 MB + 16 MB + 4 MB + 1 MB + 256 KB + 64 KB; /* 85.3 MB */ + size_t const outSize = ZSTD_compressBound(inSize); + void* const inBuffer = malloc(inSize); + void* const outBuffer = malloc(outSize); + + /* test only played in verbose mode, as they are long */ + if (g_displayLevel<3) return 0; + + /* Create compressible noise */ + if (!inBuffer || !outBuffer) { + DISPLAY("Not enough memory, aborting\n"); + exit(1); + } + RDG_genBuffer(inBuffer, inSize, compressibility, 0. /*auto*/, seed); + + /* simple compression tests */ + { int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compressCCtx level %i : ", compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } + + /* streaming compression tests */ + { int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + ZSTD_CCtx* const cstream = ZSTD_createCStream_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_initCStream(cstream, compressionLevel); + ZSTD_compressStream(cstream, &out, &in); + ZSTD_endStream(cstream, &out); + ZSTD_freeCStream(cstream); + DISPLAYLEVEL(3, "compressStream level %i : ", compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } + + /* advanced API test */ + { int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compress_generic,end level %i : ", compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } + + /* advanced MT API test */ + { int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); + ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, 2); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compress_generic,-T2,end level %i : ", compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } + + /* advanced MT streaming API test */ + { int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); + ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, 2); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compress_generic,-T2,continue level %i : ", compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } + + return 0; +} + +#else + +static int FUZ_mallocTests(unsigned seed, double compressibility) +{ + (void)seed; (void)compressibility; + return 0; +} + +#endif + +/*============================================= +* Unit tests =============================================*/ #define CHECK_V(var, fn) size_t const var = fn; if (ZSTD_isError(var)) goto _output_error @@ -108,7 +262,8 @@ static int basicUnitTests(U32 seed, double compressibility) { size_t const CNBuffSize = 5 MB; void* const CNBuffer = malloc(CNBuffSize); - void* const compressedBuffer = malloc(ZSTD_compressBound(CNBuffSize)); + size_t const compressedBufferSize = ZSTD_compressBound(CNBuffSize); + void* const compressedBuffer = malloc(compressedBufferSize); void* const decodedBuffer = malloc(CNBuffSize); ZSTD_DCtx* dctx = ZSTD_createDCtx(); int testResult = 0; @@ -123,6 +278,9 @@ static int basicUnitTests(U32 seed, double compressibility) } RDG_genBuffer(CNBuffer, CNBuffSize, compressibility, 0., seed); + /* memory tests */ + FUZ_mallocTests(seed, compressibility); + /* Basic tests */ DISPLAYLEVEL(4, "test%3i : ZSTD_getErrorName : ", testNb++); { const char* errorString = ZSTD_getErrorName(0); @@ -139,7 +297,7 @@ static int basicUnitTests(U32 seed, double compressibility) { ZSTD_CCtx* cctx = ZSTD_createCCtx(); if (cctx==NULL) goto _output_error; CHECKPLUS(r, ZSTD_compressCCtx(cctx, - compressedBuffer, ZSTD_compressBound(CNBuffSize), + compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, 1), cSize=r ); DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100); @@ -226,7 +384,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "test%3i : simple compression test with static CCtx : ", testNb++); CHECKPLUS(r, ZSTD_compressCCtx(staticCCtx, - compressedBuffer, ZSTD_compressBound(CNBuffSize), + compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, STATIC_CCTX_LEVEL), cSize=r ); DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", @@ -295,7 +453,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "test%3i : compress %u bytes with 2 threads : ", testNb++, (U32)CNBuffSize); CHECKPLUS(r, ZSTDMT_compressCCtx(mtctx, - compressedBuffer, ZSTD_compressBound(CNBuffSize), + compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, 1), cSize=r ); @@ -382,7 +540,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "test%3i : compress with flat dictionary : ", testNb++); cSize = 0; - CHECKPLUS(r, ZSTD_compressEnd(ctxOrig, compressedBuffer, ZSTD_compressBound(CNBuffSize), + CHECKPLUS(r, ZSTD_compressEnd(ctxOrig, compressedBuffer, compressedBufferSize, (const char*)CNBuffer + dictSize, CNBuffSize - dictSize), cSize += r); DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100); @@ -398,7 +556,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "test%3i : compress with duplicated context : ", testNb++); { size_t const cSizeOrig = cSize; cSize = 0; - CHECKPLUS(r, ZSTD_compressEnd(ctxDuplicated, compressedBuffer, ZSTD_compressBound(CNBuffSize), + CHECKPLUS(r, ZSTD_compressEnd(ctxDuplicated, compressedBuffer, compressedBufferSize, (const char*)CNBuffer + dictSize, CNBuffSize - dictSize), cSize += r); if (cSize != cSizeOrig) goto _output_error; /* should be identical ==> same size */ @@ -483,7 +641,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "OK : %u \n", dictID); DISPLAYLEVEL(4, "test%3i : compress with dictionary : ", testNb++); - cSize = ZSTD_compress_usingDict(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), + cSize = ZSTD_compress_usingDict(cctx, compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, dictBuffer, dictSize, 4); if (ZSTD_isError(cSize)) goto _output_error; @@ -521,7 +679,7 @@ static int basicUnitTests(U32 seed, double compressibility) 1 /* byReference */, ZSTD_dm_auto, cParams, ZSTD_defaultCMem); DISPLAYLEVEL(4, "(size : %u) : ", (U32)ZSTD_sizeof_CDict(cdict)); - cSize = ZSTD_compress_usingCDict(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), + cSize = ZSTD_compress_usingCDict(cctx, compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, cdict); ZSTD_freeCDict(cdict); if (ZSTD_isError(cSize)) goto _output_error; @@ -556,7 +714,7 @@ static int basicUnitTests(U32 seed, double compressibility) goto _output_error; } cSize = ZSTD_compress_usingCDict(cctx, - compressedBuffer, ZSTD_compressBound(CNBuffSize), + compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, cdict); if (ZSTD_isError(cSize)) { DISPLAY("ZSTD_compress_usingCDict failed "); @@ -570,7 +728,7 @@ static int basicUnitTests(U32 seed, double compressibility) { ZSTD_frameParameters const fParams = { 0 /* frameSize */, 1 /* checksum */, 1 /* noDictID*/ }; ZSTD_compressionParameters const cParams = ZSTD_getCParams(1, CNBuffSize, dictSize); ZSTD_CDict* const cdict = ZSTD_createCDict_advanced(dictBuffer, dictSize, 1 /*byRef*/, ZSTD_dm_auto, cParams, ZSTD_defaultCMem); - cSize = ZSTD_compress_usingCDict_advanced(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), + cSize = ZSTD_compress_usingCDict_advanced(cctx, compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, cdict, fParams); ZSTD_freeCDict(cdict); if (ZSTD_isError(cSize)) goto _output_error; @@ -594,7 +752,7 @@ static int basicUnitTests(U32 seed, double compressibility) DISPLAYLEVEL(4, "test%3i : ZSTD_compress_advanced, no dictID : ", testNb++); { ZSTD_parameters p = ZSTD_getParams(3, CNBuffSize, dictSize); p.fParams.noDictIDFlag = 1; - cSize = ZSTD_compress_advanced(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), + cSize = ZSTD_compress_advanced(cctx, compressedBuffer, compressedBufferSize, CNBuffer, CNBuffSize, dictBuffer, dictSize, p); if (ZSTD_isError(cSize)) goto _output_error; @@ -1245,6 +1403,7 @@ int main(int argc, const char** argv) U32 mainPause = 0; U32 maxDuration = 0; int bigTests = 1; + U32 memTestsOnly = 0; const char* const programName = argv[0]; /* Check command line */ @@ -1255,6 +1414,7 @@ int main(int argc, const char** argv) /* Handle commands. Aggregated commands are allowed */ if (argument[0]=='-') { + if (!strcmp(argument, "--memtest")) { memTestsOnly=1; continue; } if (!strcmp(argument, "--no-big-tests")) { bigTests=0; continue; } argument++; @@ -1326,6 +1486,11 @@ int main(int argc, const char** argv) DISPLAY("Seed = %u\n", seed); if (proba!=FUZ_compressibility_default) DISPLAY("Compressibility : %u%%\n", proba); + if (memTestsOnly) { + g_displayLevel=3; + return FUZ_mallocTests(seed, ((double)proba) / 100); + } + if (nbTests < testNb) nbTests = testNb; if (testNb==0) From 88da8f181660890c369bbee4aafd6eaa8f6bb830 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 14:02:33 -0700 Subject: [PATCH 02/13] fix : propagate custom allocator to ZSTDMT though ZSTD_CCtx_setParameter() also : compile fuzzer with MT enabled --- lib/compress/zstd_compress.c | 2 +- tests/Makefile | 12 ++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 9300357f2..c17645de7 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -377,7 +377,7 @@ size_t ZSTD_CCtx_setParameter(ZSTD_CCtx* cctx, ZSTD_cParameter param, unsigned v return ERROR(compressionParameter_unsupported); ZSTDMT_freeCCtx(cctx->mtctx); cctx->nbThreads = 1; - cctx->mtctx = ZSTDMT_createCCtx(value); + cctx->mtctx = ZSTDMT_createCCtx_advanced(value, cctx->customMem); if (cctx->mtctx == NULL) return ERROR(memory_allocation); } cctx->nbThreads = value; diff --git a/tests/Makefile b/tests/Makefile index 345e0e8eb..4784136e1 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -79,7 +79,7 @@ all32: fullbench32 fuzzer32 zstreamtest32 allnothread: fullbench fuzzer paramgrill datagen decodecorpus -dll: fuzzer-dll zstreamtest-dll +dll: fuzzer-dll zstreamtest-dll zstd: $(MAKE) -C $(PRGDIR) $@ @@ -108,11 +108,11 @@ fullbench-dll: $(PRGDIR)/datagen.c fullbench.c $(MAKE) -C $(ZSTDDIR) libzstd $(CC) $(FLAGS) $^ -o $@$(EXT) -DZSTD_DLL_IMPORT=1 $(ZSTDDIR)/dll/libzstd.dll -fuzzer : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c - $(CC) $(FLAGS) $^ -o $@$(EXT) - -fuzzer32 : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c - $(CC) -m32 $(FLAGS) $^ -o $@$(EXT) +fuzzer : CPPFLAGS += $(MULTITHREAD_CPP) +fuzzer : LDFLAGS += $(MULTITHREAD_LD) +fuzzer32: CFLAGS += -m32 +fuzzer fuzzer32 : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c + $(CC) $(FLAGS) $^ -o $@$(EXT) fuzzer-dll : LDFLAGS+= -L$(ZSTDDIR) -lzstd fuzzer-dll : $(ZSTDDIR)/common/xxhash.c $(PRGDIR)/datagen.c fuzzer.c From ee3423d70950f2ec41ff884a6cdacb463118f935 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 14:09:16 -0700 Subject: [PATCH 03/13] extended fuzzer MT memory tests --- tests/fuzzer.c | 88 ++++++++++++++++++++++---------------------------- 1 file changed, 39 insertions(+), 49 deletions(-) diff --git a/tests/fuzzer.c b/tests/fuzzer.c index 6586fc12e..c05ed5aea 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -160,7 +160,7 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) { int compressionLevel; mallocCounter_t malcount = INIT_MALLOC_COUNTER; ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; - for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel); ZSTD_freeCCtx(cctx); @@ -173,7 +173,7 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) { int compressionLevel; mallocCounter_t malcount = INIT_MALLOC_COUNTER; ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; - for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { + for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { ZSTD_CCtx* const cstream = ZSTD_createCStream_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; @@ -186,56 +186,46 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) malcount = INIT_MALLOC_COUNTER; } } - /* advanced API test */ - { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; - for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { - ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); - ZSTD_outBuffer out = { outBuffer, outSize, 0 }; - ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); - ZSTD_freeCCtx(cctx); - DISPLAYLEVEL(3, "compress_generic,end level %i : ", compressionLevel); - FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; - } } - /* advanced MT API test */ - { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; - for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { - ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); - ZSTD_outBuffer out = { outBuffer, outSize, 0 }; - ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); - ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, 2); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); - ZSTD_freeCCtx(cctx); - DISPLAYLEVEL(3, "compress_generic,-T2,end level %i : ", compressionLevel); - FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; - } } + { U32 nbThreads; + for (nbThreads=1; nbThreads<=4; nbThreads++) { + int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); + ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compress_generic,-T%u,end level %i : ", + nbThreads, compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } } /* advanced MT streaming API test */ - { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; - for (compressionLevel=1; compressionLevel<=5; compressionLevel++) { - ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); - ZSTD_outBuffer out = { outBuffer, outSize, 0 }; - ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); - ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, 2); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); - ZSTD_freeCCtx(cctx); - DISPLAYLEVEL(3, "compress_generic,-T2,continue level %i : ", compressionLevel); - FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; - } } + { U32 nbThreads; + for (nbThreads=1; nbThreads<=4; nbThreads++) { + int compressionLevel; + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; + for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); + ZSTD_outBuffer out = { outBuffer, outSize, 0 }; + ZSTD_inBuffer in = { inBuffer, inSize, 0 }; + ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); + ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue); + ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + ZSTD_freeCCtx(cctx); + DISPLAYLEVEL(3, "compress_generic,-T%u,continue level %i : ", + nbThreads, compressionLevel); + FUZ_displayMallocStats(malcount); + malcount = INIT_MALLOC_COUNTER; + } } } return 0; } From 3510efb02dc0f5a3d90b31546d425963811b4c90 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 14:21:40 -0700 Subject: [PATCH 04/13] fix : custom allocator correctly propagated to child contexts --- lib/compress/zstdmt_compress.c | 2 +- tests/fuzzer.c | 20 ++++++++------------ 2 files changed, 9 insertions(+), 13 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index d5607adfd..bad2db9c2 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -215,7 +215,7 @@ static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* pool) pool->availCCtx--; return pool->cctx[pool->availCCtx]; } - return ZSTD_createCCtx(); /* note : can be NULL, when creation fails ! */ + return ZSTD_createCCtx_advanced(pool->cMem); /* note : can be NULL, when creation fails ! */ } static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx) diff --git a/tests/fuzzer.c b/tests/fuzzer.c index c05ed5aea..aa1ebd48f 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -158,22 +158,21 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) /* simple compression tests */ { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel); ZSTD_freeCCtx(cctx); DISPLAYLEVEL(3, "compressCCtx level %i : ", compressionLevel); FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; } } /* streaming compression tests */ { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; ZSTD_CCtx* const cstream = ZSTD_createCStream_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; @@ -183,16 +182,15 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) ZSTD_freeCStream(cstream); DISPLAYLEVEL(3, "compressStream level %i : ", compressionLevel); FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; } } /* advanced MT API test */ { U32 nbThreads; for (nbThreads=1; nbThreads<=4; nbThreads++) { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; @@ -203,16 +201,15 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) DISPLAYLEVEL(3, "compress_generic,-T%u,end level %i : ", nbThreads, compressionLevel); FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; } } } /* advanced MT streaming API test */ { U32 nbThreads; for (nbThreads=1; nbThreads<=4; nbThreads++) { int compressionLevel; - mallocCounter_t malcount = INIT_MALLOC_COUNTER; - ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; for (compressionLevel=1; compressionLevel<=6; compressionLevel++) { + mallocCounter_t malcount = INIT_MALLOC_COUNTER; + ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; @@ -224,7 +221,6 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) DISPLAYLEVEL(3, "compress_generic,-T%u,continue level %i : ", nbThreads, compressionLevel); FUZ_displayMallocStats(malcount); - malcount = INIT_MALLOC_COUNTER; } } } return 0; From 670b1fc547424897aa324736d47a1c79bf61e355 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 16:30:55 -0700 Subject: [PATCH 05/13] optimized memory usage for ZSTDMT_compress() Previously, each job would reserve a CCtx right before being posted. The CCtx would be "part of the job description", and only released when the job is completed (aka flushed). For ZSTDMT_compress(), which creates all jobs first and only join at the end, that meant one CCtx per job. The nb of jobs used to be == nb of threads, but since latest modification, which reduces the size of jobs in order to spread the load of difficult areas, it also increases the nb of jobs for large sources / small compression level. This resulted in many more CCtx being created. In this new version, CCtx are reserved within the worker thread. It guaranteea there cannot be more CCtx reserved than workers (<= nb threads). To do that, it required to make the CCtx Pool multi-threading-safe : it can now be called from multiple threads in parallel. --- lib/compress/zstdmt_compress.c | 97 ++++++++++++++++++++-------------- tests/fuzzer.c | 45 ++++++++++------ 2 files changed, 86 insertions(+), 56 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index bad2db9c2..a176c3eea 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -126,6 +126,7 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool) * assumption : invocation from main thread only ! */ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize) { + DEBUGLOG(2, "ZSTDMT_getBuffer"); if (pool->nbBuffers) { /* try to use an existing buffer */ buffer_t const buf = pool->bTable[--(pool->nbBuffers)]; size_t const availBufferSize = buf.size; @@ -160,21 +161,23 @@ static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* pool, buffer_t buf) /* ===== CCtx Pool ===== */ +/* a single cctxPool can be called from multiple threads in parallel */ + typedef struct { + pthread_mutex_t poolMutex; unsigned totalCCtx; unsigned availCCtx; ZSTD_customMem cMem; ZSTD_CCtx* cctx[1]; /* variable size */ } ZSTDMT_CCtxPool; -/* assumption : CCtxPool invocation only from main thread */ - /* note : all CCtx borrowed from the pool should be released back to the pool _before_ freeing the pool */ static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool) { unsigned u; for (u=0; utotalCCtx; u++) ZSTD_freeCCtx(pool->cctx[u]); /* note : compatible with free on NULL */ + pthread_mutex_destroy(&pool->poolMutex); ZSTD_free(pool, pool->cMem); } @@ -186,6 +189,7 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads, ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc( sizeof(ZSTDMT_CCtxPool) + (nbThreads-1)*sizeof(ZSTD_CCtx*), cMem); if (!cctxPool) return NULL; + pthread_mutex_init(&cctxPool->poolMutex, NULL); cctxPool->cMem = cMem; cctxPool->totalCCtx = nbThreads; cctxPool->availCCtx = 1; /* at least one cctx for single-thread mode */ @@ -198,34 +202,47 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads, /* only works during initialization phase, not during compression */ static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool) { - unsigned const nbThreads = cctxPool->totalCCtx; - size_t const poolSize = sizeof(*cctxPool) - + (nbThreads-1)*sizeof(ZSTD_CCtx*); - unsigned u; - size_t totalCCtxSize = 0; - for (u=0; ucctx[u]); - - return poolSize + totalCCtxSize; + pthread_mutex_lock(&cctxPool->poolMutex); + { unsigned const nbThreads = cctxPool->totalCCtx; + size_t const poolSize = sizeof(*cctxPool) + + (nbThreads-1)*sizeof(ZSTD_CCtx*); + unsigned u; + size_t totalCCtxSize = 0; + for (u=0; ucctx[u]); + } + pthread_mutex_unlock(&cctxPool->poolMutex); + return poolSize + totalCCtxSize; + } } -static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* pool) +static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool) { - if (pool->availCCtx) { - pool->availCCtx--; - return pool->cctx[pool->availCCtx]; - } - return ZSTD_createCCtx_advanced(pool->cMem); /* note : can be NULL, when creation fails ! */ + DEBUGLOG(5, "ZSTDMT_getCCtx"); + pthread_mutex_lock(&cctxPool->poolMutex); + if (cctxPool->availCCtx) { + cctxPool->availCCtx--; + { ZSTD_CCtx* const cctx = cctxPool->cctx[cctxPool->availCCtx]; + pthread_mutex_unlock(&cctxPool->poolMutex); + return cctx; + } } + pthread_mutex_unlock(&cctxPool->poolMutex); + DEBUGLOG(5, "create one more CCtx"); + return ZSTD_createCCtx_advanced(cctxPool->cMem); /* note : can be NULL, when creation fails ! */ } static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx) { if (cctx==NULL) return; /* compatibility with release on NULL */ + pthread_mutex_lock(&pool->poolMutex); if (pool->availCCtx < pool->totalCCtx) pool->cctx[pool->availCCtx++] = cctx; - else + else { /* pool overflow : should not happen, since totalCCtx==nbThreads */ + DEBUGLOG(5, "CCtx pool overflow : free cctx"); ZSTD_freeCCtx(cctx); + } + pthread_mutex_unlock(&pool->poolMutex); } @@ -237,7 +254,6 @@ typedef struct { } inBuff_t; typedef struct { - ZSTD_CCtx* cctx; buffer_t src; const void* srcStart; size_t srcSize; @@ -253,6 +269,7 @@ typedef struct { pthread_cond_t* jobCompleted_cond; ZSTD_parameters params; const ZSTD_CDict* cdict; + ZSTDMT_CCtxPool* cctxPool; unsigned long long fullFrameSize; } ZSTDMT_jobDescription; @@ -260,37 +277,45 @@ typedef struct { void ZSTDMT_compressChunk(void* jobDescription) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; + ZSTD_CCtx* cctx = ZSTDMT_getCCtx(job->cctxPool); const void* const src = (const char*)job->srcStart + job->dictSize; buffer_t const dstBuff = job->dstBuff; DEBUGLOG(5, "job (first:%u) (last:%u) : dictSize %u, srcSize %u", job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize); + + if (cctx==NULL) { + job->cSize = ERROR(memory_allocation); + goto _endJob; + } + if (job->cdict) { /* should only happen for first segment */ - size_t const initError = ZSTD_compressBegin_usingCDict_advanced(job->cctx, job->cdict, job->params.fParams, job->fullFrameSize); + size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize); DEBUGLOG(5, "using CDict"); if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; } } else { /* srcStart points at reloaded section */ if (!job->firstChunk) job->params.fParams.contentSizeFlag = 0; /* ensure no srcSize control */ - { size_t const dictModeError = ZSTD_setCCtxParameter(job->cctx, ZSTD_p_forceRawDict, 1); /* Force loading dictionary in "content-only" mode (no header analysis) */ - size_t const initError = ZSTD_compressBegin_advanced(job->cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize); + { size_t const dictModeError = ZSTD_setCCtxParameter(cctx, ZSTD_p_forceRawDict, 1); /* Force loading dictionary in "content-only" mode (no header analysis) */ + size_t const initError = ZSTD_compressBegin_advanced(cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize); if (ZSTD_isError(initError) || ZSTD_isError(dictModeError)) { job->cSize = initError; goto _endJob; } - ZSTD_setCCtxParameter(job->cctx, ZSTD_p_forceWindow, 1); + ZSTD_setCCtxParameter(cctx, ZSTD_p_forceWindow, 1); } } if (!job->firstChunk) { /* flush and overwrite frame header when it's not first segment */ - size_t const hSize = ZSTD_compressContinue(job->cctx, dstBuff.start, dstBuff.size, src, 0); + size_t const hSize = ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, 0); if (ZSTD_isError(hSize)) { job->cSize = hSize; goto _endJob; } - ZSTD_invalidateRepCodes(job->cctx); + ZSTD_invalidateRepCodes(cctx); } DEBUGLOG(5, "Compressing : "); DEBUG_PRINTHEX(4, job->srcStart, 12); job->cSize = (job->lastChunk) ? - ZSTD_compressEnd (job->cctx, dstBuff.start, dstBuff.size, src, job->srcSize) : - ZSTD_compressContinue(job->cctx, dstBuff.start, dstBuff.size, src, job->srcSize); + ZSTD_compressEnd (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) : + ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize); DEBUGLOG(5, "compressed %u bytes into %u bytes (first:%u) (last:%u)", (unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk); DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize)); _endJob: + ZSTDMT_releaseCCtx(job->cctxPool, cctx); PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex); job->jobCompleted = 1; job->jobScanned = 0; @@ -390,8 +415,6 @@ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) mtctx->jobs[jobID].dstBuff = g_nullBuffer; ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[jobID].src); mtctx->jobs[jobID].src = g_nullBuffer; - ZSTDMT_releaseCCtx(mtctx->cctxPool, mtctx->jobs[jobID].cctx); - mtctx->jobs[jobID].cctx = NULL; } memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription)); ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->inBuff.buffer); @@ -497,10 +520,9 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, size_t const dstBufferCapacity = ZSTD_compressBound(chunkSize); buffer_t const dstAsBuffer = { (char*)dst + dstBufferPos, dstBufferCapacity }; buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : ZSTDMT_getBuffer(mtctx->buffPool, dstBufferCapacity); - ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(mtctx->cctxPool); size_t dictSize = u ? overlapSize : 0; - if ((cctx==NULL) || (dstBuffer.start==NULL)) { + if (dstBuffer.start==NULL) { mtctx->jobs[u].cSize = ERROR(memory_allocation); /* job result */ mtctx->jobs[u].jobCompleted = 1; nbChunks = u+1; /* only wait and free u jobs, instead of initially expected nbChunks ones */ @@ -516,7 +538,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, /* do not calculate checksum within sections, but write it in header for first section */ if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0; mtctx->jobs[u].dstBuff = dstBuffer; - mtctx->jobs[u].cctx = cctx; + mtctx->jobs[u].cctxPool = mtctx->cctxPool; mtctx->jobs[u].firstChunk = (u==0); mtctx->jobs[u].lastChunk = (u==nbChunks-1); mtctx->jobs[u].jobCompleted = 0; @@ -545,8 +567,6 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, pthread_mutex_unlock(&mtctx->jobCompleted_mutex); DEBUGLOG(5, "ready to write chunk %u ", chunkID); - ZSTDMT_releaseCCtx(mtctx->cctxPool, mtctx->jobs[chunkID].cctx); - mtctx->jobs[chunkID].cctx = NULL; mtctx->jobs[chunkID].srcStart = NULL; { size_t const cSize = mtctx->jobs[chunkID].cSize; if (ZSTD_isError(cSize)) error = cSize; @@ -703,10 +723,9 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi { size_t const dstBufferCapacity = ZSTD_compressBound(srcSize); buffer_t const dstBuffer = ZSTDMT_getBuffer(zcs->buffPool, dstBufferCapacity); - ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(zcs->cctxPool); unsigned const jobID = zcs->nextJobID & zcs->jobIDMask; - if ((cctx==NULL) || (dstBuffer.start==NULL)) { + if (dstBuffer.start==NULL) { zcs->jobs[jobID].jobCompleted = 1; zcs->nextJobID++; ZSTDMT_waitForAllJobsCompleted(zcs); @@ -727,7 +746,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi zcs->jobs[jobID].cdict = zcs->nextJobID==0 ? zcs->cdict : NULL; zcs->jobs[jobID].fullFrameSize = zcs->frameContentSize; zcs->jobs[jobID].dstBuff = dstBuffer; - zcs->jobs[jobID].cctx = cctx; + zcs->jobs[jobID].cctxPool = zcs->cctxPool; zcs->jobs[jobID].firstChunk = (zcs->nextJobID==0); zcs->jobs[jobID].lastChunk = endFrame; zcs->jobs[jobID].jobCompleted = 0; @@ -804,8 +823,6 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi ZSTDMT_releaseAllJobResources(zcs); return job.cSize; } - ZSTDMT_releaseCCtx(zcs->cctxPool, job.cctx); - zcs->jobs[wJobID].cctx = NULL; DEBUGLOG(5, "zcs->params.fParams.checksumFlag : %u ", zcs->params.fParams.checksumFlag); if (zcs->params.fParams.checksumFlag) { XXH64_update(&zcs->xxhState, (const char*)job.srcStart + job.dictSize, job.srcSize); @@ -884,7 +901,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, /* fill input buffer */ if ((input->src) && (mtctx->inBuff.buffer.start)) { /* support NULL input */ size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); - DEBUGLOG(2, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); + DEBUGLOG(5, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad); input->pos += toLoad; mtctx->inBuff.filled += toLoad; diff --git a/tests/fuzzer.c b/tests/fuzzer.c index aa1ebd48f..667de08ce 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -105,12 +105,13 @@ static unsigned FUZ_highbit32(U32 v32) typedef struct { unsigned long long totalMalloc; + size_t currentMalloc; size_t peakMalloc; unsigned nbMalloc; unsigned nbFree; } mallocCounter_t; -static const mallocCounter_t INIT_MALLOC_COUNTER = { 0, 0, 0, 0 }; +static const mallocCounter_t INIT_MALLOC_COUNTER = { 0, 0, 0, 0, 0 }; static void* FUZ_mallocDebug(void* counter, size_t size) { @@ -118,7 +119,9 @@ static void* FUZ_mallocDebug(void* counter, size_t size) void* const ptr = malloc(size); if (ptr==NULL) return NULL; mcPtr->totalMalloc += size; - mcPtr->peakMalloc += size; + mcPtr->currentMalloc += size; + if (mcPtr->currentMalloc > mcPtr->peakMalloc) + mcPtr->peakMalloc = mcPtr->currentMalloc; mcPtr->nbMalloc += 1; return ptr; } @@ -126,9 +129,10 @@ static void* FUZ_mallocDebug(void* counter, size_t size) static void FUZ_freeDebug(void* counter, void* address) { mallocCounter_t* const mcPtr = (mallocCounter_t*)counter; - free(address); + DISPLAYLEVEL(4, "releasing %u KB \n", (U32)(malloc_size(address) >> 10)); mcPtr->nbFree += 1; - mcPtr->peakMalloc -= malloc_size(address); /* OS-X specific */ + mcPtr->currentMalloc -= malloc_size(address); /* OS-X specific */ + free(address); } static void FUZ_displayMallocStats(mallocCounter_t count) @@ -139,6 +143,14 @@ static void FUZ_displayMallocStats(mallocCounter_t count) (U32)(count.totalMalloc >> 10)); } +#define CHECK_Z(f) { \ + size_t const err = f; \ + if (ZSTD_isError(err)) { \ + DISPLAY("Error => %s : %s ", \ + #f, ZSTD_getErrorName(err)); \ + exit(1); \ +} } + static int FUZ_mallocTests(unsigned seed, double compressibility) { size_t const inSize = 64 MB + 16 MB + 4 MB + 1 MB + 256 KB + 64 KB; /* 85.3 MB */ @@ -162,7 +174,7 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) mallocCounter_t malcount = INIT_MALLOC_COUNTER; ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount }; ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); - ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel); + CHECK_Z( ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel) ); ZSTD_freeCCtx(cctx); DISPLAYLEVEL(3, "compressCCtx level %i : ", compressionLevel); FUZ_displayMallocStats(malcount); @@ -176,9 +188,9 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) ZSTD_CCtx* const cstream = ZSTD_createCStream_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_initCStream(cstream, compressionLevel); - ZSTD_compressStream(cstream, &out, &in); - ZSTD_endStream(cstream, &out); + CHECK_Z( ZSTD_initCStream(cstream, compressionLevel) ); + CHECK_Z( ZSTD_compressStream(cstream, &out, &in) ); + CHECK_Z( ZSTD_endStream(cstream, &out) ); ZSTD_freeCStream(cstream); DISPLAYLEVEL(3, "compressStream level %i : ", compressionLevel); FUZ_displayMallocStats(malcount); @@ -194,9 +206,9 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); - ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel) ); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads) ); + while ( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end) ) {} ZSTD_freeCCtx(cctx); DISPLAYLEVEL(3, "compress_generic,-T%u,end level %i : ", nbThreads, compressionLevel); @@ -213,10 +225,10 @@ static int FUZ_mallocTests(unsigned seed, double compressibility) ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem); ZSTD_outBuffer out = { outBuffer, outSize, 0 }; ZSTD_inBuffer in = { inBuffer, inSize, 0 }; - ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel); - ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue); - ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel) ); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads) ); + CHECK_Z( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue) ); + while ( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end) ) {} ZSTD_freeCCtx(cctx); DISPLAYLEVEL(3, "compress_generic,-T%u,continue level %i : ", nbThreads, compressionLevel); @@ -1046,6 +1058,7 @@ static size_t FUZ_randomLength(U32* seed, U32 maxLog) goto _output_error; \ } } +#undef CHECK_Z #define CHECK_Z(f) { \ size_t const err = f; \ if (ZSTD_isError(err)) { \ @@ -1473,7 +1486,7 @@ int main(int argc, const char** argv) if (proba!=FUZ_compressibility_default) DISPLAY("Compressibility : %u%%\n", proba); if (memTestsOnly) { - g_displayLevel=3; + g_displayLevel = MAX(3, g_displayLevel); return FUZ_mallocTests(seed, ((double)proba) / 100); } From 4616fad18b2f8950144cdbcffc081b74cc115fb1 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 10 Jul 2017 17:16:41 -0700 Subject: [PATCH 06/13] improved ZSTDMT_compress() memory usage does not need the input buffer for streaming operations also : reduced a few tests time length --- lib/compress/zstdmt_compress.c | 31 +++++++++++++++++++------------ tests/fuzzer.c | 2 +- tests/playTests.sh | 31 +++++++++++++++---------------- tests/zstreamtest.c | 19 +------------------ 4 files changed, 36 insertions(+), 47 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index a176c3eea..fbb86b008 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -134,9 +134,11 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize) /* large enough, but not too much */ return buf; /* size conditions not respected : scratch this buffer, create new one */ + DEBUGLOG(2, "existing buffer does not meet size conditions => freeing"); ZSTD_free(buf.start, pool->cMem); } /* create new buffer */ + DEBUGLOG(2, "create a new buffer"); { buffer_t buffer; void* const start = ZSTD_malloc(bSize, pool->cMem); if (start==NULL) bSize = 0; @@ -149,12 +151,14 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize) /* store buffer for later re-use, up to pool capacity */ static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* pool, buffer_t buf) { + DEBUGLOG(2, "ZSTDMT_releaseBuffer"); if (buf.start == NULL) return; /* release on NULL */ if (pool->nbBuffers < pool->totalBuffers) { pool->bTable[pool->nbBuffers++] = buf; /* store for later re-use */ return; } /* Reached bufferPool capacity (should not happen) */ + DEBUGLOG(2, "buffer pool capacity reached => freeing "); ZSTD_free(buf.start, pool->cMem); } @@ -635,8 +639,8 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, if (zcs->nbThreads==1) { DEBUGLOG(4, "single thread mode"); return ZSTD_initCStream_internal(zcs->cctxPool->cctx[0], - dict, dictSize, cdict, - params, pledgedSrcSize); + dict, dictSize, cdict, + params, pledgedSrcSize); } if (zcs->allJobsCompleted == 0) { /* previous compression not correctly finished */ @@ -671,9 +675,7 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10)); zcs->marginSize = zcs->targetSectionSize >> 2; zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize + zcs->marginSize; - zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->buffPool, zcs->inBuffSize); - if (zcs->inBuff.buffer.start == NULL) return ERROR(memory_allocation); - zcs->inBuff.filled = 0; + zcs->inBuff.buffer = g_nullBuffer; zcs->dictSize = 0; zcs->doneJobID = 0; zcs->nextJobID = 0; @@ -899,13 +901,18 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, } /* fill input buffer */ - if ((input->src) && (mtctx->inBuff.buffer.start)) { /* support NULL input */ - size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); - DEBUGLOG(5, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); - memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad); - input->pos += toLoad; - mtctx->inBuff.filled += toLoad; - } + if (input->src) { /* support NULL input */ + if (mtctx->inBuff.buffer.start == NULL) { + mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->buffPool, mtctx->inBuffSize); + if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation); + mtctx->inBuff.filled = 0; + } + { size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); + DEBUGLOG(5, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); + memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad); + input->pos += toLoad; + mtctx->inBuff.filled += toLoad; + } } if ( (mtctx->inBuff.filled >= newJobThreshold) /* filled enough : let's compress */ && (mtctx->nextJobID <= mtctx->doneJobID + mtctx->jobIDMask) ) { /* avoid overwriting job round buffer */ diff --git a/tests/fuzzer.c b/tests/fuzzer.c index 667de08ce..1e25383f5 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -129,7 +129,7 @@ static void* FUZ_mallocDebug(void* counter, size_t size) static void FUZ_freeDebug(void* counter, void* address) { mallocCounter_t* const mcPtr = (mallocCounter_t*)counter; - DISPLAYLEVEL(4, "releasing %u KB \n", (U32)(malloc_size(address) >> 10)); + DISPLAYLEVEL(4, "freeing %u KB \n", (U32)(malloc_size(address) >> 10)); mcPtr->nbFree += 1; mcPtr->currentMalloc -= malloc_size(address); /* OS-X specific */ free(address); diff --git a/tests/playTests.sh b/tests/playTests.sh index 2e1cc6826..88a1c2ab4 100755 --- a/tests/playTests.sh +++ b/tests/playTests.sh @@ -7,17 +7,17 @@ die() { roundTripTest() { if [ -n "$3" ]; then - local_c="$3" - local_p="$2" + cLevel="$3" + proba="$2" else - local_c="$2" - local_p="" + cLevel="$2" + proba="" fi rm -f tmp1 tmp2 - $ECHO "roundTripTest: ./datagen $1 $local_p | $ZSTD -v$local_c | $ZSTD -d" - ./datagen $1 $local_p | $MD5SUM > tmp1 - ./datagen $1 $local_p | $ZSTD --ultra -v$local_c | $ZSTD -d | $MD5SUM > tmp2 + $ECHO "roundTripTest: ./datagen $1 $proba | $ZSTD -v$cLevel | $ZSTD -d" + ./datagen $1 $proba | $MD5SUM > tmp1 + ./datagen $1 $proba | $ZSTD --ultra -v$cLevel | $ZSTD -d | $MD5SUM > tmp2 $DIFF -q tmp1 tmp2 } @@ -625,16 +625,15 @@ roundTripTest -g35000000 -P75 10 roundTripTest -g35000000 -P75 11 roundTripTest -g35000000 -P75 12 -roundTripTest -g18000000 -P80 13 -roundTripTest -g18000000 -P80 14 -roundTripTest -g18000000 -P80 15 -roundTripTest -g18000000 -P80 16 -roundTripTest -g18000000 -P80 17 +roundTripTest -g18000013 -P80 13 +roundTripTest -g18000014 -P80 14 +roundTripTest -g18000015 -P80 15 +roundTripTest -g18000016 -P80 16 +roundTripTest -g18000017 -P80 17 +roundTripTest -g18000018 -P94 18 +roundTripTest -g18000019 -P94 19 -roundTripTest -g50000000 -P94 18 -roundTripTest -g50000000 -P94 19 - -roundTripTest -g99000000 -P99 20 +roundTripTest -g68000020 -P99 20 roundTripTest -g6000000000 -P99 1 fileRoundTripTest -g4193M -P99 1 diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index 9b2b8eaf8..8b84400db 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -95,19 +95,6 @@ unsigned int FUZ_rand(unsigned int* seedPtr) return rand32 >> 5; } -static void* allocFunction(void* opaque, size_t size) -{ - void* address = malloc(size); - (void)opaque; - return address; -} - -static void freeFunction(void* opaque, void* address) -{ - (void)opaque; - free(address); -} - /*====================================================== * Basic Unit tests @@ -1543,7 +1530,6 @@ int main(int argc, const char** argv) int bigTests = (sizeof(size_t) == 8); e_api selected_api = simple_api; const char* const programName = argv[0]; - ZSTD_customMem const customMem = { allocFunction, freeFunction, NULL }; ZSTD_customMem const customNULL = ZSTD_defaultCMem; /* Check command line */ @@ -1657,10 +1643,7 @@ int main(int argc, const char** argv) if (testNb==0) { result = basicUnitTests(0, ((double)proba) / 100, customNULL); /* constant seed for predictability */ - if (!result) { - DISPLAYLEVEL(3, "Unit tests using customMem :\n") - result = basicUnitTests(0, ((double)proba) / 100, customMem); /* use custom memory allocation functions */ - } } + } if (!result) { switch(selected_api) From ef0ff7fe7fd90a94721c2626d1588b053ffc76ce Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 08:54:29 -0700 Subject: [PATCH 07/13] zstdmt: removed margin for improved memory usage --- lib/compress/zstdmt_compress.c | 12 +++++------- tests/fuzzer.c | 2 ++ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index fbb86b008..255b9c7ed 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -340,7 +340,6 @@ struct ZSTDMT_CCtx_s { pthread_mutex_t jobCompleted_mutex; pthread_cond_t jobCompleted_cond; size_t targetSectionSize; - size_t marginSize; size_t inBuffSize; size_t dictSize; size_t targetDictSize; @@ -673,8 +672,7 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, zcs->targetSectionSize = MAX(ZSTDMT_SECTION_SIZE_MIN, zcs->targetSectionSize); zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize); DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10)); - zcs->marginSize = zcs->targetSectionSize >> 2; - zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize + zcs->marginSize; + zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize; zcs->inBuff.buffer = g_nullBuffer; zcs->dictSize = 0; zcs->doneJobID = 0; @@ -871,18 +869,18 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, ZSTD_inBuffer* input, ZSTD_EndDirective endOp) { - size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize + mtctx->marginSize; + size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize; assert(output->pos <= output->size); assert(input->pos <= input->size); if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) { /* current frame being ended. Only flush/end are allowed. Or start new frame with init */ return ERROR(stage_wrong); } - if (mtctx->nbThreads==1) { + if (mtctx->nbThreads==1) { /* delegate to single-thread (synchronous) */ return ZSTD_compressStream_generic(mtctx->cctxPool->cctx[0], output, input, endOp); } - /* single-pass shortcut (note : this is blocking-mode) */ + /* single-pass shortcut (note : this is synchronous-mode) */ if ( (mtctx->nextJobID==0) /* just started */ && (mtctx->inBuff.filled==0) /* nothing buffered */ && (endOp==ZSTD_e_end) /* end order */ @@ -901,7 +899,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, } /* fill input buffer */ - if (input->src) { /* support NULL input */ + if (input->size > input->pos) { /* support NULL input */ if (mtctx->inBuff.buffer.start == NULL) { mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->buffPool, mtctx->inBuffSize); if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation); diff --git a/tests/fuzzer.c b/tests/fuzzer.c index 1e25383f5..6bed9ec51 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -118,6 +118,8 @@ static void* FUZ_mallocDebug(void* counter, size_t size) mallocCounter_t* const mcPtr = (mallocCounter_t*)counter; void* const ptr = malloc(size); if (ptr==NULL) return NULL; + DISPLAYLEVEL(4, "allocating %u KB => effectively %u KB \n", + (U32)(size >> 10), (U32)(malloc_size(ptr) >> 10)); /* OS-X specific */ mcPtr->totalMalloc += size; mcPtr->currentMalloc += size; if (mcPtr->currentMalloc > mcPtr->peakMalloc) From 16261e6951910e56f0c447f012900d7e06aba733 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 14:14:07 -0700 Subject: [PATCH 08/13] buffer pool can be invoked from multiple threads --- lib/common/pool.c | 20 +++++++++--------- lib/common/pool.h | 10 ++++----- lib/compress/zstdmt_compress.c | 38 ++++++++++++++++++++++------------ 3 files changed, 40 insertions(+), 28 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index 749fa4f2f..06d8a5f57 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -92,7 +92,7 @@ POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) { * and full queues. */ ctx->queueSize = queueSize + 1; - ctx->queue = (POOL_job *)malloc(ctx->queueSize * sizeof(POOL_job)); + ctx->queue = (POOL_job*) malloc(ctx->queueSize * sizeof(POOL_job)); ctx->queueHead = 0; ctx->queueTail = 0; pthread_mutex_init(&ctx->queueMutex, NULL); @@ -100,7 +100,7 @@ POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) { pthread_cond_init(&ctx->queuePopCond, NULL); ctx->shutdown = 0; /* Allocate space for the thread handles */ - ctx->threads = (pthread_t *)malloc(numThreads * sizeof(pthread_t)); + ctx->threads = (pthread_t*)malloc(numThreads * sizeof(pthread_t)); ctx->numThreads = 0; /* Check for errors */ if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; } @@ -153,8 +153,8 @@ size_t POOL_sizeof(POOL_ctx *ctx) { + ctx->numThreads * sizeof(pthread_t); } -void POOL_add(void *ctxVoid, POOL_function function, void *opaque) { - POOL_ctx *ctx = (POOL_ctx *)ctxVoid; +void POOL_add(void* ctxVoid, POOL_function function, void *opaque) { + POOL_ctx* const ctx = (POOL_ctx*)ctxVoid; if (!ctx) { return; } pthread_mutex_lock(&ctx->queueMutex); @@ -183,22 +183,22 @@ struct POOL_ctx_s { int data; }; -POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) { +POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) { (void)numThreads; (void)queueSize; - return (POOL_ctx *)malloc(sizeof(POOL_ctx)); + return (POOL_ctx*)malloc(sizeof(POOL_ctx)); } -void POOL_free(POOL_ctx *ctx) { - if (ctx) free(ctx); +void POOL_free(POOL_ctx* ctx) { + free(ctx); } -void POOL_add(void *ctx, POOL_function function, void *opaque) { +void POOL_add(void* ctx, POOL_function function, void* opaque) { (void)ctx; function(opaque); } -size_t POOL_sizeof(POOL_ctx *ctx) { +size_t POOL_sizeof(POOL_ctx* ctx) { if (ctx==NULL) return 0; /* supports sizeof NULL */ return sizeof(*ctx); } diff --git a/lib/common/pool.h b/lib/common/pool.h index 386cd674b..957100f46 100644 --- a/lib/common/pool.h +++ b/lib/common/pool.h @@ -19,11 +19,11 @@ extern "C" { typedef struct POOL_ctx_s POOL_ctx; /*! POOL_create() : - Create a thread pool with at most `numThreads` threads. - `numThreads` must be at least 1. - The maximum number of queued jobs before blocking is `queueSize`. - `queueSize` must be at least 1. - @return : The POOL_ctx pointer on success else NULL. + * Create a thread pool with at most `numThreads` threads. + * `numThreads` must be at least 1. + * The maximum number of queued jobs before blocking is `queueSize`. + * `queueSize` must be at least 1. + * @return : POOL_ctx pointer on success, else NULL. */ POOL_ctx *POOL_create(size_t numThreads, size_t queueSize); diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 255b9c7ed..c4547c81b 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -73,6 +73,7 @@ static unsigned long long GetCurrentClockTimeMicroseconds(void) /* ===== Buffer Pool ===== */ +/* a single Buffer Pool can be invoked from multiple threads in parallel */ typedef struct buffer_s { void* start; @@ -82,6 +83,7 @@ typedef struct buffer_s { static const buffer_t g_nullBuffer = { NULL, 0 }; typedef struct ZSTDMT_bufferPool_s { + pthread_mutex_t poolMutex; unsigned totalBuffers; unsigned nbBuffers; ZSTD_customMem cMem; @@ -94,6 +96,7 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_custo ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc( sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem); if (bufPool==NULL) return NULL; + pthread_mutex_init(&bufPool->poolMutex, NULL); bufPool->totalBuffers = maxNbBuffers; bufPool->nbBuffers = 0; bufPool->cMem = cMem; @@ -106,6 +109,7 @@ static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool) if (!bufPool) return; /* compatibility with free on NULL */ for (u=0; utotalBuffers; u++) ZSTD_free(bufPool->bTable[u].start, bufPool->cMem); + pthread_mutex_destroy(&bufPool->poolMutex); ZSTD_free(bufPool, bufPool->cMem); } @@ -116,31 +120,37 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool) + (bufPool->totalBuffers - 1) * sizeof(buffer_t); unsigned u; size_t totalBufferSize = 0; + pthread_mutex_lock(&bufPool->poolMutex); for (u=0; utotalBuffers; u++) totalBufferSize += bufPool->bTable[u].size; + pthread_mutex_unlock(&bufPool->poolMutex); return poolSize + totalBufferSize; } /** ZSTDMT_getBuffer() : * assumption : invocation from main thread only ! */ -static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize) +static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool, size_t bSize) { DEBUGLOG(2, "ZSTDMT_getBuffer"); - if (pool->nbBuffers) { /* try to use an existing buffer */ - buffer_t const buf = pool->bTable[--(pool->nbBuffers)]; + pthread_mutex_lock(&bufPool->poolMutex); + if (bufPool->nbBuffers) { /* try to use an existing buffer */ + buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)]; size_t const availBufferSize = buf.size; - if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) + if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) { /* large enough, but not too much */ + pthread_mutex_unlock(&bufPool->poolMutex); return buf; + } /* size conditions not respected : scratch this buffer, create new one */ DEBUGLOG(2, "existing buffer does not meet size conditions => freeing"); - ZSTD_free(buf.start, pool->cMem); + ZSTD_free(buf.start, bufPool->cMem); } + pthread_mutex_unlock(&bufPool->poolMutex); /* create new buffer */ DEBUGLOG(2, "create a new buffer"); { buffer_t buffer; - void* const start = ZSTD_malloc(bSize, pool->cMem); + void* const start = ZSTD_malloc(bSize, bufPool->cMem); if (start==NULL) bSize = 0; buffer.start = start; /* note : start can be NULL if malloc fails ! */ buffer.size = bSize; @@ -149,23 +159,25 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize) } /* store buffer for later re-use, up to pool capacity */ -static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* pool, buffer_t buf) +static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf) { DEBUGLOG(2, "ZSTDMT_releaseBuffer"); if (buf.start == NULL) return; /* release on NULL */ - if (pool->nbBuffers < pool->totalBuffers) { - pool->bTable[pool->nbBuffers++] = buf; /* store for later re-use */ + pthread_mutex_lock(&bufPool->poolMutex); + if (bufPool->nbBuffers < bufPool->totalBuffers) { + bufPool->bTable[bufPool->nbBuffers++] = buf; /* store for later re-use */ + pthread_mutex_unlock(&bufPool->poolMutex); return; } + pthread_mutex_unlock(&bufPool->poolMutex); /* Reached bufferPool capacity (should not happen) */ DEBUGLOG(2, "buffer pool capacity reached => freeing "); - ZSTD_free(buf.start, pool->cMem); + ZSTD_free(buf.start, bufPool->cMem); } /* ===== CCtx Pool ===== */ - -/* a single cctxPool can be called from multiple threads in parallel */ +/* a single CCtx Pool can be invoked from multiple threads in parallel */ typedef struct { pthread_mutex_t poolMutex; @@ -314,7 +326,7 @@ void ZSTDMT_compressChunk(void* jobDescription) job->cSize = (job->lastChunk) ? ZSTD_compressEnd (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) : ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize); - DEBUGLOG(5, "compressed %u bytes into %u bytes (first:%u) (last:%u)", + DEBUGLOG(2, "compressed %u bytes into %u bytes (first:%u) (last:%u)", (unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk); DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize)); From 34b2b956314581f2e191697ce3b18d170cfe3d42 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 14:59:10 -0700 Subject: [PATCH 09/13] zstdmt : intermediate outBuffer allocated from within worker reduces total amount of memory needed, since jobs in queue do not have an outBuffer pre-reserved now --- lib/compress/zstdmt_compress.c | 81 ++++++++++++++++------------------ 1 file changed, 39 insertions(+), 42 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index c4547c81b..9f30d3181 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -286,6 +286,7 @@ typedef struct { ZSTD_parameters params; const ZSTD_CDict* cdict; ZSTDMT_CCtxPool* cctxPool; + ZSTDMT_bufferPool* bufPool; unsigned long long fullFrameSize; } ZSTDMT_jobDescription; @@ -295,7 +296,7 @@ void ZSTDMT_compressChunk(void* jobDescription) ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTD_CCtx* cctx = ZSTDMT_getCCtx(job->cctxPool); const void* const src = (const char*)job->srcStart + job->dictSize; - buffer_t const dstBuff = job->dstBuff; + buffer_t dstBuff = job->dstBuff; DEBUGLOG(5, "job (first:%u) (last:%u) : dictSize %u, srcSize %u", job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize); @@ -304,6 +305,16 @@ void ZSTDMT_compressChunk(void* jobDescription) goto _endJob; } + if (dstBuff.start == NULL) { + size_t const dstCapacity = ZSTD_compressBound(job->srcSize); + dstBuff = ZSTDMT_getBuffer(job->bufPool, dstCapacity); + if (dstBuff.start==NULL) { + job->cSize = ERROR(memory_allocation); + goto _endJob; + } + job->dstBuff = dstBuff; + } + if (job->cdict) { /* should only happen for first segment */ size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize); DEBUGLOG(5, "using CDict"); @@ -347,7 +358,7 @@ _endJob: struct ZSTDMT_CCtx_s { POOL_ctx* factory; ZSTDMT_jobDescription* jobs; - ZSTDMT_bufferPool* buffPool; + ZSTDMT_bufferPool* bufPool; ZSTDMT_CCtxPool* cctxPool; pthread_mutex_t jobCompleted_mutex; pthread_cond_t jobCompleted_cond; @@ -402,9 +413,9 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem) mtctx->factory = POOL_create(nbThreads, 1); mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, cMem); mtctx->jobIDMask = nbJobs - 1; - mtctx->buffPool = ZSTDMT_createBufferPool(nbThreads, cMem); + mtctx->bufPool = ZSTDMT_createBufferPool(nbThreads, cMem); mtctx->cctxPool = ZSTDMT_createCCtxPool(nbThreads, cMem); - if (!mtctx->factory | !mtctx->jobs | !mtctx->buffPool | !mtctx->cctxPool) { + if (!mtctx->factory | !mtctx->jobs | !mtctx->bufPool | !mtctx->cctxPool) { ZSTDMT_freeCCtx(mtctx); return NULL; } @@ -426,13 +437,13 @@ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) unsigned jobID; DEBUGLOG(3, "ZSTDMT_releaseAllJobResources"); for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) { - ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[jobID].dstBuff); + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff); mtctx->jobs[jobID].dstBuff = g_nullBuffer; - ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[jobID].src); + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].src); mtctx->jobs[jobID].src = g_nullBuffer; } memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription)); - ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->inBuff.buffer); + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer); mtctx->inBuff.buffer = g_nullBuffer; mtctx->allJobsCompleted = 1; } @@ -442,7 +453,7 @@ size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx) if (mtctx==NULL) return 0; /* compatible with free on NULL */ POOL_free(mtctx->factory); if (!mtctx->allJobsCompleted) ZSTDMT_releaseAllJobResources(mtctx); /* stop workers first */ - ZSTDMT_freeBufferPool(mtctx->buffPool); /* release job resources into pools first */ + ZSTDMT_freeBufferPool(mtctx->bufPool); /* release job resources into pools first */ ZSTD_free(mtctx->jobs, mtctx->cMem); ZSTDMT_freeCCtxPool(mtctx->cctxPool); ZSTD_freeCDict(mtctx->cdictLocal); @@ -456,11 +467,11 @@ size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx) { if (mtctx == NULL) return 0; /* supports sizeof NULL */ return sizeof(*mtctx) - + POOL_sizeof(mtctx->factory) - + ZSTDMT_sizeof_bufferPool(mtctx->buffPool) - + (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription) - + ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool) - + ZSTD_sizeof_CDict(mtctx->cdictLocal); + + POOL_sizeof(mtctx->factory) + + ZSTDMT_sizeof_bufferPool(mtctx->bufPool) + + (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription) + + ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool) + + ZSTD_sizeof_CDict(mtctx->cdictLocal); } size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSDTMT_parameter parameter, unsigned value) @@ -534,16 +545,9 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, size_t const chunkSize = MIN(remainingSrcSize, avgChunkSize); size_t const dstBufferCapacity = ZSTD_compressBound(chunkSize); buffer_t const dstAsBuffer = { (char*)dst + dstBufferPos, dstBufferCapacity }; - buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : ZSTDMT_getBuffer(mtctx->buffPool, dstBufferCapacity); + buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : g_nullBuffer; size_t dictSize = u ? overlapSize : 0; - if (dstBuffer.start==NULL) { - mtctx->jobs[u].cSize = ERROR(memory_allocation); /* job result */ - mtctx->jobs[u].jobCompleted = 1; - nbChunks = u+1; /* only wait and free u jobs, instead of initially expected nbChunks ones */ - break; /* let's wait for previous jobs to complete, but don't start new ones */ - } - mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize; mtctx->jobs[u].dictSize = dictSize; mtctx->jobs[u].srcSize = chunkSize; @@ -554,6 +558,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0; mtctx->jobs[u].dstBuff = dstBuffer; mtctx->jobs[u].cctxPool = mtctx->cctxPool; + mtctx->jobs[u].bufPool = mtctx->bufPool; mtctx->jobs[u].firstChunk = (u==0); mtctx->jobs[u].lastChunk = (u==nbChunks-1); mtctx->jobs[u].jobCompleted = 0; @@ -591,13 +596,13 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, memmove((char*)dst + dstPos, mtctx->jobs[chunkID].dstBuff.start, cSize); /* may overlap when chunk compressed within dst */ if (chunkID >= compressWithinDst) { /* chunk compressed into its own buffer, which must be released */ DEBUGLOG(5, "releasing buffer %u>=%u", chunkID, compressWithinDst); - ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[chunkID].dstBuff); + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[chunkID].dstBuff); } mtctx->jobs[chunkID].dstBuff = g_nullBuffer; } dstPos += cSize ; } - } + } /* for (chunkID=0; chunkIDbuffPool, dstBufferCapacity); unsigned const jobID = zcs->nextJobID & zcs->jobIDMask; - if (dstBuffer.start==NULL) { - zcs->jobs[jobID].jobCompleted = 1; - zcs->nextJobID++; - ZSTDMT_waitForAllJobsCompleted(zcs); - ZSTDMT_releaseAllJobResources(zcs); - return ERROR(memory_allocation); - } - DEBUGLOG(4, "preparing job %u to compress %u bytes with %u preload ", zcs->nextJobID, (U32)srcSize, (U32)zcs->dictSize); zcs->jobs[jobID].src = zcs->inBuff.buffer; @@ -757,8 +753,9 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi if (zcs->nextJobID) zcs->jobs[jobID].params.fParams.checksumFlag = 0; zcs->jobs[jobID].cdict = zcs->nextJobID==0 ? zcs->cdict : NULL; zcs->jobs[jobID].fullFrameSize = zcs->frameContentSize; - zcs->jobs[jobID].dstBuff = dstBuffer; + zcs->jobs[jobID].dstBuff = g_nullBuffer; zcs->jobs[jobID].cctxPool = zcs->cctxPool; + zcs->jobs[jobID].bufPool = zcs->bufPool; zcs->jobs[jobID].firstChunk = (zcs->nextJobID==0); zcs->jobs[jobID].lastChunk = endFrame; zcs->jobs[jobID].jobCompleted = 0; @@ -770,7 +767,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi if (!endFrame) { size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize); DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame); - zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->buffPool, zcs->inBuffSize); + zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool, zcs->inBuffSize); if (zcs->inBuff.buffer.start == NULL) { /* not enough memory to allocate next input buffer */ zcs->jobs[jobID].jobCompleted = 1; zcs->nextJobID++; @@ -845,19 +842,19 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi job.cSize += 4; zcs->jobs[wJobID].cSize += 4; } } - ZSTDMT_releaseBuffer(zcs->buffPool, job.src); + ZSTDMT_releaseBuffer(zcs->bufPool, job.src); zcs->jobs[wJobID].srcStart = NULL; zcs->jobs[wJobID].src = g_nullBuffer; zcs->jobs[wJobID].jobScanned = 1; } { size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos); - DEBUGLOG(5, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID); + DEBUGLOG(2, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID); memcpy((char*)output->dst + output->pos, (const char*)job.dstBuff.start + job.dstFlushed, toWrite); output->pos += toWrite; job.dstFlushed += toWrite; } if (job.dstFlushed == job.cSize) { /* output buffer fully flushed => move to next one */ - ZSTDMT_releaseBuffer(zcs->buffPool, job.dstBuff); + ZSTDMT_releaseBuffer(zcs->bufPool, job.dstBuff); zcs->jobs[wJobID].dstBuff = g_nullBuffer; zcs->jobs[wJobID].jobCompleted = 0; zcs->doneJobID++; @@ -904,7 +901,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, if (ZSTD_isError(cSize)) return cSize; input->pos = input->size; output->pos += cSize; - ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->inBuff.buffer); /* was allocated in initStream */ + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer); /* was allocated in initStream */ mtctx->allJobsCompleted = 1; mtctx->frameEnded = 1; return 0; @@ -913,7 +910,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, /* fill input buffer */ if (input->size > input->pos) { /* support NULL input */ if (mtctx->inBuff.buffer.start == NULL) { - mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->buffPool, mtctx->inBuffSize); + mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool, mtctx->inBuffSize); if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation); mtctx->inBuff.filled = 0; } From 57236184afb9dad253bee28d9a34f31f0491dca7 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 15:17:25 -0700 Subject: [PATCH 10/13] buffer pool : all buffers have same size to reduce memory fragmentation. They can be used for in or out, interchangeably. --- lib/compress/zstdmt_compress.c | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 9f30d3181..703d25e30 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -84,6 +84,7 @@ static const buffer_t g_nullBuffer = { NULL, 0 }; typedef struct ZSTDMT_bufferPool_s { pthread_mutex_t poolMutex; + size_t bufferSize; unsigned totalBuffers; unsigned nbBuffers; ZSTD_customMem cMem; @@ -97,6 +98,7 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_custo sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem); if (bufPool==NULL) return NULL; pthread_mutex_init(&bufPool->poolMutex, NULL); + bufPool->bufferSize = 64 KB; bufPool->totalBuffers = maxNbBuffers; bufPool->nbBuffers = 0; bufPool->cMem = cMem; @@ -128,10 +130,16 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool) return poolSize + totalBufferSize; } -/** ZSTDMT_getBuffer() : - * assumption : invocation from main thread only ! */ -static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool, size_t bSize) +static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* bufPool, size_t bSize) { + bufPool->bufferSize = bSize; +} + +/** ZSTDMT_getBuffer() : + * assumption : bufPool must be valid */ +static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) +{ + size_t const bSize = bufPool->bufferSize; DEBUGLOG(2, "ZSTDMT_getBuffer"); pthread_mutex_lock(&bufPool->poolMutex); if (bufPool->nbBuffers) { /* try to use an existing buffer */ @@ -151,9 +159,8 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool, size_t bSize) DEBUGLOG(2, "create a new buffer"); { buffer_t buffer; void* const start = ZSTD_malloc(bSize, bufPool->cMem); - if (start==NULL) bSize = 0; buffer.start = start; /* note : start can be NULL if malloc fails ! */ - buffer.size = bSize; + buffer.size = (start==NULL) ? 0 : bSize; return buffer; } } @@ -306,8 +313,7 @@ void ZSTDMT_compressChunk(void* jobDescription) } if (dstBuff.start == NULL) { - size_t const dstCapacity = ZSTD_compressBound(job->srcSize); - dstBuff = ZSTDMT_getBuffer(job->bufPool, dstCapacity); + dstBuff = ZSTDMT_getBuffer(job->bufPool); if (dstBuff.start==NULL) { job->cSize = ERROR(memory_allocation); goto _endJob; @@ -530,6 +536,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, return ZSTD_compress_advanced(cctx, dst, dstCapacity, src, srcSize, NULL, 0, params); } assert(avgChunkSize >= 256 KB); /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is useful to avoid allocating extra buffers */ + ZSTDMT_setBufferSize(mtctx->bufPool, ZSTD_compressBound(avgChunkSize) ); if (nbChunks > mtctx->jobIDMask+1) { /* enlarge job table */ U32 nbJobs = nbChunks; @@ -690,6 +697,7 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize); DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10)); zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize; + ZSTDMT_setBufferSize(zcs->bufPool, MAX(zcs->inBuffSize, ZSTD_compressBound(zcs->targetSectionSize)) ); zcs->inBuff.buffer = g_nullBuffer; zcs->dictSize = 0; zcs->doneJobID = 0; @@ -767,7 +775,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi if (!endFrame) { size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize); DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame); - zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool, zcs->inBuffSize); + zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool); if (zcs->inBuff.buffer.start == NULL) { /* not enough memory to allocate next input buffer */ zcs->jobs[jobID].jobCompleted = 1; zcs->nextJobID++; @@ -910,7 +918,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, /* fill input buffer */ if (input->size > input->pos) { /* support NULL input */ if (mtctx->inBuff.buffer.start == NULL) { - mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool, mtctx->inBuffSize); + mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool); if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation); mtctx->inBuff.filled = 0; } From 2a62f48bf4df42f3ca389ee891ff76da56623802 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 15:56:40 -0700 Subject: [PATCH 11/13] release input buffers from inside worker thread buffers are released sooner, which makes them available faster for next job. => decreases total nb of buffers necessary --- lib/compress/zstdmt_compress.c | 32 ++++++++++++++------------------ 1 file changed, 14 insertions(+), 18 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 703d25e30..7cf637f59 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -169,7 +169,7 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf) { DEBUGLOG(2, "ZSTDMT_releaseBuffer"); - if (buf.start == NULL) return; /* release on NULL */ + if (buf.start == NULL) return; /* compatible with release on NULL */ pthread_mutex_lock(&bufPool->poolMutex); if (bufPool->nbBuffers < bufPool->totalBuffers) { bufPool->bTable[bufPool->nbBuffers++] = buf; /* store for later re-use */ @@ -271,16 +271,11 @@ static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx) /* ===== Thread worker ===== */ -typedef struct { - buffer_t buffer; - size_t filled; -} inBuff_t; - typedef struct { buffer_t src; const void* srcStart; - size_t srcSize; size_t dictSize; + size_t srcSize; buffer_t dstBuff; size_t cSize; size_t dstFlushed; @@ -349,6 +344,8 @@ void ZSTDMT_compressChunk(void* jobDescription) _endJob: ZSTDMT_releaseCCtx(job->cctxPool, cctx); + ZSTDMT_releaseBuffer(job->bufPool, job->src); + job->src = g_nullBuffer; job->srcStart = NULL; PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex); job->jobCompleted = 1; job->jobScanned = 0; @@ -361,6 +358,11 @@ _endJob: /* ===== Multi-threaded compression ===== */ /* ------------------------------------------ */ +typedef struct { + buffer_t buffer; + size_t filled; +} inBuff_t; + struct ZSTDMT_CCtx_s { POOL_ctx* factory; ZSTDMT_jobDescription* jobs; @@ -513,6 +515,7 @@ static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbT } +/* Note : missing checksum at the end ! */ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, void* dst, size_t dstCapacity, const void* src, size_t srcSize, @@ -555,6 +558,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : g_nullBuffer; size_t dictSize = u ? overlapSize : 0; + mtctx->jobs[u].src = g_nullBuffer; mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize; mtctx->jobs[u].dictSize = dictSize; mtctx->jobs[u].srcSize = chunkSize; @@ -771,10 +775,12 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi zcs->jobs[jobID].jobCompleted_mutex = &zcs->jobCompleted_mutex; zcs->jobs[jobID].jobCompleted_cond = &zcs->jobCompleted_cond; + if (zcs->params.fParams.checksumFlag) + XXH64_update(&zcs->xxhState, (const char*)zcs->inBuff.buffer.start + zcs->dictSize, srcSize); + /* get a new buffer for next input */ if (!endFrame) { size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize); - DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame); zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool); if (zcs->inBuff.buffer.start == NULL) { /* not enough memory to allocate next input buffer */ zcs->jobs[jobID].jobCompleted = 1; @@ -783,18 +789,12 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi ZSTDMT_releaseAllJobResources(zcs); return ERROR(memory_allocation); } - DEBUGLOG(5, "inBuff currently filled to %u", (U32)zcs->inBuff.filled); zcs->inBuff.filled -= srcSize + zcs->dictSize - newDictSize; - DEBUGLOG(5, "new job : inBuff filled to %u, with %u dict and %u src", - (U32)zcs->inBuff.filled, (U32)newDictSize, - (U32)(zcs->inBuff.filled - newDictSize)); memmove(zcs->inBuff.buffer.start, (const char*)zcs->jobs[jobID].srcStart + zcs->dictSize + srcSize - newDictSize, zcs->inBuff.filled); - DEBUGLOG(5, "new inBuff pre-filled"); zcs->dictSize = newDictSize; } else { /* if (endFrame==1) */ - DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame); zcs->inBuff.buffer = g_nullBuffer; zcs->inBuff.filled = 0; zcs->dictSize = 0; @@ -842,7 +842,6 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi } DEBUGLOG(5, "zcs->params.fParams.checksumFlag : %u ", zcs->params.fParams.checksumFlag); if (zcs->params.fParams.checksumFlag) { - XXH64_update(&zcs->xxhState, (const char*)job.srcStart + job.dictSize, job.srcSize); if (zcs->frameEnded && (zcs->doneJobID+1 == zcs->nextJobID)) { /* write checksum at end of last section */ U32 const checksum = (U32)XXH64_digest(&zcs->xxhState); DEBUGLOG(5, "writing checksum : %08X \n", checksum); @@ -850,9 +849,6 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi job.cSize += 4; zcs->jobs[wJobID].cSize += 4; } } - ZSTDMT_releaseBuffer(zcs->bufPool, job.src); - zcs->jobs[wJobID].srcStart = NULL; - zcs->jobs[wJobID].src = g_nullBuffer; zcs->jobs[wJobID].jobScanned = 1; } { size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos); From 052a95f77c4a7c0c13c1c94e7b805616b5924df7 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 11 Jul 2017 17:18:26 -0700 Subject: [PATCH 12/13] fix : ZSTDMT_compress_advanced() correctly generates checksum when params.fParams.checksumFlag==1. This use case used to be impossible when only ZSTD_compress() was available --- lib/compress/zstdmt_compress.c | 49 +++++++++++++++++++++++----------- lib/compress/zstdmt_compress.h | 6 ++--- tests/fuzzer.c | 20 +++++++++++--- tests/zstreamtest.c | 7 +++-- 4 files changed, 56 insertions(+), 26 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 7cf637f59..3fdfe9e14 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -140,7 +140,7 @@ static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* bufPool, size_t bSize) static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) { size_t const bSize = bufPool->bufferSize; - DEBUGLOG(2, "ZSTDMT_getBuffer"); + DEBUGLOG(5, "ZSTDMT_getBuffer"); pthread_mutex_lock(&bufPool->poolMutex); if (bufPool->nbBuffers) { /* try to use an existing buffer */ buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)]; @@ -151,12 +151,12 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) return buf; } /* size conditions not respected : scratch this buffer, create new one */ - DEBUGLOG(2, "existing buffer does not meet size conditions => freeing"); + DEBUGLOG(5, "existing buffer does not meet size conditions => freeing"); ZSTD_free(buf.start, bufPool->cMem); } pthread_mutex_unlock(&bufPool->poolMutex); /* create new buffer */ - DEBUGLOG(2, "create a new buffer"); + DEBUGLOG(5, "create a new buffer"); { buffer_t buffer; void* const start = ZSTD_malloc(bSize, bufPool->cMem); buffer.start = start; /* note : start can be NULL if malloc fails ! */ @@ -168,17 +168,17 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) /* store buffer for later re-use, up to pool capacity */ static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf) { - DEBUGLOG(2, "ZSTDMT_releaseBuffer"); if (buf.start == NULL) return; /* compatible with release on NULL */ + DEBUGLOG(5, "ZSTDMT_releaseBuffer"); pthread_mutex_lock(&bufPool->poolMutex); if (bufPool->nbBuffers < bufPool->totalBuffers) { - bufPool->bTable[bufPool->nbBuffers++] = buf; /* store for later re-use */ + bufPool->bTable[bufPool->nbBuffers++] = buf; /* stored for later use */ pthread_mutex_unlock(&bufPool->poolMutex); return; } pthread_mutex_unlock(&bufPool->poolMutex); /* Reached bufferPool capacity (should not happen) */ - DEBUGLOG(2, "buffer pool capacity reached => freeing "); + DEBUGLOG(5, "buffer pool capacity reached => freeing "); ZSTD_free(buf.start, bufPool->cMem); } @@ -338,7 +338,7 @@ void ZSTDMT_compressChunk(void* jobDescription) job->cSize = (job->lastChunk) ? ZSTD_compressEnd (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) : ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize); - DEBUGLOG(2, "compressed %u bytes into %u bytes (first:%u) (last:%u)", + DEBUGLOG(5, "compressed %u bytes into %u bytes (first:%u) (last:%u)", (unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk); DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize)); @@ -515,13 +515,12 @@ static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbT } -/* Note : missing checksum at the end ! */ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, - void* dst, size_t dstCapacity, - const void* src, size_t srcSize, - const ZSTD_CDict* cdict, - ZSTD_parameters const params, - unsigned overlapRLog) + void* dst, size_t dstCapacity, + const void* src, size_t srcSize, + const ZSTD_CDict* cdict, + ZSTD_parameters const params, + unsigned overlapRLog) { size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog); unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, mtctx->nbThreads); @@ -531,6 +530,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, size_t remainingSrcSize = srcSize; unsigned const compressWithinDst = (dstCapacity >= ZSTD_compressBound(srcSize)) ? nbChunks : (unsigned)(dstCapacity / ZSTD_compressBound(avgChunkSize)); /* presumes avgChunkSize >= 256 KB, which should be the case */ size_t frameStartPos = 0, dstBufferPos = 0; + XXH64_state_t xxh64; DEBUGLOG(4, "nbChunks : %2u (chunkSize : %u bytes) ", nbChunks, (U32)avgChunkSize); if (nbChunks==1) { /* fallback to single-thread mode */ @@ -540,6 +540,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, } assert(avgChunkSize >= 256 KB); /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is useful to avoid allocating extra buffers */ ZSTDMT_setBufferSize(mtctx->bufPool, ZSTD_compressBound(avgChunkSize) ); + XXH64_reset(&xxh64, 0); if (nbChunks > mtctx->jobIDMask+1) { /* enlarge job table */ U32 nbJobs = nbChunks; @@ -576,6 +577,10 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, mtctx->jobs[u].jobCompleted_mutex = &mtctx->jobCompleted_mutex; mtctx->jobs[u].jobCompleted_cond = &mtctx->jobCompleted_cond; + if (params.fParams.checksumFlag) { + XXH64_update(&xxh64, srcStart + frameStartPos, chunkSize); + } + DEBUGLOG(5, "posting job %u (%u bytes)", u, (U32)chunkSize); DEBUG_PRINTHEX(6, mtctx->jobs[u].srcStart, 12); POOL_add(mtctx->factory, ZSTDMT_compressChunk, &mtctx->jobs[u]); @@ -586,8 +591,8 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, } } /* collect result */ - { unsigned chunkID; - size_t error = 0, dstPos = 0; + { size_t error = 0, dstPos = 0; + unsigned chunkID; for (chunkID=0; chunkIDjobCompleted_mutex); @@ -614,6 +619,18 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, dstPos += cSize ; } } /* for (chunkID=0; chunkID dstCapacity) { + error = ERROR(dstSize_tooSmall); + } else { + DEBUGLOG(4, "writing checksum : %08X \n", checksum); + MEM_writeLE32((char*)dst + dstPos, checksum); + dstPos += 4; + } } + if (!error) DEBUGLOG(4, "compressed size : %u ", (U32)dstPos); return error ? error : dstPos; } @@ -852,7 +869,7 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi zcs->jobs[wJobID].jobScanned = 1; } { size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos); - DEBUGLOG(2, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID); + DEBUGLOG(5, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID); memcpy((char*)output->dst + output->pos, (const char*)job.dstBuff.start + job.dstFlushed, toWrite); output->pos += toWrite; job.dstFlushed += toWrite; diff --git a/lib/compress/zstdmt_compress.h b/lib/compress/zstdmt_compress.h index a6e1759b9..7584007f1 100644 --- a/lib/compress/zstdmt_compress.h +++ b/lib/compress/zstdmt_compress.h @@ -16,8 +16,8 @@ /* Note : This is an internal API. - * Some methods are still exposed (ZSTDLIB_API), because for some time, - * it used to be the only way to invoke MT compression. + * Some methods are still exposed (ZSTDLIB_API), + * because it used to be the only way to invoke MT compression. * Now, it's recommended to use ZSTD_compress_generic() instead. * These methods will stop being exposed in a future version */ @@ -68,7 +68,7 @@ ZSTDLIB_API size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, const void* src, size_t srcSize, const ZSTD_CDict* cdict, ZSTD_parameters const params, - unsigned overlapRLog); + unsigned overlapRLog); /* overlapRLog = 9 - overlapLog */ ZSTDLIB_API size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx, const void* dict, size_t dictSize, /* dict can be released after init, a local copy is preserved within zcs */ diff --git a/tests/fuzzer.c b/tests/fuzzer.c index 6bed9ec51..904ce6fd8 100644 --- a/tests/fuzzer.c +++ b/tests/fuzzer.c @@ -278,9 +278,6 @@ static int basicUnitTests(U32 seed, double compressibility) } RDG_genBuffer(CNBuffer, CNBuffSize, compressibility, 0., seed); - /* memory tests */ - FUZ_mallocTests(seed, compressibility); - /* Basic tests */ DISPLAYLEVEL(4, "test%3i : ZSTD_getErrorName : ", testNb++); { const char* errorString = ZSTD_getErrorName(0); @@ -479,6 +476,23 @@ static int basicUnitTests(U32 seed, double compressibility) } } DISPLAYLEVEL(4, "OK \n"); + DISPLAYLEVEL(4, "test%3i : compress -T2 with checksum : ", testNb++); + { ZSTD_parameters params = ZSTD_getParams(1, CNBuffSize, 0); + params.fParams.checksumFlag = 1; + params.fParams.contentSizeFlag = 1; + CHECKPLUS(r, ZSTDMT_compress_advanced(mtctx, + compressedBuffer, compressedBufferSize, + CNBuffer, CNBuffSize, + NULL, params, 3 /*overlapRLog*/), + cSize=r ); + } + DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100); + + DISPLAYLEVEL(4, "test%3i : decompress %u bytes : ", testNb++, (U32)CNBuffSize); + { size_t const r = ZSTD_decompress(decodedBuffer, CNBuffSize, compressedBuffer, cSize); + if (r != CNBuffSize) goto _output_error; } + DISPLAYLEVEL(4, "OK \n"); + ZSTDMT_freeCCtx(mtctx); } diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index 8b84400db..3e551e331 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -1377,13 +1377,12 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double /* multi-segments compression test */ XXH64_reset(&xxhState, 0); { ZSTD_outBuffer outBuff = { cBuffer, cBufferSize, 0 } ; - U32 n; - for (n=0, cSize=0, totalTestSize=0 ; totalTestSize < maxTestSize ; n++) { + for (cSize=0, totalTestSize=0 ; (totalTestSize < maxTestSize) ; ) { /* compress random chunks into randomly sized dst buffers */ size_t const randomSrcSize = FUZ_randomLength(&lseed, maxSampleLog); size_t const srcSize = MIN(maxTestSize-totalTestSize, randomSrcSize); size_t const srcStart = FUZ_rand(&lseed) % (srcBufferSize - srcSize); - size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog); + size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog+1); size_t const dstBuffSize = MIN(cBufferSize - cSize, randomDstSize); ZSTD_EndDirective const flush = (FUZ_rand(&lseed) & 15) ? ZSTD_e_continue : ZSTD_e_flush; ZSTD_inBuffer inBuff = { srcBuffer+srcStart, srcSize, 0 }; @@ -1402,7 +1401,7 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double { size_t remainingToFlush = (size_t)(-1); while (remainingToFlush) { ZSTD_inBuffer inBuff = { NULL, 0, 0 }; - size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog); + size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog+1); size_t const adjustedDstSize = MIN(cBufferSize - cSize, randomDstSize); outBuff.size = outBuff.pos + adjustedDstSize; DISPLAYLEVEL(5, "End-flush into dst buffer of size %u \n", (U32)adjustedDstSize); From 8ef666c32509e35ebbae8229a2b15ca9d4af493d Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Wed, 12 Jul 2017 14:23:34 -0700 Subject: [PATCH 13/13] slightly increased buffer pool, to cover normal "full load" scenarios 2 buffers per active worker + 1 buffer for input loading + 1 buffer for "next input" when submitting current one + 1 buffer stuck in queue --- NEWS | 3 ++- lib/compress/zstdmt_compress.c | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/NEWS b/NEWS index 5bf184e8d..457105070 100644 --- a/NEWS +++ b/NEWS @@ -1,6 +1,7 @@ v1.3.1 +perf: substantially decreased memory usage in Multi-threading mode, thanks to reports by Tino Reichardt build: fix Visual compilation for non x86/x64 targets, reported by Greg Slazinski (#718) -API exp : breaking change : ZSTD_getframeHeader() +API exp : breaking change : ZSTD_getframeHeader() provides more information v1.3.0 cli : new : `--list` command, by Paul Cruz diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 3fdfe9e14..677e96f08 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -93,7 +93,7 @@ typedef struct ZSTDMT_bufferPool_s { static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_customMem cMem) { - unsigned const maxNbBuffers = 2*nbThreads + 2; + unsigned const maxNbBuffers = 2*nbThreads + 3; ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc( sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem); if (bufPool==NULL) return NULL;