Skip to content

Commit

Permalink
Revert "Created a test case which reliably reproduces bug #944"
Browse files Browse the repository at this point in the history 
This reverts commit 5098d1f.
  • Loading branch information
@Cyan4973
Cyan4973 committed Dec 12, 2017
1 parent 5098d1f commit 8a104fd
Show file tree
Hide file tree
Showing 5 changed files with 46 additions and 114 deletions.
11 changes: 6 additions & 5 deletions lib/compress/zstd_compress.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -477,7 +477,7 @@ size_t ZSTD_CCtx_setParametersUsingCCtxParams(

ZSTDLIB_API size_t ZSTD_CCtx_setPledgedSrcSize(ZSTD_CCtx* cctx, unsigned long long pledgedSrcSize)
{
DEBUGLOG(4, "ZSTD_CCtx_setPledgedSrcSize to %u bytes", (U32)pledgedSrcSize);
DEBUGLOG(4, " setting pledgedSrcSize to %u", (U32)pledgedSrcSize);
if (cctx->streamStage != zcss_init) return ERROR(stage_wrong);
cctx->pledgedSrcSizePlusOne = pledgedSrcSize+1;
return 0;
Expand All @@ -489,7 +489,7 @@ size_t ZSTD_CCtx_loadDictionary_advanced(
{
if (cctx->streamStage != zcss_init) return ERROR(stage_wrong);
if (cctx->staticSize) return ERROR(memory_allocation); /* no malloc for static CCtx */
DEBUGLOG(4, "ZSTD_CCtx_loadDictionary_advanced (size: %u)", (U32)dictSize);
DEBUGLOG(4, "load dictionary of size %u", (U32)dictSize);
ZSTD_freeCDict(cctx->cdictLocal); /* in case one already exists */
if (dict==NULL || dictSize==0) { /* no dictionary mode */
cctx->cdictLocal = NULL;
Expand Down Expand Up @@ -2273,7 +2273,7 @@ static size_t ZSTD_initCDict_internal(
ZSTD_dictMode_e dictMode,
ZSTD_compressionParameters cParams)
{
DEBUGLOG(3, "ZSTD_initCDict_internal, mode %u", (U32)dictMode);
DEBUGLOG(4, "ZSTD_initCDict_internal, mode %u", (U32)dictMode);
if ((dictLoadMethod == ZSTD_dlm_byRef) || (!dictBuffer) || (!dictSize)) {
cdict->dictBuffer = NULL;
cdict->dictContent = dictBuffer;
Expand Down Expand Up @@ -2303,7 +2303,7 @@ ZSTD_CDict* ZSTD_createCDict_advanced(const void* dictBuffer, size_t dictSize,
ZSTD_dictMode_e dictMode,
ZSTD_compressionParameters cParams, ZSTD_customMem customMem)
{
DEBUGLOG(3, "ZSTD_createCDict_advanced, mode %u", (U32)dictMode);
DEBUGLOG(4, "ZSTD_createCDict_advanced, mode %u", (U32)dictMode);
if (!customMem.customAlloc ^ !customMem.customFree) return NULL;

{ ZSTD_CDict* const cdict = (ZSTD_CDict*)ZSTD_malloc(sizeof(ZSTD_CDict), customMem);
Expand Down Expand Up @@ -2809,7 +2809,7 @@ size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
ZSTD_inBuffer* input,
ZSTD_EndDirective endOp)
{
DEBUGLOG(5, "ZSTD_compress_generic, endOp=%u ", (U32)endOp);
DEBUGLOG(5, "ZSTD_compress_generic");
/* check conditions */
if (output->pos > output->size) return ERROR(GENERIC);
if (input->pos > input->size) return ERROR(GENERIC);
Expand Down Expand Up @@ -2856,6 +2856,7 @@ size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
#ifdef ZSTD_MULTITHREAD
if (cctx->appliedParams.nbThreads > 1) {
size_t const flushMin = ZSTDMT_compressStream_generic(cctx->mtctx, output, input, endOp);
DEBUGLOG(5, "ZSTDMT_compressStream_generic result : %u", (U32)flushMin);
if ( ZSTD_isError(flushMin)
|| (endOp == ZSTD_e_end && flushMin == 0) ) { /* compression completed */
ZSTD_startNewCompression(cctx);
Expand Down
30 changes: 14 additions & 16 deletions lib/compress/zstdmt_compress.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,7 @@ static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
typedef struct {
buffer_t src;
const void* srcStart;
size_t prefixSize;
size_t dictSize;
size_t srcSize;
buffer_t dstBuff;
size_t cSize;
Expand All @@ -333,10 +333,10 @@ void ZSTDMT_compressChunk(void* jobDescription)
{
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(job->cctxPool);
const void* const src = (const char*)job->srcStart + job->prefixSize;
const void* const src = (const char*)job->srcStart + job->dictSize;
buffer_t dstBuff = job->dstBuff;
DEBUGLOG(5, "ZSTDMT_compressChunk: job (first:%u) (last:%u) : prefixSize %u, srcSize %u ",
job->firstChunk, job->lastChunk, (U32)job->prefixSize, (U32)job->srcSize);
DEBUGLOG(5, "ZSTDMT_compressChunk: 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);
Expand All @@ -356,21 +356,20 @@ void ZSTDMT_compressChunk(void* jobDescription)
if (job->cdict) {
size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize);
DEBUGLOG(4, "ZSTDMT_compressChunk: init using CDict");
assert(job->firstChunk); /* only allowed for first job */
assert(job->firstChunk); /* should only happen for first segment */
if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; }
} else { /* srcStart points at reloaded section */
ZSTD_CCtx_params jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */
size_t const forceWindowError = ZSTD_CCtxParam_setParameter(&jobParams, ZSTD_p_forceMaxWindow, !job->firstChunk);
U64 const pledgedSrcSize = job->firstChunk ? job->fullFrameSize : ZSTD_CONTENTSIZE_UNKNOWN;
/* load dictionary in "content-only" mode (no header analysis) */
size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, job->srcStart, job->prefixSize, ZSTD_dm_rawContent, jobParams, pledgedSrcSize);
DEBUGLOG(5, "ZSTD_compressBegin_advanced_internal called with windowLog = %u ", jobParams.cParams.windowLog);
size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, job->srcStart, job->dictSize, ZSTD_dm_rawContent, jobParams, pledgedSrcSize);
if (ZSTD_isError(initError) || ZSTD_isError(forceWindowError)) {
job->cSize = initError;
goto _endJob;
}
}
if (!job->firstChunk) { /* flush and overwrite frame header when it's not first job */
if (!job->firstChunk) { /* flush and overwrite frame header when it's not first segment */
size_t const hSize = ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, 0);
if (ZSTD_isError(hSize)) { job->cSize = hSize; goto _endJob; }
ZSTD_invalidateRepCodes(cctx);
Expand All @@ -381,9 +380,9 @@ 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(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));
DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize));

_endJob:
ZSTDMT_releaseCCtx(job->cctxPool, cctx);
Expand Down Expand Up @@ -619,7 +618,7 @@ static size_t ZSTDMT_compress_advanced_internal(
size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog);
unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, params.nbThreads);
size_t const proposedChunkSize = (srcSize + (nbChunks-1)) / nbChunks;
size_t const avgChunkSize = (((proposedChunkSize-1) & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize; /* avoid too small last block */
size_t const avgChunkSize = ((proposedChunkSize & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize; /* avoid too small last block */
const char* const srcStart = (const char*)src;
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 */
Expand All @@ -629,8 +628,7 @@ static size_t ZSTDMT_compress_advanced_internal(
assert(mtctx->cctxPool->totalCCtx == params.nbThreads);

DEBUGLOG(4, "ZSTDMT_compress_advanced_internal");
DEBUGLOG(4, "nbChunks : %2u (raw chunkSize : %u bytes; fixed chunkSize: %u) ",
nbChunks, (U32)proposedChunkSize, (U32)avgChunkSize);
DEBUGLOG(4, "nbChunks : %2u (chunkSize : %u bytes) ", nbChunks, (U32)avgChunkSize);
if (nbChunks==1) { /* fallback to single-thread mode */
ZSTD_CCtx* const cctx = mtctx->cctxPool->cctx[0];
if (cdict) return ZSTD_compress_usingCDict_advanced(cctx, dst, dstCapacity, src, srcSize, cdict, jobParams.fParams);
Expand Down Expand Up @@ -659,7 +657,7 @@ static size_t ZSTDMT_compress_advanced_internal(

mtctx->jobs[u].src = g_nullBuffer;
mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize;
mtctx->jobs[u].prefixSize = dictSize;
mtctx->jobs[u].dictSize = dictSize;
mtctx->jobs[u].srcSize = chunkSize;
mtctx->jobs[u].cdict = (u==0) ? cdict : NULL;
mtctx->jobs[u].fullFrameSize = srcSize;
Expand Down Expand Up @@ -890,7 +888,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
zcs->jobs[jobID].src = zcs->inBuff.buffer;
zcs->jobs[jobID].srcStart = zcs->inBuff.buffer.start;
zcs->jobs[jobID].srcSize = srcSize;
zcs->jobs[jobID].prefixSize = zcs->dictSize;
zcs->jobs[jobID].dictSize = zcs->dictSize;
assert(zcs->inBuff.filled >= srcSize + zcs->dictSize);
zcs->jobs[jobID].params = zcs->params;
/* do not calculate checksum within sections, but write it in header for first section */
Expand Down Expand Up @@ -1016,7 +1014,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
{
size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize;
unsigned forwardInputProgress = 0;
DEBUGLOG(5, "ZSTDMT_compressStream_generic ");
DEBUGLOG(5, "ZSTDMT_compressStream_generic");
assert(output->pos <= output->size);
assert(input->pos <= input->size);

Expand Down
19 changes: 7 additions & 12 deletions lib/decompress/zstd_decompress.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2451,16 +2451,14 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
return ZSTD_decompressLegacyStream(zds->legacyContext, legacyVersion, output, input);
}
#endif
return hSize; /* error */
return hSize; /* error */
}
if (hSize != 0) { /* need more input */
size_t const toLoad = hSize - zds->lhSize; /* if hSize!=0, hSize > zds->lhSize */
size_t const remainingInput = (size_t)(iend-ip);
assert(iend >= ip);
if (toLoad > remainingInput) { /* not enough input to load full header */
if (remainingInput > 0) {
memcpy(zds->headerBuffer + zds->lhSize, ip, remainingInput);
zds->lhSize += remainingInput;
if (toLoad > (size_t)(iend-ip)) { /* not enough input to load full header */
if (iend-ip > 0) {
memcpy(zds->headerBuffer + zds->lhSize, ip, iend-ip);
zds->lhSize += iend-ip;
}
input->pos = input->size;
return (MAX(ZSTD_frameHeaderSize_min, hSize) - zds->lhSize) + ZSTD_blockHeaderSize; /* remaining header bytes + next block header */
Expand All @@ -2475,10 +2473,8 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
&& (U64)(size_t)(oend-op) >= zds->fParams.frameContentSize) {
size_t const cSize = ZSTD_findFrameCompressedSize(istart, iend-istart);
if (cSize <= (size_t)(iend-istart)) {
/* shortcut : using single-pass mode */
size_t const decompressedSize = ZSTD_decompress_usingDDict(zds, op, oend-op, istart, cSize, zds->ddict);
if (ZSTD_isError(decompressedSize)) return decompressedSize;
DEBUGLOG(4, "shortcut to single-pass ZSTD_decompress_usingDDict()")
ip = istart + cSize;
op += decompressedSize;
zds->expected = 0;
Expand All @@ -2501,9 +2497,8 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
}

/* control buffer memory usage */
DEBUGLOG(4, "Control max memory usage (%u KB <= max %u KB)",
(U32)(zds->fParams.windowSize >>10),
(U32)(zds->maxWindowSize >> 10) );
DEBUGLOG(4, "Control max buffer memory usage (max %u KB)",
(U32)(zds->maxWindowSize >> 10));
zds->fParams.windowSize = MAX(zds->fParams.windowSize, 1U << ZSTD_WINDOWLOG_ABSOLUTEMIN);
if (zds->fParams.windowSize > zds->maxWindowSize) return ERROR(frameParameter_windowTooLarge);

Expand Down
19 changes: 6 additions & 13 deletions lib/zstd.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1016,8 +1016,7 @@ typedef enum {
* More threads improve speed, but also increase memory usage.
* Can only receive a value > 1 if ZSTD_MULTITHREAD is enabled.
* Special: value 0 means "do not change nbThreads" */
ZSTD_p_jobSize, /* Size of a compression job. This value is only enforced in streaming (non-blocking) mode.
* Each compression job is completed in parallel, so indirectly controls the nb of active threads.
ZSTD_p_jobSize, /* Size of a compression job. Each compression job is completed in parallel.
* 0 means default, which is dynamically determined based on compression parameters.
* Job size must be a minimum of overlapSize, or 1 KB, whichever is largest
* The minimum size is automatically and transparently enforced */
Expand Down Expand Up @@ -1145,19 +1144,13 @@ typedef enum {
* - Compression parameters cannot be changed once compression is started.
* - outpot->pos must be <= dstCapacity, input->pos must be <= srcSize
* - outpot->pos and input->pos will be updated. They are guaranteed to remain below their respective limit.
* - In single-thread mode (default), function is blocking : it completed its job before returning to caller.
* - In multi-thread mode, function is non-blocking : it just acquires a copy of input, and distribute job to internal worker threads,
* and then immediately returns, just indicating that there is some data remaining to be flushed.
* The function nonetheless guarantees forward progress : it will return only after it reads or write at least 1+ byte.
* - Exception : in multi-threading mode, if the first call requests a ZSTD_e_end directive, it is blocking : it will complete compression before giving back control to caller.
* - @return provides the minimum amount of data remaining to be flushed from internal buffers
* - @return provides the minimum amount of data still to flush from internal buffers
* or an error code, which can be tested using ZSTD_isError().
* if @return != 0, flush is not fully completed, there is still some data left within internal buffers.
* This is useful to determine if a ZSTD_e_flush or ZSTD_e_end directive is completed.
* - after a ZSTD_e_end directive, if internal buffer is not fully flushed (@return != 0),
* if @return != 0, flush is not fully completed, there is some data left within internal buffers.
* - after a ZSTD_e_end directive, if internal buffer is not fully flushed,
* only ZSTD_e_end or ZSTD_e_flush operations are allowed.
* Before starting a new compression job, or changing compression parameters,
* it is required to fully flush internal buffers.
* It is necessary to fully flush internal buffers
* before starting a new compression job, or changing compression parameters.
*/
ZSTDLIB_API size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
ZSTD_outBuffer* output,
Expand Down
Loading

0 comments on commit 8a104fd

Please sign in to comment.