ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #4276 from facebook/fix_patchfrom_hc_speed

Browse Source 
fix speed of --patch-from mode at high compression levels
This commit is contained in:
Yann Collet
2025-02-05 20:12:56 -08:00
committed by GitHub
parent 30e0f29c4d 23e5f80390
commit ef2bf57811
5 changed files with 180 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/dev-long-tests.yml
+2 -2
View File
@@ -67,11 +67,11 @@ jobs:
- name: thread sanitizer zstreamtest - name: thread sanitizer zstreamtest
run: CC=clang ZSTREAM_TESTTIME=-T3mn make tsan-test-zstream run: CC=clang ZSTREAM_TESTTIME=-T3mn make tsan-test-zstream
ubsan-zstreamtest: uasan-zstreamtest:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # tag=v4.1.1 - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # tag=v4.1.1
- name: undefined behavior sanitizer zstreamtest - name: ub + address sanitizer on zstreamtest
run: CC=clang make uasan-test-zstream run: CC=clang make uasan-test-zstream
# lasts ~15mn # lasts ~15mn
Makefile
+3 -3
View File
@@ -296,9 +296,9 @@ msanregressiontest:
update_regressionResults : REGRESS_RESULTS_DIR := /tmp/regress_results_dir/ update_regressionResults : REGRESS_RESULTS_DIR := /tmp/regress_results_dir/
update_regressionResults: update_regressionResults:
$(MAKE) -C programs zstd $(MAKE) -j -C programs zstd
$(MAKE) -C tests/regression test $(MAKE) -j -C tests/regression test
$(RM) -rf $(REGRESS_RESULTS_DIR) $(RM) -r $(REGRESS_RESULTS_DIR)
$(MKDIR) $(REGRESS_RESULTS_DIR) $(MKDIR) $(REGRESS_RESULTS_DIR)
./tests/regression/test \ ./tests/regression/test \
--cache tests/regression/cache \ --cache tests/regression/cache \
lib/compress/zstd_compress.c
+26 -21
View File
@@ -3117,7 +3117,7 @@ ZSTD_BlockCompressor_f ZSTD_selectBlockCompressor(ZSTD_strategy strat, ZSTD_Para
ZSTD_STATIC_ASSERT((unsigned)ZSTD_fast == 1); ZSTD_STATIC_ASSERT((unsigned)ZSTD_fast == 1);
assert(ZSTD_cParam_withinBounds(ZSTD_c_strategy, (int)strat)); assert(ZSTD_cParam_withinBounds(ZSTD_c_strategy, (int)strat));
DEBUGLOG(4, "Selected block compressor: dictMode=%d strat=%d rowMatchfinder=%d", (int)dictMode, (int)strat, (int)useRowMatchFinder); DEBUGLOG(5, "Selected block compressor: dictMode=%d strat=%d rowMatchfinder=%d", (int)dictMode, (int)strat, (int)useRowMatchFinder);
if (ZSTD_rowMatchFinderUsed(strat, useRowMatchFinder)) { if (ZSTD_rowMatchFinderUsed(strat, useRowMatchFinder)) {
static const ZSTD_BlockCompressor_f rowBasedBlockCompressors[4][3] = { static const ZSTD_BlockCompressor_f rowBasedBlockCompressors[4][3] = {
{ {
@@ -3141,7 +3141,7 @@ ZSTD_BlockCompressor_f ZSTD_selectBlockCompressor(ZSTD_strategy strat, ZSTD_Para
ZSTD_COMPRESSBLOCK_LAZY2_DEDICATEDDICTSEARCH_ROW ZSTD_COMPRESSBLOCK_LAZY2_DEDICATEDDICTSEARCH_ROW
} }
}; };
DEBUGLOG(4, "Selecting a row-based matchfinder"); DEBUGLOG(5, "Selecting a row-based matchfinder");
assert(useRowMatchFinder != ZSTD_ps_auto); assert(useRowMatchFinder != ZSTD_ps_auto);
selectedCompressor = rowBasedBlockCompressors[(int)dictMode][(int)strat - (int)ZSTD_greedy]; selectedCompressor = rowBasedBlockCompressors[(int)dictMode][(int)strat - (int)ZSTD_greedy];
} else { } else {
@@ -3619,7 +3619,7 @@ writeBlockHeader(void* op, size_t cSize, size_t blockSize, U32 lastBlock)
lastBlock + (((U32)bt_rle)<<1) + (U32)(blockSize << 3) : lastBlock + (((U32)bt_rle)<<1) + (U32)(blockSize << 3) :
lastBlock + (((U32)bt_compressed)<<1) + (U32)(cSize << 3); lastBlock + (((U32)bt_compressed)<<1) + (U32)(cSize << 3);
MEM_writeLE24(op, cBlockHeader); MEM_writeLE24(op, cBlockHeader);
DEBUGLOG(3, "writeBlockHeader: cSize: %zu blockSize: %zu lastBlock: %u", cSize, blockSize, lastBlock); DEBUGLOG(5, "writeBlockHeader: cSize: %zu blockSize: %zu lastBlock: %u", cSize, blockSize, lastBlock);
} }
/** ZSTD_buildBlockEntropyStats_literals() : /** ZSTD_buildBlockEntropyStats_literals() :
@@ -4151,18 +4151,18 @@ ZSTD_compressSeqStore_singleBlock(ZSTD_CCtx* zc,
if (cSeqsSize == 0) { if (cSeqsSize == 0) {
cSize = ZSTD_noCompressBlock(op, dstCapacity, ip, srcSize, lastBlock); cSize = ZSTD_noCompressBlock(op, dstCapacity, ip, srcSize, lastBlock);
FORWARD_IF_ERROR(cSize, "Nocompress block failed"); FORWARD_IF_ERROR(cSize, "Nocompress block failed");
DEBUGLOG(4, "Writing out nocompress block, size: %zu", cSize); DEBUGLOG(5, "Writing out nocompress block, size: %zu", cSize);
*dRep = dRepOriginal; /* reset simulated decompression repcode history */ *dRep = dRepOriginal; /* reset simulated decompression repcode history */
} else if (cSeqsSize == 1) { } else if (cSeqsSize == 1) {
cSize = ZSTD_rleCompressBlock(op, dstCapacity, *ip, srcSize, lastBlock); cSize = ZSTD_rleCompressBlock(op, dstCapacity, *ip, srcSize, lastBlock);
FORWARD_IF_ERROR(cSize, "RLE compress block failed"); FORWARD_IF_ERROR(cSize, "RLE compress block failed");
DEBUGLOG(4, "Writing out RLE block, size: %zu", cSize); DEBUGLOG(5, "Writing out RLE block, size: %zu", cSize);
*dRep = dRepOriginal; /* reset simulated decompression repcode history */ *dRep = dRepOriginal; /* reset simulated decompression repcode history */
} else { } else {
ZSTD_blockState_confirmRepcodesAndEntropyTables(&zc->blockState); ZSTD_blockState_confirmRepcodesAndEntropyTables(&zc->blockState);
writeBlockHeader(op, cSeqsSize, srcSize, lastBlock); writeBlockHeader(op, cSeqsSize, srcSize, lastBlock);
cSize = ZSTD_blockHeaderSize + cSeqsSize; cSize = ZSTD_blockHeaderSize + cSeqsSize;
DEBUGLOG(4, "Writing out compressed block, size: %zu", cSize); DEBUGLOG(5, "Writing out compressed block, size: %zu", cSize);
} }
if (zc->blockState.prevCBlock->entropy.fse.offcode_repeatMode == FSE_repeat_valid) if (zc->blockState.prevCBlock->entropy.fse.offcode_repeatMode == FSE_repeat_valid)
@@ -4357,7 +4357,7 @@ ZSTD_compressBlock_splitBlock(ZSTD_CCtx* zc,
{ {
U32 nbSeq; U32 nbSeq;
size_t cSize; size_t cSize;
DEBUGLOG(4, "ZSTD_compressBlock_splitBlock"); DEBUGLOG(5, "ZSTD_compressBlock_splitBlock");
assert(zc->appliedParams.postBlockSplitter == ZSTD_ps_enable); assert(zc->appliedParams.postBlockSplitter == ZSTD_ps_enable);
{ const size_t bss = ZSTD_buildSeqStore(zc, src, srcSize); { const size_t bss = ZSTD_buildSeqStore(zc, src, srcSize);
@@ -4368,7 +4368,7 @@ ZSTD_compressBlock_splitBlock(ZSTD_CCtx* zc,
RETURN_ERROR_IF(zc->seqCollector.collectSequences, sequenceProducer_failed, "Uncompressible block"); RETURN_ERROR_IF(zc->seqCollector.collectSequences, sequenceProducer_failed, "Uncompressible block");
cSize = ZSTD_noCompressBlock(dst, dstCapacity, src, srcSize, lastBlock); cSize = ZSTD_noCompressBlock(dst, dstCapacity, src, srcSize, lastBlock);
FORWARD_IF_ERROR(cSize, "ZSTD_noCompressBlock failed"); FORWARD_IF_ERROR(cSize, "ZSTD_noCompressBlock failed");
DEBUGLOG(4, "ZSTD_compressBlock_splitBlock: Nocompress block"); DEBUGLOG(5, "ZSTD_compressBlock_splitBlock: Nocompress block");
return cSize; return cSize;
} }
nbSeq = (U32)(zc->seqStore.sequences - zc->seqStore.sequencesStart); nbSeq = (U32)(zc->seqStore.sequences - zc->seqStore.sequencesStart);
@@ -4603,7 +4603,7 @@ static size_t ZSTD_compress_frameChunk(ZSTD_CCtx* cctx,
assert(cctx->appliedParams.cParams.windowLog <= ZSTD_WINDOWLOG_MAX); assert(cctx->appliedParams.cParams.windowLog <= ZSTD_WINDOWLOG_MAX);
DEBUGLOG(4, "ZSTD_compress_frameChunk (blockSizeMax=%u)", (unsigned)blockSizeMax); DEBUGLOG(5, "ZSTD_compress_frameChunk (srcSize=%u, blockSizeMax=%u)", (unsigned)srcSize, (unsigned)blockSizeMax);
if (cctx->appliedParams.fParams.checksumFlag && srcSize) if (cctx->appliedParams.fParams.checksumFlag && srcSize)
XXH64_update(&cctx->xxhState, src, srcSize); XXH64_update(&cctx->xxhState, src, srcSize);
@@ -4898,7 +4898,8 @@ size_t ZSTD_compressBlock(ZSTD_CCtx* cctx, void* dst, size_t dstCapacity, const
/*! ZSTD_loadDictionaryContent() : /*! ZSTD_loadDictionaryContent() :
* @return : 0, or an error code * @return : 0, or an error code
*/ */
static size_t ZSTD_loadDictionaryContent(ZSTD_MatchState_t* ms, static size_t
ZSTD_loadDictionaryContent(ZSTD_MatchState_t* ms,
ldmState_t* ls, ldmState_t* ls,
ZSTD_cwksp* ws, ZSTD_cwksp* ws,
ZSTD_CCtx_params const* params, ZSTD_CCtx_params const* params,
@@ -4948,17 +4949,18 @@ static size_t ZSTD_loadDictionaryContent(ZSTD_MatchState_t* ms,
} }
ZSTD_window_update(&ms->window, src, srcSize, /* forceNonContiguous */ 0); ZSTD_window_update(&ms->window, src, srcSize, /* forceNonContiguous */ 0);
DEBUGLOG(4, "ZSTD_loadDictionaryContent(): useRowMatchFinder=%d", (int)params->useRowMatchFinder); DEBUGLOG(4, "ZSTD_loadDictionaryContent: useRowMatchFinder=%d", (int)params->useRowMatchFinder);
if (loadLdmDict) { /* Load the entire dict into LDM matchfinders. */ if (loadLdmDict) { /* Load the entire dict into LDM matchfinders. */
DEBUGLOG(4, "ZSTD_loadDictionaryContent: Trigger loadLdmDict");
ZSTD_window_update(&ls->window, src, srcSize, /* forceNonContiguous */ 0); ZSTD_window_update(&ls->window, src, srcSize, /* forceNonContiguous */ 0);
ls->loadedDictEnd = params->forceWindow ? 0 : (U32)(iend - ls->window.base); ls->loadedDictEnd = params->forceWindow ? 0 : (U32)(iend - ls->window.base);
ZSTD_ldm_fillHashTable(ls, ip, iend, &params->ldmParams); ZSTD_ldm_fillHashTable(ls, ip, iend, &params->ldmParams);
DEBUGLOG(4, "ZSTD_loadDictionaryContent: ZSTD_ldm_fillHashTable completes");
} }
/* If the dict is larger than we can reasonably index in our tables, only load the suffix. */ /* If the dict is larger than we can reasonably index in our tables, only load the suffix. */
if (params->cParams.strategy < ZSTD_btultra) { { U32 maxDictSize = 1U << MIN(MAX(params->cParams.hashLog + 3, params->cParams.chainLog + 1), 31);
U32 maxDictSize = 8U << MIN(MAX(params->cParams.hashLog, params->cParams.chainLog), 28);
if (srcSize > maxDictSize) { if (srcSize > maxDictSize) {
ip = iend - maxDictSize; ip = iend - maxDictSize;
src = ip; src = ip;
@@ -5022,6 +5024,7 @@ static size_t ZSTD_loadDictionaryContent(ZSTD_MatchState_t* ms,
|| !defined(ZSTD_EXCLUDE_BTOPT_BLOCK_COMPRESSOR) \ || !defined(ZSTD_EXCLUDE_BTOPT_BLOCK_COMPRESSOR) \
|| !defined(ZSTD_EXCLUDE_BTULTRA_BLOCK_COMPRESSOR) || !defined(ZSTD_EXCLUDE_BTULTRA_BLOCK_COMPRESSOR)
assert(srcSize >= HASH_READ_SIZE); assert(srcSize >= HASH_READ_SIZE);
DEBUGLOG(4, "Fill %u bytes into the Binary Tree", (unsigned)srcSize);
ZSTD_updateTree(ms, iend-HASH_READ_SIZE, iend); ZSTD_updateTree(ms, iend-HASH_READ_SIZE, iend);
#else #else
assert(0); /* shouldn't be called: cparams should've been adjusted. */ assert(0); /* shouldn't be called: cparams should've been adjusted. */
@@ -5598,7 +5601,8 @@ static size_t ZSTD_initCDict_internal(
return 0; return 0;
} }
static ZSTD_CDict* ZSTD_createCDict_advanced_internal(size_t dictSize, static ZSTD_CDict*
ZSTD_createCDict_advanced_internal(size_t dictSize,
ZSTD_dictLoadMethod_e dictLoadMethod, ZSTD_dictLoadMethod_e dictLoadMethod,
ZSTD_compressionParameters cParams, ZSTD_compressionParameters cParams,
ZSTD_ParamSwitch_e useRowMatchFinder, ZSTD_ParamSwitch_e useRowMatchFinder,
@@ -5606,6 +5610,7 @@ static ZSTD_CDict* ZSTD_createCDict_advanced_internal(size_t dictSize,
ZSTD_customMem customMem) ZSTD_customMem customMem)
{ {
if ((!customMem.customAlloc) ^ (!customMem.customFree)) return NULL; if ((!customMem.customAlloc) ^ (!customMem.customFree)) return NULL;
DEBUGLOG(3, "ZSTD_createCDict_advanced_internal (dictSize=%u)", (unsigned)dictSize);
{ size_t const workspaceSize = { size_t const workspaceSize =
ZSTD_cwksp_alloc_size(sizeof(ZSTD_CDict)) + ZSTD_cwksp_alloc_size(sizeof(ZSTD_CDict)) +
@@ -5642,6 +5647,7 @@ ZSTD_CDict* ZSTD_createCDict_advanced(const void* dictBuffer, size_t dictSize,
{ {
ZSTD_CCtx_params cctxParams; ZSTD_CCtx_params cctxParams;
ZSTD_memset(&cctxParams, 0, sizeof(cctxParams)); ZSTD_memset(&cctxParams, 0, sizeof(cctxParams));
DEBUGLOG(3, "ZSTD_createCDict_advanced, dictSize=%u, mode=%u", (unsigned)dictSize, (unsigned)dictContentType);
ZSTD_CCtxParams_init(&cctxParams, 0); ZSTD_CCtxParams_init(&cctxParams, 0);
cctxParams.cParams = cParams; cctxParams.cParams = cParams;
cctxParams.customMem = customMem; cctxParams.customMem = customMem;
@@ -5662,7 +5668,7 @@ ZSTD_CDict* ZSTD_createCDict_advanced2(
ZSTD_compressionParameters cParams; ZSTD_compressionParameters cParams;
ZSTD_CDict* cdict; ZSTD_CDict* cdict;
DEBUGLOG(3, "ZSTD_createCDict_advanced2, mode %u", (unsigned)dictContentType); DEBUGLOG(3, "ZSTD_createCDict_advanced2, dictSize=%u, mode=%u", (unsigned)dictSize, (unsigned)dictContentType);
if (!customMem.customAlloc ^ !customMem.customFree) return NULL; if (!customMem.customAlloc ^ !customMem.customFree) return NULL;
if (cctxParams.enableDedicatedDictSearch) { if (cctxParams.enableDedicatedDictSearch) {
@@ -5681,7 +5687,7 @@ ZSTD_CDict* ZSTD_createCDict_advanced2(
&cctxParams, ZSTD_CONTENTSIZE_UNKNOWN, dictSize, ZSTD_cpm_createCDict); &cctxParams, ZSTD_CONTENTSIZE_UNKNOWN, dictSize, ZSTD_cpm_createCDict);
} }
DEBUGLOG(3, "ZSTD_createCDict_advanced2: DDS: %u", cctxParams.enableDedicatedDictSearch); DEBUGLOG(3, "ZSTD_createCDict_advanced2: DedicatedDictSearch=%u", cctxParams.enableDedicatedDictSearch);
cctxParams.cParams = cParams; cctxParams.cParams = cParams;
cctxParams.useRowMatchFinder = ZSTD_resolveRowMatchFinderMode(cctxParams.useRowMatchFinder, &cParams); cctxParams.useRowMatchFinder = ZSTD_resolveRowMatchFinderMode(cctxParams.useRowMatchFinder, &cParams);
@@ -5767,6 +5773,7 @@ const ZSTD_CDict* ZSTD_initStaticCDict(
ZSTD_CDict* cdict; ZSTD_CDict* cdict;
ZSTD_CCtx_params params; ZSTD_CCtx_params params;
DEBUGLOG(4, "ZSTD_initStaticCDict (dictSize==%u)", (unsigned)dictSize);
if ((size_t)workspace & 7) return NULL; /* 8-aligned */ if ((size_t)workspace & 7) return NULL; /* 8-aligned */
{ {
@@ -5777,8 +5784,6 @@ const ZSTD_CDict* ZSTD_initStaticCDict(
ZSTD_cwksp_move(&cdict->workspace, &ws); ZSTD_cwksp_move(&cdict->workspace, &ws);
} }
DEBUGLOG(4, "(workspaceSize < neededSize) : (%u < %u) => %u",
(unsigned)workspaceSize, (unsigned)neededSize, (unsigned)(workspaceSize < neededSize));
if (workspaceSize < neededSize) return NULL; if (workspaceSize < neededSize) return NULL;
ZSTD_CCtxParams_init(&params, 0); ZSTD_CCtxParams_init(&params, 0);
@@ -6357,7 +6362,7 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx,
*/ */
params.compressionLevel = cctx->cdict->compressionLevel; params.compressionLevel = cctx->cdict->compressionLevel;
} }
DEBUGLOG(4, "ZSTD_compressStream2 : transparent init stage"); DEBUGLOG(4, "ZSTD_CCtx_init_compressStream2 : transparent init stage");
if (endOp == ZSTD_e_end) cctx->pledgedSrcSizePlusOne = inSize + 1; /* auto-determine pledgedSrcSize */ if (endOp == ZSTD_e_end) cctx->pledgedSrcSizePlusOne = inSize + 1; /* auto-determine pledgedSrcSize */
{ size_t const dictSize = prefixDict.dict { size_t const dictSize = prefixDict.dict
@@ -6388,9 +6393,9 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx,
params.nbWorkers = 0; /* do not invoke multi-threading when src size is too small */ params.nbWorkers = 0; /* do not invoke multi-threading when src size is too small */
} }
if (params.nbWorkers > 0) { if (params.nbWorkers > 0) {
#if ZSTD_TRACE # if ZSTD_TRACE
cctx->traceCtx = (ZSTD_trace_compress_begin != NULL) ? ZSTD_trace_compress_begin(cctx) : 0; cctx->traceCtx = (ZSTD_trace_compress_begin != NULL) ? ZSTD_trace_compress_begin(cctx) : 0;
#endif # endif
/* mt context creation */ /* mt context creation */
if (cctx->mtctx == NULL) { if (cctx->mtctx == NULL) {
DEBUGLOG(4, "ZSTD_compressStream2: creating new mtctx for nbWorkers=%u", DEBUGLOG(4, "ZSTD_compressStream2: creating new mtctx for nbWorkers=%u",
lib/compress/zstdmt_compress.c
+114 -73
View File
@@ -90,9 +90,9 @@ static unsigned long long GetCurrentClockTimeMicroseconds(void)
typedef struct buffer_s { typedef struct buffer_s {
void* start; void* start;
size_t capacity; size_t capacity;
} buffer_t; } Buffer;
static const buffer_t g_nullBuffer = { NULL, 0 }; static const Buffer g_nullBuffer = { NULL, 0 };
typedef struct ZSTDMT_bufferPool_s { typedef struct ZSTDMT_bufferPool_s {
ZSTD_pthread_mutex_t poolMutex; ZSTD_pthread_mutex_t poolMutex;
@@ -100,7 +100,7 @@ typedef struct ZSTDMT_bufferPool_s {
unsigned totalBuffers; unsigned totalBuffers;
unsigned nbBuffers; unsigned nbBuffers;
ZSTD_customMem cMem; ZSTD_customMem cMem;
buffer_t* buffers; Buffer* buffers;
} ZSTDMT_bufferPool; } ZSTDMT_bufferPool;
static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool) static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)
@@ -128,7 +128,7 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned maxNbBuffers, ZSTD_cu
ZSTD_customFree(bufPool, cMem); ZSTD_customFree(bufPool, cMem);
return NULL; return NULL;
} }
bufPool->buffers = (buffer_t*)ZSTD_customCalloc(maxNbBuffers * sizeof(buffer_t), cMem); bufPool->buffers = (Buffer*)ZSTD_customCalloc(maxNbBuffers * sizeof(Buffer), cMem);
if (bufPool->buffers==NULL) { if (bufPool->buffers==NULL) {
ZSTDMT_freeBufferPool(bufPool); ZSTDMT_freeBufferPool(bufPool);
return NULL; return NULL;
@@ -144,7 +144,7 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned maxNbBuffers, ZSTD_cu
static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool) static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
{ {
size_t const poolSize = sizeof(*bufPool); size_t const poolSize = sizeof(*bufPool);
size_t const arraySize = bufPool->totalBuffers * sizeof(buffer_t); size_t const arraySize = bufPool->totalBuffers * sizeof(Buffer);
unsigned u; unsigned u;
size_t totalBufferSize = 0; size_t totalBufferSize = 0;
ZSTD_pthread_mutex_lock(&bufPool->poolMutex); ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
@@ -189,13 +189,13 @@ static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool,
* assumption : bufPool must be valid * assumption : bufPool must be valid
* @return : a buffer, with start pointer and size * @return : a buffer, with start pointer and size
* note: allocation may fail, in this case, start==NULL and size==0 */ * note: allocation may fail, in this case, start==NULL and size==0 */
static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) static Buffer ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
{ {
size_t const bSize = bufPool->bufferSize; size_t const bSize = bufPool->bufferSize;
DEBUGLOG(5, "ZSTDMT_getBuffer: bSize = %u", (U32)bufPool->bufferSize); DEBUGLOG(5, "ZSTDMT_getBuffer: bSize = %u", (U32)bufPool->bufferSize);
ZSTD_pthread_mutex_lock(&bufPool->poolMutex); ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers) { /* try to use an existing buffer */ if (bufPool->nbBuffers) { /* try to use an existing buffer */
buffer_t const buf = bufPool->buffers[--(bufPool->nbBuffers)]; Buffer const buf = bufPool->buffers[--(bufPool->nbBuffers)];
size_t const availBufferSize = buf.capacity; size_t const availBufferSize = buf.capacity;
bufPool->buffers[bufPool->nbBuffers] = g_nullBuffer; bufPool->buffers[bufPool->nbBuffers] = g_nullBuffer;
if ((availBufferSize >= bSize) & ((availBufferSize>>3) <= bSize)) { if ((availBufferSize >= bSize) & ((availBufferSize>>3) <= bSize)) {
@@ -212,7 +212,7 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex); ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
/* create new buffer */ /* create new buffer */
DEBUGLOG(5, "ZSTDMT_getBuffer: create a new buffer"); DEBUGLOG(5, "ZSTDMT_getBuffer: create a new buffer");
{ buffer_t buffer; { Buffer buffer;
void* const start = ZSTD_customMalloc(bSize, bufPool->cMem); void* const start = ZSTD_customMalloc(bSize, bufPool->cMem);
buffer.start = start; /* note : start can be NULL if malloc fails ! */ buffer.start = start; /* note : start can be NULL if malloc fails ! */
buffer.capacity = (start==NULL) ? 0 : bSize; buffer.capacity = (start==NULL) ? 0 : bSize;
@@ -231,12 +231,12 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
* @return : a buffer that is at least the buffer pool buffer size. * @return : a buffer that is at least the buffer pool buffer size.
* If a reallocation happens, the data in the input buffer is copied. * If a reallocation happens, the data in the input buffer is copied.
*/ */
static buffer_t ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buffer) static Buffer ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, Buffer buffer)
{ {
size_t const bSize = bufPool->bufferSize; size_t const bSize = bufPool->bufferSize;
if (buffer.capacity < bSize) { if (buffer.capacity < bSize) {
void* const start = ZSTD_customMalloc(bSize, bufPool->cMem); void* const start = ZSTD_customMalloc(bSize, bufPool->cMem);
buffer_t newBuffer; Buffer newBuffer;
newBuffer.start = start; newBuffer.start = start;
newBuffer.capacity = start == NULL ? 0 : bSize; newBuffer.capacity = start == NULL ? 0 : bSize;
if (start != NULL) { if (start != NULL) {
@@ -252,7 +252,7 @@ static buffer_t ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buffer)
#endif #endif
/* store buffer for later re-use, up to pool capacity */ /* store buffer for later re-use, up to pool capacity */
static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf) static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, Buffer buf)
{ {
DEBUGLOG(5, "ZSTDMT_releaseBuffer"); DEBUGLOG(5, "ZSTDMT_releaseBuffer");
if (buf.start == NULL) return; /* compatible with release on NULL */ if (buf.start == NULL) return; /* compatible with release on NULL */
@@ -290,7 +290,7 @@ static size_t ZSTDMT_sizeof_seqPool(ZSTDMT_seqPool* seqPool)
return ZSTDMT_sizeof_bufferPool(seqPool); return ZSTDMT_sizeof_bufferPool(seqPool);
} }
static RawSeqStore_t bufferToSeq(buffer_t buffer) static RawSeqStore_t bufferToSeq(Buffer buffer)
{ {
RawSeqStore_t seq = kNullRawSeqStore; RawSeqStore_t seq = kNullRawSeqStore;
seq.seq = (rawSeq*)buffer.start; seq.seq = (rawSeq*)buffer.start;
@@ -298,9 +298,9 @@ static RawSeqStore_t bufferToSeq(buffer_t buffer)
return seq; return seq;
} }
static buffer_t seqToBuffer(RawSeqStore_t seq) static Buffer seqToBuffer(RawSeqStore_t seq)
{ {
buffer_t buffer; Buffer buffer;
buffer.start = seq.seq; buffer.start = seq.seq;
buffer.capacity = seq.capacity * sizeof(rawSeq); buffer.capacity = seq.capacity * sizeof(rawSeq);
return buffer; return buffer;
@@ -466,7 +466,7 @@ static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
typedef struct { typedef struct {
void const* start; void const* start;
size_t size; size_t size;
} range_t; } Range;
typedef struct { typedef struct {
/* All variables in the struct are protected by mutex. */ /* All variables in the struct are protected by mutex. */
@@ -482,10 +482,10 @@ typedef struct {
ZSTD_pthread_mutex_t ldmWindowMutex; ZSTD_pthread_mutex_t ldmWindowMutex;
ZSTD_pthread_cond_t ldmWindowCond; /* Signaled when ldmWindow is updated */ ZSTD_pthread_cond_t ldmWindowCond; /* Signaled when ldmWindow is updated */
ZSTD_window_t ldmWindow; /* A thread-safe copy of ldmState.window */ ZSTD_window_t ldmWindow; /* A thread-safe copy of ldmState.window */
} serialState_t; } SerialState;
static int static int
ZSTDMT_serialState_reset(serialState_t* serialState, ZSTDMT_serialState_reset(SerialState* serialState,
ZSTDMT_seqPool* seqPool, ZSTDMT_seqPool* seqPool,
ZSTD_CCtx_params params, ZSTD_CCtx_params params,
size_t jobSize, size_t jobSize,
@@ -555,7 +555,7 @@ ZSTDMT_serialState_reset(serialState_t* serialState,
return 0; return 0;
} }
static int ZSTDMT_serialState_init(serialState_t* serialState) static int ZSTDMT_serialState_init(SerialState* serialState)
{ {
int initError = 0; int initError = 0;
ZSTD_memset(serialState, 0, sizeof(*serialState)); ZSTD_memset(serialState, 0, sizeof(*serialState));
@@ -566,7 +566,7 @@ static int ZSTDMT_serialState_init(serialState_t* serialState)
return initError; return initError;
} }
static void ZSTDMT_serialState_free(serialState_t* serialState) static void ZSTDMT_serialState_free(SerialState* serialState)
{ {
ZSTD_customMem cMem = serialState->params.customMem; ZSTD_customMem cMem = serialState->params.customMem;
ZSTD_pthread_mutex_destroy(&serialState->mutex); ZSTD_pthread_mutex_destroy(&serialState->mutex);
@@ -577,9 +577,10 @@ static void ZSTDMT_serialState_free(serialState_t* serialState)
ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem); ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem);
} }
static void ZSTDMT_serialState_update(serialState_t* serialState, static void
ZSTD_CCtx* jobCCtx, RawSeqStore_t seqStore, ZSTDMT_serialState_genSequences(SerialState* serialState,
range_t src, unsigned jobID) RawSeqStore_t* seqStore,
Range src, unsigned jobID)
{ {
/* Wait for our turn */ /* Wait for our turn */
ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex);
@@ -592,12 +593,13 @@ static void ZSTDMT_serialState_update(serialState_t* serialState,
/* It is now our turn, do any processing necessary */ /* It is now our turn, do any processing necessary */
if (serialState->params.ldmParams.enableLdm == ZSTD_ps_enable) { if (serialState->params.ldmParams.enableLdm == ZSTD_ps_enable) {
size_t error; size_t error;
assert(seqStore.seq != NULL && seqStore.pos == 0 && DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update");
seqStore.size == 0 && seqStore.capacity > 0); assert(seqStore->seq != NULL && seqStore->pos == 0 &&
seqStore->size == 0 && seqStore->capacity > 0);
assert(src.size <= serialState->params.jobSize); assert(src.size <= serialState->params.jobSize);
ZSTD_window_update(&serialState->ldmState.window, src.start, src.size, /* forceNonContiguous */ 0); ZSTD_window_update(&serialState->ldmState.window, src.start, src.size, /* forceNonContiguous */ 0);
error = ZSTD_ldm_generateSequences( error = ZSTD_ldm_generateSequences(
&serialState->ldmState, &seqStore, &serialState->ldmState, seqStore,
&serialState->params.ldmParams, src.start, src.size); &serialState->params.ldmParams, src.start, src.size);
/* We provide a large enough buffer to never fail. */ /* We provide a large enough buffer to never fail. */
assert(!ZSTD_isError(error)); (void)error; assert(!ZSTD_isError(error)); (void)error;
@@ -616,14 +618,22 @@ static void ZSTDMT_serialState_update(serialState_t* serialState,
serialState->nextJobID++; serialState->nextJobID++;
ZSTD_pthread_cond_broadcast(&serialState->cond); ZSTD_pthread_cond_broadcast(&serialState->cond);
ZSTD_pthread_mutex_unlock(&serialState->mutex); ZSTD_pthread_mutex_unlock(&serialState->mutex);
}
if (seqStore.size > 0) { static void
ZSTD_referenceExternalSequences(jobCCtx, seqStore.seq, seqStore.size); ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an assert() check */
assert(serialState->params.ldmParams.enableLdm == ZSTD_ps_enable); ZSTD_CCtx* jobCCtx,
const RawSeqStore_t* seqStore)
{
if (seqStore->size > 0) {
DEBUGLOG(5, "ZSTDMT_serialState_applySequences: uploading %u external sequences", (unsigned)seqStore->size);
assert(serialState->params.ldmParams.enableLdm == ZSTD_ps_enable); (void)serialState;
assert(jobCCtx);
ZSTD_referenceExternalSequences(jobCCtx, seqStore->seq, seqStore->size);
} }
} }
static void ZSTDMT_serialState_ensureFinished(serialState_t* serialState, static void ZSTDMT_serialState_ensureFinished(SerialState* serialState,
unsigned jobID, size_t cSize) unsigned jobID, size_t cSize)
{ {
ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex);
@@ -647,7 +657,7 @@ static void ZSTDMT_serialState_ensureFinished(serialState_t* serialState,
/* ===== Worker thread ===== */ /* ===== Worker thread ===== */
/* ------------------------------------------ */ /* ------------------------------------------ */
static const range_t kNullRange = { NULL, 0 }; static const Range kNullRange = { NULL, 0 };
typedef struct { typedef struct {
size_t consumed; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx */ size_t consumed; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx */
@@ -657,10 +667,10 @@ typedef struct {
ZSTDMT_CCtxPool* cctxPool; /* Thread-safe - used by mtctx and (all) workers */ ZSTDMT_CCtxPool* cctxPool; /* Thread-safe - used by mtctx and (all) workers */
ZSTDMT_bufferPool* bufPool; /* Thread-safe - used by mtctx and (all) workers */ ZSTDMT_bufferPool* bufPool; /* Thread-safe - used by mtctx and (all) workers */
ZSTDMT_seqPool* seqPool; /* Thread-safe - used by mtctx and (all) workers */ ZSTDMT_seqPool* seqPool; /* Thread-safe - used by mtctx and (all) workers */
serialState_t* serial; /* Thread-safe - used by mtctx and (all) workers */ SerialState* serial; /* Thread-safe - used by mtctx and (all) workers */
buffer_t dstBuff; /* set by worker (or mtctx), then read by worker & mtctx, then modified by mtctx => no barrier */ Buffer dstBuff; /* set by worker (or mtctx), then read by worker & mtctx, then modified by mtctx => no barrier */
range_t prefix; /* set by mtctx, then read by worker & mtctx => no barrier */ Range prefix; /* set by mtctx, then read by worker & mtctx => no barrier */
range_t src; /* set by mtctx, then read by worker & mtctx => no barrier */ Range src; /* set by mtctx, then read by worker & mtctx => no barrier */
unsigned jobID; /* set by mtctx, then read by worker => no barrier */ unsigned jobID; /* set by mtctx, then read by worker => no barrier */
unsigned firstJob; /* set by mtctx, then read by worker => no barrier */ unsigned firstJob; /* set by mtctx, then read by worker => no barrier */
unsigned lastJob; /* set by mtctx, then read by worker => no barrier */ unsigned lastJob; /* set by mtctx, then read by worker => no barrier */
@@ -686,9 +696,10 @@ static void ZSTDMT_compressionJob(void* jobDescription)
ZSTD_CCtx_params jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */ ZSTD_CCtx_params jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */
ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(job->cctxPool); ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(job->cctxPool);
RawSeqStore_t rawSeqStore = ZSTDMT_getSeq(job->seqPool); RawSeqStore_t rawSeqStore = ZSTDMT_getSeq(job->seqPool);
buffer_t dstBuff = job->dstBuff; Buffer dstBuff = job->dstBuff;
size_t lastCBlockSize = 0; size_t lastCBlockSize = 0;
DEBUGLOG(5, "ZSTDMT_compressionJob: job %u", job->jobID);
/* resources */ /* resources */
if (cctx==NULL) JOB_ERROR(ERROR(memory_allocation)); if (cctx==NULL) JOB_ERROR(ERROR(memory_allocation));
if (dstBuff.start == NULL) { /* streaming job : doesn't provide a dstBuffer */ if (dstBuff.start == NULL) { /* streaming job : doesn't provide a dstBuffer */
@@ -710,11 +721,15 @@ static void ZSTDMT_compressionJob(void* jobDescription)
/* init */ /* init */
/* Perform serial step as early as possible */
ZSTDMT_serialState_genSequences(job->serial, &rawSeqStore, job->src, job->jobID);
if (job->cdict) { if (job->cdict) {
size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, &jobParams, job->fullFrameSize); size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, &jobParams, job->fullFrameSize);
assert(job->firstJob); /* only allowed for first job */ assert(job->firstJob); /* only allowed for first job */
if (ZSTD_isError(initError)) JOB_ERROR(initError); if (ZSTD_isError(initError)) JOB_ERROR(initError);
} else { /* srcStart points at reloaded section */ } else {
U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size;
{ size_t const forceWindowError = ZSTD_CCtxParams_setParameter(&jobParams, ZSTD_c_forceMaxWindow, !job->firstJob); { size_t const forceWindowError = ZSTD_CCtxParams_setParameter(&jobParams, ZSTD_c_forceMaxWindow, !job->firstJob);
if (ZSTD_isError(forceWindowError)) JOB_ERROR(forceWindowError); if (ZSTD_isError(forceWindowError)) JOB_ERROR(forceWindowError);
@@ -723,16 +738,17 @@ static void ZSTDMT_compressionJob(void* jobDescription)
size_t const err = ZSTD_CCtxParams_setParameter(&jobParams, ZSTD_c_deterministicRefPrefix, 0); size_t const err = ZSTD_CCtxParams_setParameter(&jobParams, ZSTD_c_deterministicRefPrefix, 0);
if (ZSTD_isError(err)) JOB_ERROR(err); if (ZSTD_isError(err)) JOB_ERROR(err);
} }
DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size);
{ size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, { size_t const initError = ZSTD_compressBegin_advanced_internal(cctx,
job->prefix.start, job->prefix.size, ZSTD_dct_rawContent, /* load dictionary in "content-only" mode (no header analysis) */ job->prefix.start, job->prefix.size, ZSTD_dct_rawContent,
ZSTD_dtlm_fast, ZSTD_dtlm_fast,
NULL, /*cdict*/ NULL, /*cdict*/
&jobParams, pledgedSrcSize); &jobParams, pledgedSrcSize);
if (ZSTD_isError(initError)) JOB_ERROR(initError); if (ZSTD_isError(initError)) JOB_ERROR(initError);
} } } }
/* Perform serial step as early as possible, but after CCtx initialization */ /* External Sequences can only be applied after CCtx initialization */
ZSTDMT_serialState_update(job->serial, cctx, rawSeqStore, job->src, job->jobID); ZSTDMT_serialState_applySequences(job->serial, cctx, &rawSeqStore);
if (!job->firstJob) { /* flush and overwrite frame header when it's not first job */ if (!job->firstJob) { /* flush and overwrite frame header when it's not first job */
size_t const hSize = ZSTD_compressContinue_public(cctx, dstBuff.start, dstBuff.capacity, job->src.start, 0); size_t const hSize = ZSTD_compressContinue_public(cctx, dstBuff.start, dstBuff.capacity, job->src.start, 0);
@@ -741,7 +757,7 @@ static void ZSTDMT_compressionJob(void* jobDescription)
ZSTD_invalidateRepCodes(cctx); ZSTD_invalidateRepCodes(cctx);
} }
/* compress */ /* compress the entire job by smaller chunks, for better granularity */
{ size_t const chunkSize = 4*ZSTD_BLOCKSIZE_MAX; { size_t const chunkSize = 4*ZSTD_BLOCKSIZE_MAX;
int const nbChunks = (int)((job->src.size + (chunkSize-1)) / chunkSize); int const nbChunks = (int)((job->src.size + (chunkSize-1)) / chunkSize);
const BYTE* ip = (const BYTE*) job->src.start; const BYTE* ip = (const BYTE*) job->src.start;
@@ -809,10 +825,10 @@ _endJob:
/* ------------------------------------------ */ /* ------------------------------------------ */
typedef struct { typedef struct {
range_t prefix; /* read-only non-owned prefix buffer */ Range prefix; /* read-only non-owned prefix buffer */
buffer_t buffer; Buffer buffer;
size_t filled; size_t filled;
} inBuff_t; } InBuff_t;
typedef struct { typedef struct {
BYTE* buffer; /* The round input buffer. All jobs get references BYTE* buffer; /* The round input buffer. All jobs get references
@@ -826,9 +842,9 @@ typedef struct {
* the inBuff is sent to the worker thread. * the inBuff is sent to the worker thread.
* pos <= capacity. * pos <= capacity.
*/ */
} roundBuff_t; } RoundBuff_t;
static const roundBuff_t kNullRoundBuff = {NULL, 0, 0}; static const RoundBuff_t kNullRoundBuff = {NULL, 0, 0};
#define RSYNC_LENGTH 32 #define RSYNC_LENGTH 32
/* Don't create chunks smaller than the zstd block size. /* Don't create chunks smaller than the zstd block size.
@@ -845,7 +861,7 @@ typedef struct {
U64 hash; U64 hash;
U64 hitMask; U64 hitMask;
U64 primePower; U64 primePower;
} rsyncState_t; } RSyncState_t;
struct ZSTDMT_CCtx_s { struct ZSTDMT_CCtx_s {
POOL_ctx* factory; POOL_ctx* factory;
@@ -857,10 +873,10 @@ struct ZSTDMT_CCtx_s {
size_t targetSectionSize; size_t targetSectionSize;
size_t targetPrefixSize; size_t targetPrefixSize;
int jobReady; /* 1 => one job is already prepared, but pool has shortage of workers. Don't create a new job. */ int jobReady; /* 1 => one job is already prepared, but pool has shortage of workers. Don't create a new job. */
inBuff_t inBuff; InBuff_t inBuff;
roundBuff_t roundBuff; RoundBuff_t roundBuff;
serialState_t serial; SerialState serial;
rsyncState_t rsync; RSyncState_t rsync;
unsigned jobIDMask; unsigned jobIDMask;
unsigned doneJobID; unsigned doneJobID;
unsigned nextJobID; unsigned nextJobID;
@@ -1245,13 +1261,11 @@ size_t ZSTDMT_initCStream_internal(
/* init */ /* init */
if (params.nbWorkers != mtctx->params.nbWorkers) if (params.nbWorkers != mtctx->params.nbWorkers)
FORWARD_IF_ERROR( ZSTDMT_resize(mtctx, params.nbWorkers) , ""); FORWARD_IF_ERROR( ZSTDMT_resize(mtctx, (unsigned)params.nbWorkers) , "");
if (params.jobSize != 0 && params.jobSize < ZSTDMT_JOBSIZE_MIN) params.jobSize = ZSTDMT_JOBSIZE_MIN; if (params.jobSize != 0 && params.jobSize < ZSTDMT_JOBSIZE_MIN) params.jobSize = ZSTDMT_JOBSIZE_MIN;
if (params.jobSize > (size_t)ZSTDMT_JOBSIZE_MAX) params.jobSize = (size_t)ZSTDMT_JOBSIZE_MAX; if (params.jobSize > (size_t)ZSTDMT_JOBSIZE_MAX) params.jobSize = (size_t)ZSTDMT_JOBSIZE_MAX;
DEBUGLOG(4, "ZSTDMT_initCStream_internal: %u workers", params.nbWorkers);
if (mtctx->allJobsCompleted == 0) { /* previous compression not correctly finished */ if (mtctx->allJobsCompleted == 0) { /* previous compression not correctly finished */
ZSTDMT_waitForAllJobsCompleted(mtctx); ZSTDMT_waitForAllJobsCompleted(mtctx);
ZSTDMT_releaseAllJobResources(mtctx); ZSTDMT_releaseAllJobResources(mtctx);
@@ -1260,15 +1274,14 @@ size_t ZSTDMT_initCStream_internal(
mtctx->params = params; mtctx->params = params;
mtctx->frameContentSize = pledgedSrcSize; mtctx->frameContentSize = pledgedSrcSize;
if (dict) {
ZSTD_freeCDict(mtctx->cdictLocal); ZSTD_freeCDict(mtctx->cdictLocal);
if (dict) {
mtctx->cdictLocal = ZSTD_createCDict_advanced(dict, dictSize, mtctx->cdictLocal = ZSTD_createCDict_advanced(dict, dictSize,
ZSTD_dlm_byCopy, dictContentType, /* note : a loadPrefix becomes an internal CDict */ ZSTD_dlm_byCopy, dictContentType, /* note : a loadPrefix becomes an internal CDict */
params.cParams, mtctx->cMem); params.cParams, mtctx->cMem);
mtctx->cdict = mtctx->cdictLocal; mtctx->cdict = mtctx->cdictLocal;
if (mtctx->cdictLocal == NULL) return ERROR(memory_allocation); if (mtctx->cdictLocal == NULL) return ERROR(memory_allocation);
} else { } else {
ZSTD_freeCDict(mtctx->cdictLocal);
mtctx->cdictLocal = NULL; mtctx->cdictLocal = NULL;
mtctx->cdict = cdict; mtctx->cdict = cdict;
} }
@@ -1334,9 +1347,32 @@ size_t ZSTDMT_initCStream_internal(
mtctx->allJobsCompleted = 0; mtctx->allJobsCompleted = 0;
mtctx->consumed = 0; mtctx->consumed = 0;
mtctx->produced = 0; mtctx->produced = 0;
/* update dictionary */
ZSTD_freeCDict(mtctx->cdictLocal);
mtctx->cdictLocal = NULL;
mtctx->cdict = NULL;
if (dict) {
if (dictContentType == ZSTD_dct_rawContent) {
mtctx->inBuff.prefix.start = (const BYTE*)dict;
mtctx->inBuff.prefix.size = dictSize;
} else {
/* note : a loadPrefix becomes an internal CDict */
mtctx->cdictLocal = ZSTD_createCDict_advanced(dict, dictSize,
ZSTD_dlm_byRef, dictContentType,
params.cParams, mtctx->cMem);
mtctx->cdict = mtctx->cdictLocal;
if (mtctx->cdictLocal == NULL) return ERROR(memory_allocation);
}
} else {
mtctx->cdict = cdict;
}
if (ZSTDMT_serialState_reset(&mtctx->serial, mtctx->seqPool, params, mtctx->targetSectionSize, if (ZSTDMT_serialState_reset(&mtctx->serial, mtctx->seqPool, params, mtctx->targetSectionSize,
dict, dictSize, dictContentType)) dict, dictSize, dictContentType))
return ERROR(memory_allocation); return ERROR(memory_allocation);
return 0; return 0;
} }
@@ -1403,7 +1439,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* mtctx, size_t srcSize, ZS
mtctx->roundBuff.pos += srcSize; mtctx->roundBuff.pos += srcSize;
mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.buffer = g_nullBuffer;
mtctx->inBuff.filled = 0; mtctx->inBuff.filled = 0;
/* Set the prefix */ /* Set the prefix for next job */
if (!endFrame) { if (!endFrame) {
size_t const newPrefixSize = MIN(srcSize, mtctx->targetPrefixSize); size_t const newPrefixSize = MIN(srcSize, mtctx->targetPrefixSize);
mtctx->inBuff.prefix.start = src + srcSize - newPrefixSize; mtctx->inBuff.prefix.start = src + srcSize - newPrefixSize;
@@ -1540,12 +1576,17 @@ static size_t ZSTDMT_flushProduced(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, u
* If the data of the first job is broken up into two segments, we cover both * If the data of the first job is broken up into two segments, we cover both
* sections. * sections.
*/ */
static range_t ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx) static Range ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx)
{ {
unsigned const firstJobID = mtctx->doneJobID; unsigned const firstJobID = mtctx->doneJobID;
unsigned const lastJobID = mtctx->nextJobID; unsigned const lastJobID = mtctx->nextJobID;
unsigned jobID; unsigned jobID;
/* no need to check during first round */
size_t roundBuffCapacity = mtctx->roundBuff.capacity;
size_t nbJobs1stRoundMin = roundBuffCapacity / mtctx->targetSectionSize;
if (lastJobID < nbJobs1stRoundMin) return kNullRange;
for (jobID = firstJobID; jobID < lastJobID; ++jobID) { for (jobID = firstJobID; jobID < lastJobID; ++jobID) {
unsigned const wJobID = jobID & mtctx->jobIDMask; unsigned const wJobID = jobID & mtctx->jobIDMask;
size_t consumed; size_t consumed;
@@ -1555,7 +1596,7 @@ static range_t ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx)
ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex); ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex);
if (consumed < mtctx->jobs[wJobID].src.size) { if (consumed < mtctx->jobs[wJobID].src.size) {
range_t range = mtctx->jobs[wJobID].prefix; Range range = mtctx->jobs[wJobID].prefix;
if (range.size == 0) { if (range.size == 0) {
/* Empty prefix */ /* Empty prefix */
range = mtctx->jobs[wJobID].src; range = mtctx->jobs[wJobID].src;
@@ -1571,7 +1612,7 @@ static range_t ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx)
/** /**
* Returns non-zero iff buffer and range overlap. * Returns non-zero iff buffer and range overlap.
*/ */
static int ZSTDMT_isOverlapped(buffer_t buffer, range_t range) static int ZSTDMT_isOverlapped(Buffer buffer, Range range)
{ {
BYTE const* const bufferStart = (BYTE const*)buffer.start; BYTE const* const bufferStart = (BYTE const*)buffer.start;
BYTE const* const rangeStart = (BYTE const*)range.start; BYTE const* const rangeStart = (BYTE const*)range.start;
@@ -1591,10 +1632,10 @@ static int ZSTDMT_isOverlapped(buffer_t buffer, range_t range)
} }
} }
static int ZSTDMT_doesOverlapWindow(buffer_t buffer, ZSTD_window_t window) static int ZSTDMT_doesOverlapWindow(Buffer buffer, ZSTD_window_t window)
{ {
range_t extDict; Range extDict;
range_t prefix; Range prefix;
DEBUGLOG(5, "ZSTDMT_doesOverlapWindow"); DEBUGLOG(5, "ZSTDMT_doesOverlapWindow");
extDict.start = window.dictBase + window.lowLimit; extDict.start = window.dictBase + window.lowLimit;
@@ -1613,7 +1654,7 @@ static int ZSTDMT_doesOverlapWindow(buffer_t buffer, ZSTD_window_t window)
|| ZSTDMT_isOverlapped(buffer, prefix); || ZSTDMT_isOverlapped(buffer, prefix);
} }
static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, buffer_t buffer) static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer)
{ {
if (mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable) { if (mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable) {
ZSTD_pthread_mutex_t* mutex = &mtctx->serial.ldmWindowMutex; ZSTD_pthread_mutex_t* mutex = &mtctx->serial.ldmWindowMutex;
@@ -1638,16 +1679,16 @@ static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, buffer_t buffer)
*/ */
static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx) static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx)
{ {
range_t const inUse = ZSTDMT_getInputDataInUse(mtctx); Range const inUse = ZSTDMT_getInputDataInUse(mtctx);
size_t const spaceLeft = mtctx->roundBuff.capacity - mtctx->roundBuff.pos; size_t const spaceLeft = mtctx->roundBuff.capacity - mtctx->roundBuff.pos;
size_t const target = mtctx->targetSectionSize; size_t const spaceNeeded = mtctx->targetSectionSize;
buffer_t buffer; Buffer buffer;
DEBUGLOG(5, "ZSTDMT_tryGetInputRange"); DEBUGLOG(5, "ZSTDMT_tryGetInputRange");
assert(mtctx->inBuff.buffer.start == NULL); assert(mtctx->inBuff.buffer.start == NULL);
assert(mtctx->roundBuff.capacity >= target); assert(mtctx->roundBuff.capacity >= spaceNeeded);
if (spaceLeft < target) { if (spaceLeft < spaceNeeded) {
/* ZSTD_invalidateRepCodes() doesn't work for extDict variants. /* ZSTD_invalidateRepCodes() doesn't work for extDict variants.
* Simply copy the prefix to the beginning in that case. * Simply copy the prefix to the beginning in that case.
*/ */
@@ -1666,7 +1707,7 @@ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx)
mtctx->roundBuff.pos = prefixSize; mtctx->roundBuff.pos = prefixSize;
} }
buffer.start = mtctx->roundBuff.buffer + mtctx->roundBuff.pos; buffer.start = mtctx->roundBuff.buffer + mtctx->roundBuff.pos;
buffer.capacity = target; buffer.capacity = spaceNeeded;
if (ZSTDMT_isOverlapped(buffer, inUse)) { if (ZSTDMT_isOverlapped(buffer, inUse)) {
DEBUGLOG(5, "Waiting for buffer..."); DEBUGLOG(5, "Waiting for buffer...");
@@ -1693,7 +1734,7 @@ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx)
typedef struct { typedef struct {
size_t toLoad; /* The number of bytes to load from the input. */ size_t toLoad; /* The number of bytes to load from the input. */
int flush; /* Boolean declaring if we must flush because we found a synchronization point. */ int flush; /* Boolean declaring if we must flush because we found a synchronization point. */
} syncPoint_t; } SyncPoint;
/** /**
* Searches through the input for a synchronization point. If one is found, we * Searches through the input for a synchronization point. If one is found, we
@@ -1701,14 +1742,14 @@ typedef struct {
* Otherwise, we will load as many bytes as possible and instruct the caller * Otherwise, we will load as many bytes as possible and instruct the caller
* to continue as normal. * to continue as normal.
*/ */
static syncPoint_t static SyncPoint
findSynchronizationPoint(ZSTDMT_CCtx const* mtctx, ZSTD_inBuffer const input) findSynchronizationPoint(ZSTDMT_CCtx const* mtctx, ZSTD_inBuffer const input)
{ {
BYTE const* const istart = (BYTE const*)input.src + input.pos; BYTE const* const istart = (BYTE const*)input.src + input.pos;
U64 const primePower = mtctx->rsync.primePower; U64 const primePower = mtctx->rsync.primePower;
U64 const hitMask = mtctx->rsync.hitMask; U64 const hitMask = mtctx->rsync.hitMask;
syncPoint_t syncPoint; SyncPoint syncPoint;
U64 hash; U64 hash;
BYTE const* prev; BYTE const* prev;
size_t pos; size_t pos;
@@ -1840,7 +1881,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
DEBUGLOG(5, "ZSTDMT_tryGetInputRange completed successfully : mtctx->inBuff.buffer.start = %p", mtctx->inBuff.buffer.start); DEBUGLOG(5, "ZSTDMT_tryGetInputRange completed successfully : mtctx->inBuff.buffer.start = %p", mtctx->inBuff.buffer.start);
} }
if (mtctx->inBuff.buffer.start != NULL) { if (mtctx->inBuff.buffer.start != NULL) {
syncPoint_t const syncPoint = findSynchronizationPoint(mtctx, *input); SyncPoint const syncPoint = findSynchronizationPoint(mtctx, *input);
if (syncPoint.flush && endOp == ZSTD_e_continue) { if (syncPoint.flush && endOp == ZSTD_e_continue) {
endOp = ZSTD_e_flush; endOp = ZSTD_e_flush;
} }
tests/regression/results.csv
+9 -9
View File
@@ -395,19 +395,19 @@ github, level 13 with dict, advanced
github, level 13 with dict dms, advanced one pass, 39900 github, level 13 with dict dms, advanced one pass, 39900
github, level 13 with dict dds, advanced one pass, 39900 github, level 13 with dict dds, advanced one pass, 39900
github, level 13 with dict copy, advanced one pass, 39948 github, level 13 with dict copy, advanced one pass, 39948
github, level 13 with dict load, advanced one pass, 42624 github, level 13 with dict load, advanced one pass, 42643
github, level 16, advanced one pass, 133209 github, level 16, advanced one pass, 133209
github, level 16 with dict, advanced one pass, 37902 github, level 16 with dict, advanced one pass, 37902
github, level 16 with dict dms, advanced one pass, 37902 github, level 16 with dict dms, advanced one pass, 37902
github, level 16 with dict dds, advanced one pass, 37902 github, level 16 with dict dds, advanced one pass, 37902
github, level 16 with dict copy, advanced one pass, 37892 github, level 16 with dict copy, advanced one pass, 37892
github, level 16 with dict load, advanced one pass, 42402 github, level 16 with dict load, advanced one pass, 42434
github, level 19, advanced one pass, 132879 github, level 19, advanced one pass, 132879
github, level 19 with dict, advanced one pass, 37916 github, level 19 with dict, advanced one pass, 37916
github, level 19 with dict dms, advanced one pass, 37916 github, level 19 with dict dms, advanced one pass, 37916
github, level 19 with dict dds, advanced one pass, 37916 github, level 19 with dict dds, advanced one pass, 37916
github, level 19 with dict copy, advanced one pass, 37906 github, level 19 with dict copy, advanced one pass, 37906
github, level 19 with dict load, advanced one pass, 39770 github, level 19 with dict load, advanced one pass, 40405
github, no source size, advanced one pass, 136331 github, no source size, advanced one pass, 136331
github, no source size with dict, advanced one pass, 41118 github, no source size with dict, advanced one pass, 41118
github, long distance mode, advanced one pass, 136331 github, long distance mode, advanced one pass, 136331
@@ -713,19 +713,19 @@ github, level 13 with dict, advanced
github, level 13 with dict dms, advanced one pass small out, 39900 github, level 13 with dict dms, advanced one pass small out, 39900
github, level 13 with dict dds, advanced one pass small out, 39900 github, level 13 with dict dds, advanced one pass small out, 39900
github, level 13 with dict copy, advanced one pass small out, 39948 github, level 13 with dict copy, advanced one pass small out, 39948
github, level 13 with dict load, advanced one pass small out, 42624 github, level 13 with dict load, advanced one pass small out, 42643
github, level 16, advanced one pass small out, 133209 github, level 16, advanced one pass small out, 133209
github, level 16 with dict, advanced one pass small out, 37902 github, level 16 with dict, advanced one pass small out, 37902
github, level 16 with dict dms, advanced one pass small out, 37902 github, level 16 with dict dms, advanced one pass small out, 37902
github, level 16 with dict dds, advanced one pass small out, 37902 github, level 16 with dict dds, advanced one pass small out, 37902
github, level 16 with dict copy, advanced one pass small out, 37892 github, level 16 with dict copy, advanced one pass small out, 37892
github, level 16 with dict load, advanced one pass small out, 42402 github, level 16 with dict load, advanced one pass small out, 42434
github, level 19, advanced one pass small out, 132879 github, level 19, advanced one pass small out, 132879
github, level 19 with dict, advanced one pass small out, 37916 github, level 19 with dict, advanced one pass small out, 37916
github, level 19 with dict dms, advanced one pass small out, 37916 github, level 19 with dict dms, advanced one pass small out, 37916
github, level 19 with dict dds, advanced one pass small out, 37916 github, level 19 with dict dds, advanced one pass small out, 37916
github, level 19 with dict copy, advanced one pass small out, 37906 github, level 19 with dict copy, advanced one pass small out, 37906
github, level 19 with dict load, advanced one pass small out, 39770 github, level 19 with dict load, advanced one pass small out, 40405
github, no source size, advanced one pass small out, 136331 github, no source size, advanced one pass small out, 136331
github, no source size with dict, advanced one pass small out, 41118 github, no source size with dict, advanced one pass small out, 41118
github, long distance mode, advanced one pass small out, 136331 github, long distance mode, advanced one pass small out, 136331
@@ -1031,19 +1031,19 @@ github, level 13 with dict, advanced
github, level 13 with dict dms, advanced streaming, 39900 github, level 13 with dict dms, advanced streaming, 39900
github, level 13 with dict dds, advanced streaming, 39900 github, level 13 with dict dds, advanced streaming, 39900
github, level 13 with dict copy, advanced streaming, 39948 github, level 13 with dict copy, advanced streaming, 39948
github, level 13 with dict load, advanced streaming, 42624 github, level 13 with dict load, advanced streaming, 42643
github, level 16, advanced streaming, 133209 github, level 16, advanced streaming, 133209
github, level 16 with dict, advanced streaming, 37902 github, level 16 with dict, advanced streaming, 37902
github, level 16 with dict dms, advanced streaming, 37902 github, level 16 with dict dms, advanced streaming, 37902
github, level 16 with dict dds, advanced streaming, 37902 github, level 16 with dict dds, advanced streaming, 37902
github, level 16 with dict copy, advanced streaming, 37892 github, level 16 with dict copy, advanced streaming, 37892
github, level 16 with dict load, advanced streaming, 42402 github, level 16 with dict load, advanced streaming, 42434
github, level 19, advanced streaming, 132879 github, level 19, advanced streaming, 132879
github, level 19 with dict, advanced streaming, 37916 github, level 19 with dict, advanced streaming, 37916
github, level 19 with dict dms, advanced streaming, 37916 github, level 19 with dict dms, advanced streaming, 37916
github, level 19 with dict dds, advanced streaming, 37916 github, level 19 with dict dds, advanced streaming, 37916
github, level 19 with dict copy, advanced streaming, 37906 github, level 19 with dict copy, advanced streaming, 37906
github, level 19 with dict load, advanced streaming, 39770 github, level 19 with dict load, advanced streaming, 40405
github, no source size, advanced streaming, 136331 github, no source size, advanced streaming, 136331
github, no source size with dict, advanced streaming, 41118 github, no source size with dict, advanced streaming, 41118
github, long distance mode, advanced streaming, 136331 github, long distance mode, advanced streaming, 136331
1 Data Config Method Total compressed size
395 github level 13 with dict dms advanced one pass 39900
396 github level 13 with dict dds advanced one pass 39900
397 github level 13 with dict copy advanced one pass 39948
398 github level 13 with dict load advanced one pass 42624 42643
399 github level 16 advanced one pass 133209
400 github level 16 with dict advanced one pass 37902
401 github level 16 with dict dms advanced one pass 37902
402 github level 16 with dict dds advanced one pass 37902
403 github level 16 with dict copy advanced one pass 37892
404 github level 16 with dict load advanced one pass 42402 42434
405 github level 19 advanced one pass 132879
406 github level 19 with dict advanced one pass 37916
407 github level 19 with dict dms advanced one pass 37916
408 github level 19 with dict dds advanced one pass 37916
409 github level 19 with dict copy advanced one pass 37906
410 github level 19 with dict load advanced one pass 39770 40405
411 github no source size advanced one pass 136331
412 github no source size with dict advanced one pass 41118
413 github long distance mode advanced one pass 136331
713 github level 13 with dict dms advanced one pass small out 39900
714 github level 13 with dict dds advanced one pass small out 39900
715 github level 13 with dict copy advanced one pass small out 39948
716 github level 13 with dict load advanced one pass small out 42624 42643
717 github level 16 advanced one pass small out 133209
718 github level 16 with dict advanced one pass small out 37902
719 github level 16 with dict dms advanced one pass small out 37902
720 github level 16 with dict dds advanced one pass small out 37902
721 github level 16 with dict copy advanced one pass small out 37892
722 github level 16 with dict load advanced one pass small out 42402 42434
723 github level 19 advanced one pass small out 132879
724 github level 19 with dict advanced one pass small out 37916
725 github level 19 with dict dms advanced one pass small out 37916
726 github level 19 with dict dds advanced one pass small out 37916
727 github level 19 with dict copy advanced one pass small out 37906
728 github level 19 with dict load advanced one pass small out 39770 40405
729 github no source size advanced one pass small out 136331
730 github no source size with dict advanced one pass small out 41118
731 github long distance mode advanced one pass small out 136331
1031 github level 13 with dict dms advanced streaming 39900
1032 github level 13 with dict dds advanced streaming 39900
1033 github level 13 with dict copy advanced streaming 39948
1034 github level 13 with dict load advanced streaming 42624 42643
1035 github level 16 advanced streaming 133209
1036 github level 16 with dict advanced streaming 37902
1037 github level 16 with dict dms advanced streaming 37902
1038 github level 16 with dict dds advanced streaming 37902
1039 github level 16 with dict copy advanced streaming 37892
1040 github level 16 with dict load advanced streaming 42402 42434
1041 github level 19 advanced streaming 132879
1042 github level 19 with dict advanced streaming 37916
1043 github level 19 with dict dms advanced streaming 37916
1044 github level 19 with dict dds advanced streaming 37916
1045 github level 19 with dict copy advanced streaming 37906
1046 github level 19 with dict load advanced streaming 39770 40405
1047 github no source size advanced streaming 136331
1048 github no source size with dict advanced streaming 41118
1049 github long distance mode advanced streaming 136331