diff --git a/NEWS b/NEWS index deebacf9..043fcca8 100644 --- a/NEWS +++ b/NEWS @@ -31,7 +31,8 @@ perf : level 2 compresses slightly better (~0.1-0.5%) and faster (~+1-2%) perf : faster levels 11 (~+25%) and 12 (~+10%) perf : `--favor-decSpeed` : faster decompression (~+2-4% at level 12), with fewer sequences perf : `lz4` multithreaded compression relies on liblz4, reading files in place (memory-mapped, `--[no-]mmap`) : faster at fast levels -api : lz4frame compression accepts any `dstCapacity`, and fails only if the output doesn't fit +api : lz4frame compression accepts any `dstCapacity` : `LZ4F_compressFrame()` and `LZ4F_compressUpdate()` fail only if the output doesn't fit +api : `LZ4F_flush()` and `LZ4F_compressEnd()` fill `dstBuffer` or complete : below `LZ4F_compressBound(0)`, invoke them again while they return `dstCapacity` api : `LZ4F_ERROR_dstMaxSize_tooSmall` renamed `LZ4F_ERROR_dstCapacity_tooSmall` (former name still supported) api : lz4frame : after an error, and outside of a frame, compression operations fail until `LZ4F_compressBegin()` api : lz4frame output no longer depends on the compression context's history diff --git a/lib/lz4file.c b/lib/lz4file.c index fb306b95..e13e22d3 100644 --- a/lib/lz4file.c +++ b/lib/lz4file.c @@ -561,14 +561,20 @@ LZ4F_errorCode_t LZ4F_writeClose(LZ4_writeFile_t* lz4fWrite) ret = lz4fWrite->errCode; /* a previous LZ4F_write() failure leaves the frame incomplete */ if (ret == LZ4F_OK_NoError) { - size_t const endSize = LZ4F_compressEnd(lz4fWrite->cctxPtr, - lz4fWrite->dstBuf, lz4fWrite->dstBufMaxSize, - NULL); - if (LZ4F_isError(endSize)) { - ret = endSize; - } else if (endSize != fwrite(lz4fWrite->dstBuf, 1, endSize, lz4fWrite->fp)) { - ret = returnErrorCode(LZ4F_ERROR_io_write); - } + size_t endSize; + do { /* dstBuf >= LZ4F_compressBound(0) : a single invocation completes the frame */ + endSize = LZ4F_compressEnd(lz4fWrite->cctxPtr, + lz4fWrite->dstBuf, lz4fWrite->dstBufMaxSize, + NULL); + if (LZ4F_isError(endSize)) { + ret = endSize; + break; + } + if (endSize != fwrite(lz4fWrite->dstBuf, 1, endSize, lz4fWrite->fp)) { + ret = returnErrorCode(LZ4F_ERROR_io_write); + break; + } + } while (endSize == lz4fWrite->dstBufMaxSize); } freeWriteFileResources(lz4fWrite); diff --git a/lib/lz4frame.c b/lib/lz4frame.c index 4572047c..14bdff4c 100644 --- a/lib/lz4frame.c +++ b/lib/lz4frame.c @@ -289,7 +289,7 @@ typedef struct LZ4F_cctx_s LZ4F_CustomMem cmem; LZ4F_preferences_t prefs; U32 version; - U32 cStage; /* 0 : compression uninitialized ; 1 : initialized, can compress */ + U32 cStage; /* 0 : compression uninitialized ; 1 : initialized, can compress ; 2 : frame ending (LZ4F_compressEnd() in progress) */ const LZ4F_CDict* cdict; size_t maxBlockSize; size_t tmpBuffCapacity; @@ -304,6 +304,13 @@ typedef struct LZ4F_cctx_s LZ4F_BlockCompressMode_e blockCompressMode; struct LZ4F_mtCtx_s* mt; /* multi-threading resources, kept across frames; NULL when never needed */ U32 mtActive; /* current frame is cut into jobs (prefs.nbWorkers >= 1) */ + /* LZ4F_flush() and LZ4F_compressEnd() : output not fitting into dstBuffer yet, see LZ4F_emitPending() */ + U32 flushStage; /* 0 : none ; 1 : flush point set, emitting ; LZ4F_compressEnd() only : 2 : footer ; 3 : completed */ + const BYTE* pending; /* remaining output of a block (stageBuff), or of the frame footer */ + size_t pendingSize; + BYTE* stageBuff; /* a block not fitting into dstBuffer : allocated on first need */ + size_t stageCapacity; + BYTE footer[8]; /* endMark + optional content checksum */ } LZ4F_cctx_t; /* multi-threaded compression, see LZ4F_preferences_t.nbWorkers */ @@ -609,7 +616,15 @@ size_t LZ4F_compressFrame_usingCDict(LZ4F_cctx* cctx, assert(dstEnd >= dstPtr); { size_t const tailSize = LZ4F_compressEnd(cctx, dstPtr, (size_t)(dstEnd-dstPtr), &options); /* flush last block, and generate suffix */ - FORWARD_IF_ERROR(tailSize); + int const completed = !LZ4F_isError(tailSize) && ((cctx->cStage == 0) || (cctx->flushStage == 3)); + if (!completed) { + /* the frame doesn't fit */ + if (cctx->mtActive) LZ4F_mt_abort(cctx->mt); + cctx->cStage = 0; + return LZ4F_isError(tailSize) ? tailSize : LZ4F_returnErrorCode(LZ4F_ERROR_dstCapacity_tooSmall); + } + cctx->cStage = 0; + cctx->flushStage = 0; dstPtr += tailSize; } assert(dstEnd >= dstStart); @@ -664,6 +679,7 @@ size_t LZ4F_compressFrame(void* dstBuffer, size_t dstCapacity, LZ4F_free(cctxPtr->lz4CtxPtr, cctxPtr->cmem); } LZ4F_mt_free(cctxPtr->mt, cctxPtr->cmem); + LZ4F_free(cctxPtr->stageBuff, cctxPtr->cmem); #endif return result; } @@ -784,6 +800,7 @@ LZ4F_errorCode_t LZ4F_freeCompressionContext(LZ4F_cctx* cctxPtr) LZ4F_mt_free(cctxPtr->mt, cctxPtr->cmem); /* first : waits for jobs still in flight */ LZ4F_free(cctxPtr->lz4CtxPtr, cctxPtr->cmem); /* note: LZ4_streamHC_t and LZ4_stream_t are simple POD types */ LZ4F_free(cctxPtr->tmpBuff, cctxPtr->cmem); + LZ4F_free(cctxPtr->stageBuff, cctxPtr->cmem); LZ4F_free(cctxPtr, cctxPtr->cmem); } return LZ4F_OK_NoError; @@ -864,7 +881,7 @@ size_t LZ4F_cctx_size(const LZ4F_cctx* cctx) { if (cctx == NULL) { return 0; } - return sizeof(*cctx) + cctx->tmpBuffCapacity + LZ4F_sizeofLZ4State((LZ4F_CtxType_e)cctx->lz4CtxAlloc) + return sizeof(*cctx) + cctx->tmpBuffCapacity + cctx->stageCapacity + LZ4F_sizeofLZ4State((LZ4F_CtxType_e)cctx->lz4CtxAlloc) + LZ4F_mt_sizeof(cctx->mt); } @@ -894,6 +911,8 @@ static size_t LZ4F_compressBegin_internal(LZ4F_cctx* cctx, if (cctx->mt) LZ4F_mt_abort(cctx->mt); /* jobs of an abandoned frame : must complete before @cctx is modified */ cctx->mtActive = 0; cctx->cStage = 0; + cctx->flushStage = 0; + cctx->pendingSize = 0; cctx->prefs = *preferencesPtr; DEBUGLOG(5, "LZ4F_compressBegin_internal: Independent_blocks=%u", cctx->prefs.frameInfo.blockMode); @@ -1065,7 +1084,9 @@ size_t LZ4F_compressBound(size_t srcSize, const LZ4F_preferences_t* preferencesP return LZ4F_mtCompressBound(srcSize, preferencesPtr); } if (preferencesPtr && preferencesPtr->autoFlush) { - return LZ4F_compressBound_internal(srcSize, preferencesPtr, 0); + /* + 1 : output of LZ4F_flush() and LZ4F_compressEnd() remains strictly below LZ4F_compressBound(0), + * since a filled dstBuffer means more output may remain */ + return LZ4F_compressBound_internal(srcSize, preferencesPtr, 0) + (srcSize == 0); } bound = LZ4F_compressBound_internal(srcSize+1, preferencesPtr, (size_t)-1); /* A mode switch may emit buffered input as a separate partial block. */ @@ -1238,6 +1259,100 @@ static const LZ4F_compressOptions_t k_cOptionsNull = { 0, 0, 0, 0 }; } \ } while (0) +/*-*************************************************** +* Output not fitting into dstBuffer yet (LZ4F_flush(), LZ4F_compressEnd()) +*****************************************************/ + +/* stageBuff holds one block, stored uncompressed in worst case. + * @return : 0, or an error code (allocation_failed) */ +static size_t LZ4F_reserveStage(LZ4F_cctx_t* cctx) +{ + size_t const needed = cctx->maxBlockSize + BHSize + BFSize; + if (cctx->stageCapacity >= needed) return 0; + LZ4F_free(cctx->stageBuff, cctx->cmem); + cctx->stageCapacity = 0; + cctx->stageBuff = (BYTE*)LZ4F_malloc(needed, cctx->cmem); + RETURN_ERROR_IF(cctx->stageBuff == NULL, allocation_failed); + cctx->stageCapacity = needed; + return 0; +} + +/* copies pending output into @dst + @pos, as much as fits. + * @return : nb of bytes copied */ +static size_t LZ4F_copyPending(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity, size_t pos) +{ + size_t const n = MIN(cctx->pendingSize, dstCapacity - pos); + assert(pos <= dstCapacity); + if (n) memcpy(dst + pos, cctx->pending, n); + cctx->pending += n; + cctx->pendingSize -= n; + return n; +} + +/* compression operations first emit output left pending by an incomplete LZ4F_flush(), + * entirely, since they must succeed : the flush ends there. + * @return : nb of bytes written into @dst, or an error code (callers invalidate the frame) */ +static size_t LZ4F_emitPendingWhole(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity) +{ + cctx->flushStage = 0; + RETURN_ERROR_IF(cctx->pendingSize > dstCapacity, dstCapacity_tooSmall); + return LZ4F_copyPending(cctx, dst, dstCapacity, 0); +} + +/*! LZ4F_flushBlock() : + * compresses input buffered within cctx (single-threaded mode), if any, into @dst. + * @return : nb of bytes written into @dst, or an error code (the frame is then invalidated) */ +static size_t LZ4F_flushBlock(LZ4F_cctx_t* cctxPtr, void* dstBuffer, size_t dstCapacity) +{ + compressFunc_t compress; + size_t cSize; + + DEBUGLOG(5, "LZ4F_flushBlock: %zu buffered bytes (saved dict size = %i) (dstCapacity=%u)", + cctxPtr->tmpInSize, (int)(cctxPtr->tmpIn - cctxPtr->tmpBuff), (unsigned)dstCapacity); + if (cctxPtr->tmpInSize == 0) return 0; /* nothing to flush */ + + /* select compression function */ + compress = LZ4F_selectCompression(cctxPtr->prefs.frameInfo.blockMode, cctxPtr->prefs.compressionLevel, cctxPtr->blockCompressMode); + + /* compress tmp buffer */ + cSize = LZ4F_makeBlock(dstBuffer, dstCapacity, + cctxPtr->tmpIn, cctxPtr->tmpInSize, + compress, cctxPtr->lz4CtxPtr, cctxPtr->prefs.compressionLevel, + cctxPtr->cdict, + cctxPtr->prefs.frameInfo.blockChecksumFlag); + FORWARD_IF_ERROR_INVALIDATE(cSize, cctxPtr); + assert(((void)"flush overflows dstBuffer!", cSize <= dstCapacity)); + + if (cctxPtr->prefs.frameInfo.blockMode == LZ4F_blockLinked) + cctxPtr->tmpIn += cctxPtr->tmpInSize; + cctxPtr->tmpInSize = 0; + + /* keep tmpIn within limits */ + if ((cctxPtr->tmpIn + cctxPtr->maxBlockSize) > (cctxPtr->tmpBuff + cctxPtr->tmpBuffCapacity)) { + assert(cctxPtr->prefs.frameInfo.blockMode == LZ4F_blockLinked); + LZ4F_localSaveDict(cctxPtr); + } + + return cSize; +} + +/* flush point, single-threaded mode : the buffered block is compressed into @dst when it fits in any case, + * otherwise into stageBuff, as pending output. Same output : compression capacity is the same. + * @return : nb of bytes written into @dst, or an error code (the frame is then invalidated) */ +static size_t LZ4F_flushBlockPartial(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity) +{ + size_t const worstCase = cctx->tmpInSize + BHSize + (size_t)cctx->prefs.frameInfo.blockChecksumFlag * BFSize; + size_t cSize; + if ((cctx->tmpInSize == 0) || (dstCapacity >= worstCase)) + return LZ4F_flushBlock(cctx, dst, dstCapacity); + FORWARD_IF_ERROR_INVALIDATE(LZ4F_reserveStage(cctx), cctx); + cSize = LZ4F_flushBlock(cctx, cctx->stageBuff, cctx->stageCapacity); + FORWARD_IF_ERROR(cSize); /* already invalidated */ + cctx->pending = cctx->stageBuff; + cctx->pendingSize = cSize; + return 0; +} + /*-*************************************************** * Multi-threaded compression @@ -1674,6 +1789,7 @@ struct LZ4F_mtCtx_s { unsigned nbInFree; unsigned maxInBuffs; unsigned oldest; /* oldest submitted job, not emitted yet */ + size_t emitted; /* part of the oldest job already emitted, see LZ4F_mt_emitPartial() */ unsigned nbSubmitted; int filling; /* the slot following submitted jobs is being filled */ BYTE* history; /* last bytes of content, up to LZ4F_MT_PREFIX_SIZE (linked blocks) */ @@ -1779,16 +1895,19 @@ static BYTE* LZ4F_mt_acquireInBuff(LZ4F_mtCtx_t* mt, LZ4F_CustomMem cmem) static size_t LZ4F_mt_emitOldest(LZ4F_mtCtx_t* mt, BYTE* dst, size_t dstCapacity, size_t pos, unsigned minQueue) { LZ4F_mtJob_t* const job = mt->jobs + mt->oldest; + size_t toEmit; assert(mt->nbSubmitted > 0); assert(pos <= dstCapacity); LZ4F_pool_waitDone(mt->pool, &job->done, minQueue); FORWARD_IF_ERROR(job->cSize); - RETURN_ERROR_IF(job->cSize > dstCapacity - pos, dstCapacity_tooSmall); - memcpy(dst + pos, job->outBuff, job->cSize); + toEmit = job->cSize - mt->emitted; /* part already emitted by an incomplete LZ4F_flush() */ + RETURN_ERROR_IF(toEmit > dstCapacity - pos, dstCapacity_tooSmall); + memcpy(dst + pos, job->outBuff + mt->emitted, toEmit); LZ4F_mt_releaseInBuff(mt, job); mt->oldest = (mt->oldest + 1) % mt->nbJobSlots; mt->nbSubmitted--; - return job->cSize; + mt->emitted = 0; + return toEmit; } /* emits submitted jobs in order : all of them (@all), or only the already completed ones fitting into @dst @@ -1801,7 +1920,7 @@ static size_t LZ4F_mt_emit(LZ4F_mtCtx_t* mt, BYTE* dst, size_t dstCapacity, size const LZ4F_mtJob_t* const job = mt->jobs + mt->oldest; if (!all) { if (!LZ4F_pool_isDone(mt->pool, &job->done)) break; - if (job->cSize > dstCapacity - (pos + written)) break; /* also defers an error code */ + if (job->cSize - mt->emitted > dstCapacity - (pos + written)) break; /* also defers an error code */ } { size_t const r = LZ4F_mt_emitOldest(mt, dst, dstCapacity, pos + written, 1); /* no more input */ FORWARD_IF_ERROR(r); @@ -1823,6 +1942,7 @@ static void LZ4F_mt_abort(LZ4F_mtCtx_t* mt) LZ4F_pool_setCancel(mt->pool, 0); if (mt->filling) LZ4F_mt_releaseInBuff(mt, LZ4F_mt_fillingSlot(mt)); mt->filling = 0; + mt->emitted = 0; mt->stableEnd = NULL; mt->uncTailSize = 0; } @@ -2001,6 +2121,7 @@ static size_t LZ4F_mt_begin(LZ4F_cctx_t* cctx) mt->blockSize = blockSize; mt->jobSize = jobSize; mt->oldest = 0; + mt->emitted = 0; mt->nbSubmitted = 0; mt->filling = 0; mt->historySize = 0; @@ -2040,6 +2161,59 @@ static size_t LZ4F_mt_flush(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity, si } } +/*! LZ4F_mt_flushPoint() : + * LZ4F_flush() and LZ4F_compressEnd(), multi-threaded mode : input retained within cctx is compressed. + * The incomplete uncompressed block is written into @dst when it fits in any case, otherwise into stageBuff (pending output). + * Otherwise, the job being filled is submitted, and all jobs are waited for : input is no longer read, + * and their output is emitted by LZ4F_mt_emitPartial(). + * @return : nb of bytes written into @dst, or an error code (the frame is then invalidated) */ +static size_t LZ4F_mt_flushPoint(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity) +{ + LZ4F_mtCtx_t* const mt = cctx->mt; + if (mt->uncTailSize) { + size_t const worstCase = mt->uncTailSize + BHSize + (size_t)cctx->prefs.frameInfo.blockChecksumFlag * BFSize; + size_t r; + if (dstCapacity >= worstCase) return LZ4F_mt_emitUncTail(cctx, dst, dstCapacity, 0); + FORWARD_IF_ERROR_MT(LZ4F_reserveStage(cctx), cctx); + r = LZ4F_mt_emitUncTail(cctx, cctx->stageBuff, cctx->stageCapacity, 0); + FORWARD_IF_ERROR(r); /* already invalidated */ + cctx->pending = cctx->stageBuff; + cctx->pendingSize = r; + return 0; + } + if (mt->filling) LZ4F_mt_submit(mt, LZ4F_mt_fillingSlot(mt)); + { unsigned n; + for (n = 0; n < mt->nbSubmitted; n++) + LZ4F_pool_waitDone(mt->pool, &mt->jobs[(mt->oldest + n) % mt->nbJobSlots].done, 1); + } + return 0; +} + +/*! LZ4F_mt_emitPartial() : + * emits completed jobs in order into @dst + @pos, as much as fits : the oldest one can be emitted partially. + * Requires all submitted jobs completed, see LZ4F_mt_flushPoint(). + * @return : nb of bytes written, or an error code */ +static size_t LZ4F_mt_emitPartial(LZ4F_mtCtx_t* mt, BYTE* dst, size_t dstCapacity, size_t pos) +{ + size_t const start = pos; + while (mt->nbSubmitted > 0) { + LZ4F_mtJob_t* const job = mt->jobs + mt->oldest; + size_t n; + assert(LZ4F_pool_isDone(mt->pool, &job->done)); + FORWARD_IF_ERROR(job->cSize); + n = MIN(job->cSize - mt->emitted, dstCapacity - pos); + if (n) memcpy(dst + pos, job->outBuff + mt->emitted, n); + pos += n; + mt->emitted += n; + if (mt->emitted < job->cSize) break; /* dst is full */ + LZ4F_mt_releaseInBuff(mt, job); + mt->oldest = (mt->oldest + 1) % mt->nbJobSlots; + mt->nbSubmitted--; + mt->emitted = 0; + } + return pos - start; +} + /*! LZ4F_compressUpdate_MT() : * Input is copied into jobs, submitted once full. * Job content is read in place instead, when it remains available until the job completes : @@ -2068,6 +2242,10 @@ static size_t LZ4F_compressUpdate_MT(LZ4F_cctx_t* cctx, DEBUGLOG(4, "LZ4F_compressUpdate_MT (srcSize=%u)", (unsigned)srcSize); RETURN_ERROR_IF(cctx->cStage != 1, compressionState_uninitialized); + { size_t const r = LZ4F_emitPendingWhole(cctx, dst, dstCapacity); /* left by an incomplete LZ4F_flush() */ + FORWARD_IF_ERROR_MT(r, cctx); + pos += r; + } { size_t const r = LZ4F_mt_emitUncTail(cctx, dst, dstCapacity, pos); /* switch from uncompressed blocks */ FORWARD_IF_ERROR_MT(r, cctx); pos += r; @@ -2152,9 +2330,12 @@ static size_t LZ4F_uncompressedUpdate_MT(LZ4F_cctx_t* cctx, size_t pos = 0; RETURN_ERROR_IF(cctx->cStage != 1, compressionState_uninitialized); RETURN_ERROR_IF(cctx->prefs.frameInfo.blockMode != LZ4F_blockIndependent, blockMode_invalid); + pos = LZ4F_emitPendingWhole(cctx, dst, dstCapacity); /* left by an incomplete LZ4F_flush() */ + FORWARD_IF_ERROR_MT(pos, cctx); if (mt->nbSubmitted || mt->filling) { /* switch from compressed blocks */ - pos = LZ4F_mt_flush(cctx, dst, dstCapacity, 0); - FORWARD_IF_ERROR_MT(pos, cctx); + size_t const r = LZ4F_mt_flush(cctx, dst, dstCapacity, pos); + FORWARD_IF_ERROR_MT(r, cctx); + pos += r; } mt->stableEnd = NULL; /* next input doesn't follow previous stable input anymore */ if (cctx->prefs.frameInfo.contentChecksumFlag == LZ4F_contentChecksumEnabled) @@ -2234,9 +2415,14 @@ static size_t LZ4F_compressUpdateImpl(LZ4F_cctx* cctxPtr, RETURN_ERROR_IF(cctxPtr->cStage != 1, compressionState_uninitialized); /* state must be initialized and waiting for next block */ + /* output left pending by an incomplete LZ4F_flush() */ + bytesWritten = LZ4F_emitPendingWhole(cctxPtr, dstPtr, dstCapacity); + FORWARD_IF_ERROR_INVALIDATE(bytesWritten, cctxPtr); + dstPtr += bytesWritten; + /* flush currently written block, to continue with new block compression */ if (cctxPtr->blockCompressMode != blockCompression) { - bytesWritten = LZ4F_flush(cctxPtr, dstBuffer, dstCapacity, compressOptionsPtr); + bytesWritten = LZ4F_flushBlock(cctxPtr, dstPtr, (size_t)(dstEnd - dstPtr)); FORWARD_IF_ERROR(bytesWritten); /* note : a failed flush invalidates the frame */ dstPtr += bytesWritten; cctxPtr->blockCompressMode = blockCompression; @@ -2372,54 +2558,78 @@ size_t LZ4F_uncompressedUpdate(LZ4F_cctx* cctxPtr, } +/* emits pending output into @dst + @pos, as much as fits : block or footer (pending), then jobs (multi-threaded mode). + * @return : new position, or an error code (the frame is then invalidated) */ +static size_t LZ4F_emitPending(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity, size_t pos) +{ + pos += LZ4F_copyPending(cctx, dst, dstCapacity, pos); + if (cctx->mtActive && (cctx->pendingSize == 0)) { + size_t const r = LZ4F_mt_emitPartial(cctx->mt, dst, dstCapacity, pos); + FORWARD_IF_ERROR_MT(r, cctx); + pos += r; + } + return pos; +} + +static int LZ4F_outputPending(const LZ4F_cctx_t* cctx) +{ + return (cctx->pendingSize > 0) || (cctx->mtActive && (cctx->mt->nbSubmitted > 0)); +} + +/* @return : 1 if LZ4F_flush() has output to produce */ +static int LZ4F_somethingToFlush(const LZ4F_cctx_t* cctx) +{ + if (LZ4F_outputPending(cctx)) return 1; + if (cctx->mtActive) return (cctx->mt->uncTailSize > 0) || cctx->mt->filling; + return cctx->tmpInSize > 0; +} + +/* flush point : input retained within cctx is compressed, and no longer read. + * @return : nb of bytes written into @dst, or an error code (the frame is then invalidated) */ +static size_t LZ4F_flushPoint(LZ4F_cctx_t* cctx, BYTE* dst, size_t dstCapacity) +{ + size_t const r = cctx->mtActive ? LZ4F_mt_flushPoint(cctx, dst, dstCapacity) + : LZ4F_flushBlockPartial(cctx, dst, dstCapacity); + if (!LZ4F_isError(r)) cctx->flushStage = 1; + return r; +} + /*! LZ4F_flush() : * When compressed data must be sent immediately, without waiting for a block to be filled, * invoke LZ4F_flush(), which will immediately compress any remaining data stored within LZ4F_cctx. + * The invocation fills @dstBuffer, or completes the flush : + * as long as it returns @dstCapacity, output may remain : invoke LZ4F_flush() again. + * The first invocation sets the flush point : input retained within cctx is compressed (all jobs completed). * The result of the function is the number of bytes written into dstBuffer. * It can be zero, this means there was no data left within LZ4F_cctx. * The function outputs an error code if it fails (can be tested using LZ4F_isError()) - * Success is guaranteed when dstCapacity >= LZ4F_compressBound(0, preferencesPtr). - * Otherwise, it fails with dstCapacity_tooSmall if its output doesn't fit, - * in which case the frame is invalidated, and must be restarted with LZ4F_compressBegin(). - * LZ4F_compressOptions_t* is optional. NULL is a valid argument. + * A single invocation completes the flush when dstCapacity >= LZ4F_compressBound(0, preferencesPtr). + * dstCapacity == 0 is rejected (dstCapacity_tooSmall) when there is something to flush, without modifying the frame. + * LZ4F_compressOptions_t* is optional : NULL is a valid argument. */ size_t LZ4F_flush(LZ4F_cctx* cctxPtr, void* dstBuffer, size_t dstCapacity, const LZ4F_compressOptions_t* compressOptionsPtr) { - compressFunc_t compress; - size_t cSize; - - DEBUGLOG(5, "LZ4F_flush: %zu buffered bytes (saved dict size = %i) (dstCapacity=%u)", - cctxPtr->tmpInSize, (int)(cctxPtr->tmpIn - cctxPtr->tmpBuff), (unsigned)dstCapacity); - RETURN_ERROR_IF(cctxPtr->cStage != 1, compressionState_uninitialized); - if (cctxPtr->mtActive) return LZ4F_mt_flush(cctxPtr, (BYTE*)dstBuffer, dstCapacity, 0); - if (cctxPtr->tmpInSize == 0) return 0; /* nothing to flush */ + BYTE* const dst = (BYTE*)dstBuffer; + size_t pos = 0; + DEBUGLOG(5, "LZ4F_flush (dstCapacity=%u)", (unsigned)dstCapacity); (void)compressOptionsPtr; /* not useful (yet) */ - - /* select compression function */ - compress = LZ4F_selectCompression(cctxPtr->prefs.frameInfo.blockMode, cctxPtr->prefs.compressionLevel, cctxPtr->blockCompressMode); - - /* compress tmp buffer */ - cSize = LZ4F_makeBlock(dstBuffer, dstCapacity, - cctxPtr->tmpIn, cctxPtr->tmpInSize, - compress, cctxPtr->lz4CtxPtr, cctxPtr->prefs.compressionLevel, - cctxPtr->cdict, - cctxPtr->prefs.frameInfo.blockChecksumFlag); - FORWARD_IF_ERROR_INVALIDATE(cSize, cctxPtr); - assert(((void)"flush overflows dstBuffer!", cSize <= dstCapacity)); - - if (cctxPtr->prefs.frameInfo.blockMode == LZ4F_blockLinked) - cctxPtr->tmpIn += cctxPtr->tmpInSize; - cctxPtr->tmpInSize = 0; - - /* keep tmpIn within limits */ - if ((cctxPtr->tmpIn + cctxPtr->maxBlockSize) > (cctxPtr->tmpBuff + cctxPtr->tmpBuffCapacity)) { - assert(cctxPtr->prefs.frameInfo.blockMode == LZ4F_blockLinked); - LZ4F_localSaveDict(cctxPtr); + RETURN_ERROR_IF(cctxPtr->cStage != 1, compressionState_uninitialized); + if (dstCapacity == 0) { + /* no progress possible : rejected, the frame continues */ + RETURN_ERROR_IF(LZ4F_somethingToFlush(cctxPtr), dstCapacity_tooSmall); + cctxPtr->flushStage = 0; + return 0; } - - return cSize; + if (cctxPtr->flushStage == 0) { + pos = LZ4F_flushPoint(cctxPtr, dst, dstCapacity); + FORWARD_IF_ERROR(pos); /* already invalidated */ + } + pos = LZ4F_emitPending(cctxPtr, dst, dstCapacity, pos); + FORWARD_IF_ERROR(pos); /* already invalidated */ + if (pos < dstCapacity) cctxPtr->flushStage = 0; /* completed */ + return pos; } @@ -2427,48 +2637,67 @@ size_t LZ4F_flush(LZ4F_cctx* cctxPtr, * When you want to properly finish the compressed frame, just call LZ4F_compressEnd(). * It will flush whatever data remained within compressionContext (like LZ4_flush()) * but also properly finalize the frame, with an endMark and an (optional) checksum. + * The invocation fills @dstBuffer, or completes the frame : + * as long as it returns @dstCapacity, output may remain : invoke LZ4F_compressEnd() again. * LZ4F_compressOptions_t structure is optional : you can provide NULL as argument. - * @return: the number of bytes written into dstBuffer (necessarily >= 4 (endMark size)) + * @return: the number of bytes written into dstBuffer (the last invocation writes the endMark, unless already written) * or an error code if it fails (can be tested using LZ4F_isError()) - * The context can then be used again to compress a new frame, starting with LZ4F_compressBegin(). + * Once completed, the context can be used again to compress a new frame, starting with LZ4F_compressBegin(). */ size_t LZ4F_compressEnd(LZ4F_cctx* cctxPtr, void* dstBuffer, size_t dstCapacity, const LZ4F_compressOptions_t* compressOptionsPtr) { - BYTE* const dstStart = (BYTE*)dstBuffer; - BYTE* dstPtr = dstStart; - - size_t const flushSize = LZ4F_flush(cctxPtr, dstBuffer, dstCapacity, compressOptionsPtr); + BYTE* const dst = (BYTE*)dstBuffer; + size_t pos = 0; DEBUGLOG(5,"LZ4F_compressEnd: dstCapacity=%u", (unsigned)dstCapacity); - FORWARD_IF_ERROR(flushSize); /* note : a failed flush invalidates the frame */ - dstPtr += flushSize; + (void)compressOptionsPtr; /* not useful (yet) */ + RETURN_ERROR_IF((cctxPtr->cStage != 1) && (cctxPtr->cStage != 2), compressionState_uninitialized); + RETURN_ERROR_IF(dstCapacity == 0, dstCapacity_tooSmall); /* no progress possible : rejected, the frame continues */ - assert(flushSize <= dstCapacity); - dstCapacity -= flushSize; - - /* From now on, the frame is either completed, or failed : it can't be continued. - * State is re-usable for a new frame (with identical preferences) */ - cctxPtr->cStage = 0; - - RETURN_ERROR_IF(dstCapacity < 4, dstCapacity_tooSmall); - LZ4F_writeLE32(dstPtr, 0); - dstPtr += 4; /* endMark */ - - if (cctxPtr->prefs.frameInfo.contentChecksumFlag == LZ4F_contentChecksumEnabled) { - U32 const xxh = XXH32_digest(&(cctxPtr->xxh)); - RETURN_ERROR_IF(dstCapacity < 8, dstCapacity_tooSmall); - DEBUGLOG(5,"Writing 32-bit content checksum (0x%0X)", xxh); - LZ4F_writeLE32(dstPtr, xxh); - dstPtr+=4; /* content Checksum */ - } - - if (cctxPtr->prefs.frameInfo.contentSize) { - if (cctxPtr->prefs.frameInfo.contentSize != cctxPtr->totalInSize) + if (cctxPtr->cStage == 1) { + /* end of frame : no more input */ + if (cctxPtr->prefs.frameInfo.contentSize && (cctxPtr->prefs.frameInfo.contentSize != cctxPtr->totalInSize)) { + if (cctxPtr->mtActive) LZ4F_mt_abort(cctxPtr->mt); + cctxPtr->cStage = 0; RETURN_ERROR(frameSize_wrong); + } + cctxPtr->cStage = 2; + if (cctxPtr->flushStage == 0) { + pos = LZ4F_flushPoint(cctxPtr, dst, dstCapacity); + FORWARD_IF_ERROR(pos); /* already invalidated */ + } } + + if (cctxPtr->flushStage == 1) { + pos = LZ4F_emitPending(cctxPtr, dst, dstCapacity, pos); + FORWARD_IF_ERROR(pos); /* already invalidated */ + if (LZ4F_outputPending(cctxPtr)) return pos; /* dstBuffer is full */ + /* footer : endMark, then optional content checksum */ + LZ4F_writeLE32(cctxPtr->footer, 0); + cctxPtr->pendingSize = 4; + if (cctxPtr->prefs.frameInfo.contentChecksumFlag == LZ4F_contentChecksumEnabled) { + U32 const xxh = XXH32_digest(&(cctxPtr->xxh)); + DEBUGLOG(5,"Writing 32-bit content checksum (0x%0X)", xxh); + LZ4F_writeLE32(cctxPtr->footer + 4, xxh); + cctxPtr->pendingSize = 8; + } + cctxPtr->pending = cctxPtr->footer; + cctxPtr->flushStage = 2; } - return (size_t)(dstPtr - dstStart); + if (cctxPtr->flushStage == 2) { + pos += LZ4F_copyPending(cctxPtr, dst, dstCapacity, pos); + if (cctxPtr->pendingSize) return pos; /* dstBuffer is full */ + cctxPtr->flushStage = 3; + } + + assert(cctxPtr->flushStage == 3); + if (pos < dstCapacity) { + /* completed : the context is available for a new frame */ + cctxPtr->cStage = 0; + cctxPtr->flushStage = 0; + } /* otherwise, completed, but dstBuffer is full : next invocation returns 0 */ + return pos; } diff --git a/lib/lz4frame.h b/lib/lz4frame.h index 2073a285..ed7a3e92 100644 --- a/lib/lz4frame.h +++ b/lib/lz4frame.h @@ -214,7 +214,8 @@ typedef struct { * - With linked blocks, each 4 MB job restarts with previous 64 KB as dictionary : * output can differ (slightly) from nbWorkers == 0, at fast levels. * - Streaming (LZ4F_compressUpdate()) : output lags behind input, so LZ4F_compressBound() is larger, - * covering jobs in flight (~ 2 * nbWorkers jobs of 4 MB) ; LZ4F_compressEnd() emits all of them. + * covering jobs in flight (~ 2 * nbWorkers jobs of 4 MB), emitted by LZ4F_flush() and LZ4F_compressEnd(), + * possibly in several invocations with a smaller dstCapacity. * Without autoFlush, the amount emitted by each invocation depends on scheduling : * only the whole frame is deterministic. Completed jobs are only emitted if they fit : * with a smaller dstCapacity, LZ4F_compressUpdate() only fails if output must be emitted (ex: all job slots busy). @@ -369,7 +370,8 @@ LZ4FLIB_API size_t LZ4F_compressBegin(LZ4F_cctx* cctx, * Provides minimum dstCapacity required to guarantee success of * LZ4F_compressUpdate() or LZ4F_uncompressedUpdate(), given a srcSize and preferences, for a worst case scenario. * Smaller capacities are allowed, but then the operation may fail. - * When srcSize==0, LZ4F_compressBound() provides an upper bound for LZ4F_flush() and LZ4F_compressEnd() instead. + * When srcSize==0, LZ4F_compressBound() provides an upper bound for LZ4F_flush() and LZ4F_compressEnd() instead : + * their output remains strictly below it, so that a single invocation completes them. * Note that the result is only valid for a single invocation of LZ4F_compressUpdate(). * When invoking LZ4F_compressUpdate() multiple times, * if the output buffer is gradually filled up instead of emptied and re-used from its start, @@ -406,12 +408,15 @@ LZ4FLIB_API size_t LZ4F_compressUpdate(LZ4F_cctx* cctx, /*! LZ4F_flush() : * When data must be generated and sent immediately, without waiting for a block to be completely filled, * it's possible to call LZ4_flush(). It will immediately compress any data buffered within cctx. + * Each invocation fills dstBuffer, or completes the flush (v1.10.1+) : + * while it returns dstCapacity, output may remain : invoke LZ4F_flush() again, with any dstCapacity. + * A single invocation completes the flush when dstCapacity >= LZ4F_compressBound(0, prefsPtr). + * The first invocation compresses buffered input : after it returns, input is no longer read + * (see LZ4F_srcStableUntilFlush). LZ4F_compressUpdate() may follow an incomplete flush : remaining output comes first. * `cOptPtr` is optional : it's possible to provide NULL, all options will be set to default. * @return : nb of bytes written into dstBuffer (can be zero, when there is no data stored within cctx) * or an error code if it fails (which can be tested using LZ4F_isError()) - * Note : LZ4F_flush() is guaranteed to be successful when dstCapacity >= LZ4F_compressBound(0, prefsPtr). - * A smaller dstCapacity is allowed : it fails with LZ4F_ERROR_dstCapacity_tooSmall if the output doesn't fit, - * in which case the frame is invalidated (see LZ4F_compressUpdate()). + * dstCapacity == 0 is rejected (LZ4F_ERROR_dstCapacity_tooSmall) when output remains, without modifying the frame. * LZ4F_flush() requires a frame in progress. */ LZ4FLIB_API size_t LZ4F_flush(LZ4F_cctx* cctx, @@ -422,13 +427,15 @@ LZ4FLIB_API size_t LZ4F_flush(LZ4F_cctx* cctx, * To properly finish an LZ4 frame, invoke LZ4F_compressEnd(). * It will flush whatever data remained within `cctx` (like LZ4_flush()) * and properly finalize the frame, with an endMark and a checksum. + * Each invocation fills dstBuffer, or completes the frame (v1.10.1+) : + * while it returns dstCapacity, output may remain : invoke LZ4F_compressEnd() again, with any dstCapacity. + * A single invocation completes the frame when dstCapacity >= LZ4F_compressBound(0, prefsPtr). + * Until completion, only LZ4F_compressEnd() can continue the frame (LZ4F_compressBegin() abandons it). * `cOptPtr` is optional : NULL can be provided, in which case all options will be set to default. - * @return : nb of bytes written into dstBuffer, necessarily >= 4 (endMark), + * @return : nb of bytes written into dstBuffer, * or an error code if it fails (which can be tested using LZ4F_isError()) - * Note : LZ4F_compressEnd() is guaranteed to be successful when dstCapacity >= LZ4F_compressBound(0, prefsPtr). - * A smaller dstCapacity is allowed : it fails with LZ4F_ERROR_dstCapacity_tooSmall if the output doesn't fit, - * in which case the frame is invalidated (see LZ4F_compressUpdate()). - * Whether it succeeds or not, LZ4F_compressEnd() makes `cctx` available again for another compression task. + * dstCapacity == 0 is rejected (LZ4F_ERROR_dstCapacity_tooSmall), without modifying the frame. + * Once completed, or after an error, `cctx` is available again for another compression task. */ LZ4FLIB_API size_t LZ4F_compressEnd(LZ4F_cctx* cctx, void* dstBuffer, size_t dstCapacity, diff --git a/ossfuzz/compress_frame_stream_fuzzer.c b/ossfuzz/compress_frame_stream_fuzzer.c index 64b7ebce..bd6b7a22 100644 --- a/ossfuzz/compress_frame_stream_fuzzer.c +++ b/ossfuzz/compress_frame_stream_fuzzer.c @@ -1,10 +1,12 @@ /** * This fuzz target compresses the fuzzed data with the LZ4F streaming API, * using a random sequence of operations, into random output capacities. - * Capacities may be smaller than LZ4F_compressBound() : each operation must - * succeed if and only if its output fits, and then produce the same output + * Capacities may be smaller than LZ4F_compressBound() : begin and updates must + * succeed if and only if their output fits, and then produce the same output * as a reference context which always receives a bound-sized buffer. * A failed operation invalidates the frame. + * Flush and end fill dstBuffer, or complete : they are invoked again while they fill it, + * producing the same output in pieces. Capacity 0 is rejected when output remains. * Completed frames must round trip correctly. * Note : output of linked blocks at fast levels may depend on context history. * Hence, after a failure, both contexts restart from a fresh state. @@ -93,6 +95,39 @@ static void checkInvalidated(LZ4F_cctx* cctx) "A failed operation must invalidate the frame"); } +/* Flush and end fill dstBuffer, or complete : invoked again, with other capacities, while they fill it. + * Each invocation receives exactly its capacity, to detect overflows. + * @return : total output, written into @out */ +static size_t runPieces(const state_t* s, opType_e type, char* out, size_t outCapacity, size_t capacity, uint32_t* seed) +{ + size_t pos = 0; + for (;;) { + char* const dst = (char*)malloc(capacity ? capacity : 1); + size_t r; + FUZZ_ASSERT(dst != NULL); + r = runOp(s, s->cctx, s->prefs, type, dst, capacity, NULL, 0); + if (capacity == 0) { + free(dst); + if (!LZ4F_isError(r)) { + FUZZ_ASSERT_MSG((type == op_flush) && (r == 0), "Only an empty flush succeeds into 0 byte"); + return pos; + } + FUZZ_ASSERT_MSG(LZ4F_getErrorCode(r) == LZ4F_ERROR_dstCapacity_tooSmall, + "Capacity 0 must be rejected when output remains"); + capacity = FUZZ_rand32(seed, 1, 256); + continue; + } + FUZZ_ASSERT_MSG(!LZ4F_isError(r), "Flush and end must not fail for lack of capacity"); + FUZZ_ASSERT(r <= capacity); + FUZZ_ASSERT(pos + r <= outCapacity); + memcpy(out + pos, dst, r); + pos += r; + free(dst); + if (r < capacity) return pos; + capacity = FUZZ_rand32(seed, 1, 4096); + } +} + /* Runs @type on both contexts, and appends its output to @s->frame. * @return : 1 on success, 0 if the operation failed because capacity was too small */ static int runBoth(state_t* s, opType_e type, const uint8_t* src, size_t srcSize, uint32_t* seed) @@ -106,6 +141,10 @@ static int runBoth(state_t* s, opType_e type, const uint8_t* src, size_t srcSize FUZZ_ASSERT_MSG(!LZ4F_isError(refSize), "Operation must succeed with a bound-sized buffer"); FUZZ_ASSERT(refSize <= refBound); + if ((type == op_flush) || (type == op_end)) { + FUZZ_ASSERT_MSG(refSize < refBound, "Output must remain strictly below LZ4F_compressBound(0)"); + } + if (s->lagging) { FUZZ_ASSERT(s->refFrameSize + refSize <= s->frameCapacity); memcpy(s->refFrame + s->refFrameSize, s->refOut, refSize); @@ -116,6 +155,10 @@ static int runBoth(state_t* s, opType_e type, const uint8_t* src, size_t srcSize case 2: capacity = FUZZ_rand32(seed, 0, (uint32_t)bound); break; default: capacity = FUZZ_rand32(seed, 0, (uint32_t)MIN(bound, 256)); break; } + if ((type == op_flush) || (type == op_end)) { + s->frameSize += runPieces(s, type, s->frame + s->frameSize, s->frameCapacity - s->frameSize, capacity, seed); + return 1; + } char* const dst = (char*)malloc(capacity ? capacity : 1); FUZZ_ASSERT(dst != NULL); size_t const r = runOp(s, s->cctx, s->prefs, type, dst, capacity, src, srcSize); @@ -146,6 +189,14 @@ static int runBoth(state_t* s, opType_e type, const uint8_t* src, size_t srcSize default: capacity = FUZZ_rand32(seed, 0, (uint32_t)bound); break; } + if ((type == op_flush) || (type == op_end)) { + size_t const total = runPieces(s, type, s->frame + s->frameSize, s->frameCapacity - s->frameSize, capacity, seed); + FUZZ_ASSERT_MSG(total == refSize, "Output in pieces must have the same size"); + FUZZ_ASSERT_MSG(!memcmp(s->frame + s->frameSize, s->refOut, refSize), "Output must not depend on capacity"); + s->frameSize += total; + return 1; + } + /* exact size, to detect overflows */ char* const dst = (char*)malloc(capacity ? capacity : 1); FUZZ_ASSERT(dst != NULL); diff --git a/programs/lz4io.c b/programs/lz4io.c index 51a469ff..6471ee8f 100644 --- a/programs/lz4io.c +++ b/programs/lz4io.c @@ -1540,9 +1540,11 @@ LZ4IO_compressFilename_extRess(unsigned long long* inStreamSize, U64 const fileSize = UTIL_getOpenFileSize(srcFile); /* 0 : unknown (ex: stdin) */ if (fileSize && (fileSize <= LZ4IO_jobSize(&prefs))) prefs.nbWorkers = 0; } - /* memory-mapped input is released (only a memory hint) behind jobs in flight : - * liblz4 keeps up to 2*nbWorkers+1 jobs of at most 4 MB in flight, each reading 64 KB of history before it */ - releaseLag = (2 * (size_t)prefs.nbWorkers + 1) * (4 MB) + 64 KB; + /* memory-mapped input is released (only a memory hint) behind input still read by liblz4 : + * jobs in flight, and the 64 KB of history before them. + * LZ4F_compressBound(0) bounds the output of LZ4F_compressEnd(), which emits all input retained within cctx : + * since incompressible input is stored as is, it also bounds the amount of retained input */ + releaseLag = LZ4F_compressBound(0, &prefs) + 64 KB; /* read first chunk */ readSize = LZ4IO_readInput(&input, chunkSize, &chunk); diff --git a/tests/frameMT.c b/tests/frameMT.c index 1a9f8026..b5e7ef4b 100644 --- a/tests/frameMT.c +++ b/tests/frameMT.c @@ -35,8 +35,9 @@ * abandoned frames (jobs in flight) are handled * - with independent blocks, output is identical to nbWorkers == 0 * - serial processing (dstCapacity too small for parallel jobs) produces the same output - * - destination too small : the frame is invalidated (no truncated frame), and the context remains usable ; - * completed jobs not fitting yet are emitted later, without failure + * - destination too small : compressUpdate fails and invalidates the frame (no truncated frame), the context remains usable ; + * completed jobs not fitting yet are emitted later, without failure ; flush and end produce output in pieces + * - flush and end in pieces, for any capacity : same output ; flush point, exact fill, update after an incomplete flush * - LZ4F_compressBegin() rejecting its arguments leaves the frame in progress (jobs in flight) untouched ; * its allocation failures invalidate it * - LZ4F_getMaxNbWorkers() reports multi-threading support @@ -186,6 +187,7 @@ static int validate(XXH64_state_t* digest) unsigned seed = 1; size_t si, li; int linked; + g_step = "one-shot"; CHECK(src && dict && ref0 && ref1 && out && decoded && cctx, "allocation failed"); RDG_genBuffer(src, srcSizeMax, 0.5, 0.0, 7); @@ -376,6 +378,7 @@ static size_t streamCompress(LZ4F_cctx* cctx, char* dst, if (mode.flushes && ((r >> 16) % 4 == 0)) { size_t const fSize = LZ4F_flush(cctx, tmpOut, LZ4F_compressBound(0, &p), NULL); CHECK_LZ4F(fSize); + CHECK(fSize < LZ4F_compressBound(0, &p), "flush fills LZ4F_compressBound(0) : not completed"); memcpy(dst + pos, tmpOut, fSize); pos += fSize; flushed = 1; @@ -384,6 +387,7 @@ static size_t streamCompress(LZ4F_cctx* cctx, char* dst, } { size_t const eSize = LZ4F_compressEnd(cctx, tmpOut, LZ4F_compressBound(0, &p), NULL); CHECK_LZ4F(eSize); + CHECK(eSize < LZ4F_compressBound(0, &p), "compressEnd fills LZ4F_compressBound(0) : not completed"); memcpy(dst + pos, tmpOut, eSize); pos += eSize; } @@ -744,7 +748,8 @@ static int validateAllocFailures(void) } /* begin, update (whole input), flush, end : from operation @smallFrom onwards (0:update, 1:flush, 2:end ; 3:never), - * each operation only receives @capacity, instead of all remaining capacity. @smallFrom == 2 : no flush. + * each invocation only receives @capacity, instead of all remaining capacity. @smallFrom == 2 : no flush. + * LZ4F_flush() and LZ4F_compressEnd() fill dstBuffer, or complete : they are invoked again while they fill it. * @return : frame size, or the first error, *failedOp being then the failed operation */ static size_t streamOps(LZ4F_cctx* cctx, const LZ4F_preferences_t* prefs, char* dst, size_t dstCapacity, const char* src, size_t srcSize, @@ -755,22 +760,28 @@ static size_t streamOps(LZ4F_cctx* cctx, const LZ4F_preferences_t* prefs, *failedOp = -1; if (LZ4F_isError(pos)) return pos; for (op = 0; op < 3; op++) { - size_t const c = (op >= smallFrom) ? capacity : dstCapacity - pos; - size_t r; + int again; if ((op == 1) && (smallFrom == 2)) continue; - if (op == 0) r = LZ4F_compressUpdate(cctx, dst + pos, c, src, srcSize, NULL); - else if (op == 1) r = LZ4F_flush(cctx, dst + pos, c, NULL); - else r = LZ4F_compressEnd(cctx, dst + pos, c, NULL); - if (LZ4F_isError(r)) { *failedOp = op; return r; } - pos += r; + do { + size_t const c = (op >= smallFrom) ? capacity : dstCapacity - pos; + size_t r; + if (op == 0) r = LZ4F_compressUpdate(cctx, dst + pos, c, src, srcSize, NULL); + else if (op == 1) r = LZ4F_flush(cctx, dst + pos, c, NULL); + else r = LZ4F_compressEnd(cctx, dst + pos, c, NULL); + if (LZ4F_isError(r)) { *failedOp = op; return r; } + pos += r; + again = (op > 0) && (r == c); + } while (again); } return pos; } -/* Destination too small : the operation fails with dstCapacity_tooSmall, and invalidates the frame : - * next operations fail with compressionState_uninitialized (in particular, no end mark : no truncated frame), - * until LZ4F_compressBegin() starts a new frame, identical to the reference. Same contract as nbWorkers == 0. - * Without autoFlush, the failing operation depends on scheduling : the first one given too little capacity, or a later one. */ +/* Destination too small : + * - LZ4F_compressUpdate() must ingest all input : it fails with dstCapacity_tooSmall when output must be written + * (with autoFlush, a full job slot ring, or nbWorkers == 0), and invalidates the frame : + * next operations fail with compressionState_uninitialized (in particular, no end mark : no truncated frame), + * until LZ4F_compressBegin() starts a new frame, identical to the reference. Same contract as nbWorkers == 0. + * - LZ4F_flush() and LZ4F_compressEnd() fill dstBuffer, or complete : the frame is produced in pieces, identical to the reference. */ static int validateDstTooSmall(void) { static const unsigned nbWorkersList[] = { 0, 1, 3 }; /* 1 : job slots fill up within LZ4F_compressUpdate() */ @@ -781,7 +792,7 @@ static int validateDstTooSmall(void) char* const ref = (char*)malloc(dstCapacity); char* const out = (char*)malloc(dstCapacity); LZ4F_CustomMem const countingMem = { countingAlloc, NULL, countingFree, NULL }; - unsigned nbConfigs = 0, nbFailures = 0; + unsigned nbConfigs = 0, nbFailures = 0, nbPartial = 0; size_t wi; int linked, autoFlush, smallFrom; @@ -793,7 +804,8 @@ static int validateDstTooSmall(void) for (autoFlush = 0; autoFlush <= 1; autoFlush++) for (smallFrom = 0; smallFrom <= 2; smallFrom++) { LZ4F_preferences_t prefs = LZ4F_INIT_PREFERENCES; - int const mustFail = !autoFlush || (smallFrom == 0); /* otherwise, only the end mark remains to be written */ + /* compressUpdate() must write output into too little capacity ; otherwise, it can defer it (jobs) */ + int const mustFail = (smallFrom == 0) && (autoFlush || (nbWorkersList[wi] == 0)); LZ4F_cctx* cctx; size_t refSize, r; int failedOp; @@ -812,11 +824,13 @@ static int validateDstTooSmall(void) r = streamOps(cctx, &prefs, out, dstCapacity, src, srcSize, smallFrom, smallCapacity, &failedOp); if (!LZ4F_isError(r)) { CHECK(!mustFail, "%u workers, smallFrom=%i : should have failed", prefs.nbWorkers, smallFrom); - CHECK(r == refSize && !memcmp(out, ref, r), "%u workers : output differs", prefs.nbWorkers); + CHECK(r == refSize && !memcmp(out, ref, r), "%u workers, smallFrom=%i : output differs", prefs.nbWorkers, smallFrom); + nbPartial++; } else { CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_dstCapacity_tooSmall, "%u workers : unexpected error %s", prefs.nbWorkers, LZ4F_getErrorName(r)); - CHECK(failedOp >= smallFrom, "%u workers : failed with enough capacity", prefs.nbWorkers); + CHECK(failedOp == 0, "%u workers : %s failed", prefs.nbWorkers, (failedOp == 1) ? "flush" : "end"); + CHECK(smallFrom == 0, "%u workers : failed with enough capacity", prefs.nbWorkers); nbFailures++; /* frame invalidated */ r = LZ4F_compressUpdate(cctx, out, dstCapacity, src, 1 KB, NULL); @@ -836,11 +850,220 @@ static int validateDstTooSmall(void) CHECK(g_nbAllocs == g_nbFrees, "memory leak (%u allocations, %u frees)", (unsigned)g_nbAllocs, (unsigned)g_nbFrees); nbConfigs++; } - DISPLAY("dstCapacity too small : %u configurations validated, %u failures \n", nbConfigs, nbFailures); + DISPLAY("dstCapacity too small : %u configurations validated, %u compressUpdate failures, %u frames in pieces \n", + nbConfigs, nbFailures, nbPartial); free(src); free(ref); free(out); return 0; } +/* invokes LZ4F_flush() (@end == 0) or LZ4F_compressEnd() with @capacity, until it doesn't fill it, + * or up to @maxCalls invocations (0 : no limit). + * @return : nb of bytes written into @dst, or an error code */ +static size_t flushPieces(LZ4F_cctx* cctx, int end, char* dst, size_t capacity, unsigned maxCalls, unsigned* nbCalls) +{ + size_t pos = 0; + unsigned n = 0; + if (nbCalls) *nbCalls = 0; + for (;;) { + size_t const r = end ? LZ4F_compressEnd(cctx, dst + pos, capacity, NULL) + : LZ4F_flush(cctx, dst + pos, capacity, NULL); + if (LZ4F_isError(r)) return r; + pos += r; + n++; + if (r < capacity) break; + if (maxCalls && (n >= maxCalls)) break; + } + if (nbCalls) *nbCalls = n; + return pos; +} + +/* reference : begin, update(@part1), flush, then (@uncSize > 0) uncompressedUpdate(@uncSize), update(rest), end, + * each invocation receiving its bound. Position and size of flush output are reported. */ +static size_t partialRef(LZ4F_cctx* cctx, const LZ4F_preferences_t* p, const LZ4F_compressOptions_t* cOpts, + char* dst, const char* src, size_t srcSize, + size_t part1, size_t uncSize, size_t* flushPos, size_t* flushSize) +{ + size_t pos, r; + pos = LZ4F_compressBegin(cctx, dst, LZ4F_HEADER_SIZE_MAX, p); + CHECK_LZ4F(pos); + r = LZ4F_compressUpdate(cctx, dst + pos, LZ4F_compressBound(part1, p), src, part1, cOpts); + CHECK_LZ4F(r); pos += r; + *flushPos = pos; + r = LZ4F_flush(cctx, dst + pos, LZ4F_compressBound(0, p), NULL); + CHECK_LZ4F(r); pos += r; + *flushSize = r; + if (uncSize) { + r = LZ4F_uncompressedUpdate(cctx, dst + pos, LZ4F_compressBound(uncSize, p), src + part1, uncSize, cOpts); + CHECK_LZ4F(r); pos += r; + } + r = LZ4F_compressUpdate(cctx, dst + pos, LZ4F_compressBound(srcSize - part1 - uncSize, p), + src + part1 + uncSize, srcSize - part1 - uncSize, cOpts); + CHECK_LZ4F(r); pos += r; + r = LZ4F_compressEnd(cctx, dst + pos, LZ4F_compressBound(0, p), NULL); + CHECK_LZ4F(r); + return pos + r; +} + +/* LZ4F_flush() and LZ4F_compressEnd() fill dstBuffer, or complete : + * output produced in pieces is identical to the reference, for any capacity, even 1 byte. + * - the first invocation of a flush sets the flush point : input is no longer read (LZ4F_srcStableUntilFlush) ; + * - capacity exactly equal to the remaining output : the next invocation returns 0 ; + * - compressUpdate() or uncompressedUpdate() after an incomplete flush first emit the remaining output ; + * - compressBegin() abandons an incomplete end ; + * - capacity 0 is rejected when output remains, without modifying the frame. */ +static int validatePartialFlush(void) +{ + static const unsigned nbWorkersList[] = { 0, 1, 3 }; + static const size_t capacities[] = { 1, 7, 4 KB + 1, 1 MB - 3 }; + size_t const srcSize = 9 MB + 123; + size_t const part1 = 5 MB + 7; /* jobs in flight at flush point */ + size_t const uncSize = 64 KB + 5; + size_t const dstCapacity = 2 * LZ4F_compressFrameBound(srcSize, NULL) + 64 MB; + char* const src = (char*)malloc(srcSize); + char* const work = (char*)malloc(srcSize); + char* const ref = (char*)malloc(dstCapacity); + char* const refFull = (char*)malloc(dstCapacity); + char* const out = (char*)malloc(dstCapacity); + LZ4F_CustomMem const countingMem = { countingAlloc, NULL, countingFree, NULL }; + unsigned nbConfigs = 0, nbCalls = 0; + size_t wi, ci; + int linked, variant; + + g_step = "partial flush"; + CHECK(src && work && ref && refFull && out, "allocation failed"); + RDG_genBuffer(src, srcSize, 0.5, 0.0, 31); + for (wi = 0; wi < sizeof(nbWorkersList)/sizeof(nbWorkersList[0]); wi++) + for (linked = 0; linked <= 1; linked++) { + LZ4F_preferences_t p = LZ4F_INIT_PREFERENCES; + LZ4F_cctx* const cctx = LZ4F_createCompressionContext_advanced(countingMem, LZ4F_VERSION); + size_t const unc = linked ? 0 : uncSize; /* uncompressed blocks : independent blocks only */ + size_t refSize, refFullSize, flushPos, flushSize; + CHECK(cctx != NULL, "allocation failed"); + p.nbWorkers = nbWorkersList[wi]; + p.frameInfo.blockMode = linked ? LZ4F_blockLinked : LZ4F_blockIndependent; + p.frameInfo.blockChecksumFlag = (LZ4F_blockChecksum_t)linked; + p.frameInfo.contentChecksumFlag = LZ4F_contentChecksumEnabled; + { LZ4F_compressOptions_t refOpts; + size_t r; + memset(&refOpts, 0, sizeof(refOpts)); + refOpts.stableSrc = LZ4F_srcStableUntilFlush; /* single-threaded linked output depends on stableSrc */ + refSize = partialRef(cctx, &p, &refOpts, ref, src, srcSize, part1, unc, &flushPos, &flushSize); + refFullSize = LZ4F_compressBegin(cctx, refFull, LZ4F_HEADER_SIZE_MAX, &p); + CHECK_LZ4F(refFullSize); + r = LZ4F_compressUpdate(cctx, refFull + refFullSize, LZ4F_compressBound(srcSize, &p), src, srcSize, NULL); + CHECK_LZ4F(r); refFullSize += r; + r = LZ4F_compressEnd(cctx, refFull + refFullSize, LZ4F_compressBound(0, &p), NULL); + CHECK_LZ4F(r); refFullSize += r; + } + + for (ci = 0; ci < sizeof(capacities)/sizeof(capacities[0]); ci++) + for (variant = 0; variant < 5; variant++) { + size_t const cap = capacities[ci]; + LZ4F_compressOptions_t cOpts; + size_t pos, r, flushRemaining; + unsigned n = 0; + if (g_quick && (ci == 1)) continue; + if ((variant >= 2) && (ci != 2)) continue; /* one capacity is enough */ + memset(&cOpts, 0, sizeof(cOpts)); + cOpts.stableSrc = LZ4F_srcStableUntilFlush; + memcpy(work, src, srcSize); + pos = LZ4F_compressBegin(cctx, out, LZ4F_HEADER_SIZE_MAX, &p); + CHECK_LZ4F(pos); + r = LZ4F_compressUpdate(cctx, out + pos, LZ4F_compressBound(part1, &p), work, part1, &cOpts); + CHECK_LZ4F(r); pos += r; + /* output split between update and flush depends on scheduling (jobs) */ + CHECK((pos <= flushPos + flushSize) && !memcmp(out, ref, pos), "update output differs"); + flushRemaining = flushPos + flushSize - pos; + if (variant == 4) { + r = LZ4F_flush(cctx, out + pos, 0, NULL); + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_dstCapacity_tooSmall, "flush into 0 byte : %s", + LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should have been rejected"); + } + if (variant == 1) { + /* capacity exactly equal to remaining output */ + r = LZ4F_flush(cctx, out + pos, flushRemaining, NULL); + CHECK(r == flushRemaining, "flush into exact capacity : %s", LZ4F_isError(r) ? LZ4F_getErrorName(r) : "wrong size"); + r = LZ4F_flush(cctx, out + pos + flushRemaining, cap, NULL); + CHECK(r == 0, "flush after exact fill : %s", LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should be empty"); + r = flushRemaining; + } else { + r = flushPieces(cctx, 0, out + pos, cap, (variant == 2) ? 1 : 0, &n); + CHECK_LZ4F(r); + nbCalls += n; + } + memset(work, 0xA5, part1 - 64 KB); /* flush point passed : input no longer read, except the last 64 KB (history) */ + if (variant == 2) { + /* update before the flush completes : remaining flush output comes first */ + size_t const flushed = r; + CHECK(flushed <= flushRemaining, "flush output beyond reference"); + if (unc) { + r = LZ4F_uncompressedUpdate(cctx, out + pos + flushed, LZ4F_compressBound(unc, &p), work + part1, unc, &cOpts); + CHECK_LZ4F(r); + r += flushed; + /* uncompressed blocks follow all compressed ones : no more output pending */ + } else { + r = flushed; + } + } else { + CHECK(r == flushRemaining && !memcmp(out + pos, ref + pos, r), "%u workers, capacity %u : flush output differs", + p.nbWorkers, (unsigned)cap); + if (unc) { + size_t const u = LZ4F_uncompressedUpdate(cctx, out + pos + r, LZ4F_compressBound(unc, &p), work + part1, unc, &cOpts); + CHECK_LZ4F(u); + r += u; + } + } + pos += r; + r = LZ4F_compressUpdate(cctx, out + pos, LZ4F_compressBound(srcSize - part1 - unc, &p), + work + part1 + unc, srcSize - part1 - unc, &cOpts); + CHECK_LZ4F(r); pos += r; + if (variant == 3) { + /* compressBegin() abandons an incomplete end */ + r = flushPieces(cctx, 1, out + pos, cap, 1, NULL); + CHECK_LZ4F(r); + pos = LZ4F_compressBegin(cctx, out, LZ4F_HEADER_SIZE_MAX, &p); + CHECK_LZ4F(pos); + r = LZ4F_compressUpdate(cctx, out + pos, LZ4F_compressBound(srcSize, &p), src, srcSize, NULL); + CHECK_LZ4F(r); pos += r; + r = LZ4F_compressEnd(cctx, out + pos, LZ4F_compressBound(0, &p), NULL); + CHECK_LZ4F(r); pos += r; + CHECK(pos == refFullSize && !memcmp(out, refFull, pos), "%u workers : frame after abandoned end differs", p.nbWorkers); + nbConfigs++; + continue; + } + if (variant == 4) { + r = LZ4F_compressEnd(cctx, out + pos, 0, NULL); + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_dstCapacity_tooSmall, "end into 0 byte : %s", + LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should have been rejected"); + } + if (variant == 1) { + size_t const endSize = refSize - pos; /* output split between update and end depends on scheduling */ + r = LZ4F_compressEnd(cctx, out + pos, endSize, NULL); + CHECK(r == endSize, "end into exact capacity : %s", LZ4F_isError(r) ? LZ4F_getErrorName(r) : "wrong size"); + r = LZ4F_compressEnd(cctx, out + pos + endSize, cap, NULL); + CHECK(r == 0, "end after exact fill : %s", LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should be empty"); + r = endSize; + } else { + r = flushPieces(cctx, 1, out + pos, cap, 0, &n); + CHECK_LZ4F(r); + nbCalls += n; + } + pos += r; + CHECK(pos == refSize && !memcmp(out, ref, pos), "%u workers, linked=%i, capacity %u, variant %i : frame differs", + p.nbWorkers, linked, (unsigned)cap, variant); + r = LZ4F_compressEnd(cctx, out, cap, NULL); /* completed : the frame is over */ + CHECK(LZ4F_getErrorCode(r) == LZ4F_ERROR_compressionState_uninitialized, "end after completed end : %s", + LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should have failed"); + nbConfigs++; + } + LZ4F_freeCompressionContext(cctx); + CHECK(g_nbAllocs == g_nbFrees, "memory leak (%u allocations, %u frees)", (unsigned)g_nbAllocs, (unsigned)g_nbFrees); + } + DISPLAY("partial flush : %u configurations validated (%u invocations) \n", nbConfigs, nbCalls); + free(src); free(work); free(ref); free(refFull); free(out); + return 0; +} + /* Without autoFlush, completed jobs are only emitted when they fit : * a dstCapacity too small for them doesn't fail, as long as no job must be emitted (free job slot). * Deterministic without worker threads (jobs complete within LZ4F_compressUpdate()), otherwise depends on scheduling. */ @@ -1365,6 +1588,7 @@ int main(int argc, const char** argv) validateFlushContract(); validateAllocFailures(); validateDstTooSmall(); + validatePartialFlush(); validateDeferredEmission(); validateBeginFailures(); validatePoolRetry(); diff --git a/tests/frametest.c b/tests/frametest.c index a2fc629e..1d64167d 100644 --- a/tests/frametest.c +++ b/tests/frametest.c @@ -1026,6 +1026,59 @@ static int sd_opExactOK(const sd_stream* s, const sd_op* op, BYTE* dst, size_t c return 1; } +/* sd_partialOpOK() : + * LZ4F_flush() and LZ4F_compressEnd() fill dstBuffer, or complete : + * runs @op into @capacity (< @refSize), then completes it with various capacities, + * sometimes exactly the remaining size (then, a last invocation returns 0). + * Concatenated output must be identical to reference. Capacity 0 is rejected, without modifying the frame. + * @return 1 if OK, 0 otherwise */ +static int sd_partialOpOK(const sd_stream* s, const sd_op* op, BYTE* dst, size_t capacity, + const BYTE* ref, size_t refSize, U32* rand) +{ + size_t pos = 0; + unsigned nbCalls; + for (nbCalls = 0; nbCalls < 1000; nbCalls++) { + size_t r; + sd_setCanary(dst + pos + capacity); + r = sd_runOp(s, op, dst + pos, capacity); + if (!sd_canaryIntact(dst + pos + capacity)) { + DISPLAY("%s into capacity %u : write beyond capacity \n", sd_opName(op->type), (unsigned)capacity); + return 0; + } + if (capacity == 0) { + if (!sd_isTooSmall(r)) { + DISPLAY("%s into capacity 0 : %s \n", sd_opName(op->type), + LZ4F_isError(r) ? LZ4F_getErrorName(r) : "should have been rejected"); + return 0; + } + } else { + if (LZ4F_isError(r) || (r > capacity)) { + DISPLAY("%s into capacity %u : %s \n", sd_opName(op->type), (unsigned)capacity, + LZ4F_isError(r) ? LZ4F_getErrorName(r) : "exceeds capacity"); + return 0; + } + pos += r; + if (pos > refSize) { + DISPLAY("%s : more output than reference (%u > %u) \n", sd_opName(op->type), (unsigned)pos, (unsigned)refSize); + return 0; + } + if (r < capacity) break; /* completed */ + } + { size_t const remaining = refSize - pos; + switch (FUZ_rand(rand) % 4) { + case 0: capacity = remaining; break; + case 1: capacity = 1 + FUZ_rand(rand) % (remaining + 1); break; + default: capacity = remaining + 1 + FUZ_rand(rand) % 16; break; + } + if (capacity == 0) capacity = 1; + } } + if ((pos != refSize) || memcmp(dst, ref, refSize)) { + DISPLAY("%s : partial output differs from reference \n", sd_opName(op->type)); + return 0; + } + return 1; +} + /* sd_runScript() : * runs all @ops, each one using its bound as capacity, writing the frame into @frame. * @opStarts and @opSizes are optional. @@ -1054,6 +1107,11 @@ static size_t sd_runScript(const sd_stream* s, const sd_op* ops, size_t nbOps, (unsigned)n, sd_opName(ops[n].type), (unsigned)ops[n].srcSize); return (size_t)-1; } + if (((ops[n].type == sd_flush) || (ops[n].type == sd_end)) && (r == bound)) { + /* filling dstBuffer would mean "invoke again" : output must remain strictly below LZ4F_compressBound(0) */ + DISPLAY("op %u (%s) : output reaches its bound \n", (unsigned)n, sd_opName(ops[n].type)); + return (size_t)-1; + } if (opStarts) opStarts[n] = pos; if (opSizes) opSizes[n] = r; pos += r; @@ -1166,6 +1224,15 @@ static int testSmallDstStream(U32 seed, unsigned nbConfigs) for (n = 0; n < k; n++) SD_CHECK(sd_opExactOK(&s, ops + n, dst, opSizes[n], frame + opStarts[n], opSizes[n]), "config %u, probe %u : replay op %u", configNb, probe, (unsigned)n); + if ((ops[k].type == sd_flush) || (ops[k].type == sd_end)) { + /* output in several invocations, the frame continues */ + SD_CHECK(sd_partialOpOK(&s, ops + k, dst, capacity, frame + opStarts[k], opSizes[k], &seed), + "config %u, probe %u : partial op %u", configNb, probe, (unsigned)k); + for (n = k + 1; n < nbOps; n++) + SD_CHECK(sd_opExactOK(&s, ops + n, dst, sd_opBound(&s, ops + n), frame + opStarts[n], opSizes[n]), + "config %u, probe %u : op %u after partial op %u", configNb, probe, (unsigned)n, (unsigned)k); + continue; + } sd_setCanary(dst + capacity); r = sd_runOp(&s, ops + k, dst, capacity); SD_CHECK(sd_canaryIntact(dst + capacity), "config %u : op %u (%s) writes beyond capacity", @@ -3578,14 +3645,16 @@ int fuzzerTests(U32 seed, unsigned nbTests, unsigned startTest, double compressi { unsigned const forceFlush = neverFlush ? 0 : ((FUZ_rand(&randState) & 3) == 1); if (forceFlush) { - /* occasionally provide less than the bound : the flush may then fail */ + /* occasionally provide less than the bound : the flush then fills dstBuffer, + * and output remains, emitted by next invocations of LZ4F_flush() or LZ4F_compressUpdate() */ unsigned const smallDst = (FUZ_rand(&randState) & 31) == 7; size_t const smallCapacity = FUZ_rand(&randState) % (LZ4F_compressBound(0, prefsPtr) + 1); - size_t const flushCapacity = smallDst + size_t flushCapacity = smallDst ? MIN(smallCapacity, (size_t)(oend-op) - 1) : (size_t)(oend-op); BYTE const canaryByte = (BYTE)FUZ_rand(&randState); size_t flushSize; + int flushCompleted; if (smallDst) op[flushCapacity] = canaryByte; flushSize = LZ4F_flush(cCtx, op, flushCapacity, &cOptions); DISPLAYLEVEL(6, "flushing %u bytes \n", (unsigned)flushSize); @@ -3595,12 +3664,23 @@ int fuzzerTests(U32 seed, unsigned nbTests, unsigned startTest, double compressi CHECK(LZ4F_getErrorCode(flushSize) != LZ4F_ERROR_dstCapacity_tooSmall, "Flush into %u bytes : unexpected error %s", (unsigned)flushCapacity, LZ4F_getErrorName(flushSize)); - frameAbandoned = 1; - break; + CHECK(flushCapacity != 0, "Flush into %u bytes must not fail", (unsigned)flushCapacity); + flushSize = 0; /* 0 byte : rejected, the frame continues */ } } CHECK(LZ4F_isError(flushSize), "Compression failed (error %i)", (int)flushSize); op += flushSize; - if ((FUZ_rand(&randState) % 1024) == 3) { + flushCompleted = (flushSize < flushCapacity); + if (!flushCompleted && (FUZ_rand(&randState) & 1)) { + /* complete the flush ; otherwise, next update emits remaining output first */ + do { + flushCapacity = (size_t)(oend-op); + flushSize = LZ4F_flush(cCtx, op, flushCapacity, &cOptions); + CHECK(LZ4F_isError(flushSize), "Flush continuation failed (%s)", LZ4F_getErrorName(flushSize)); + op += flushSize; + } while (flushSize == flushCapacity); + flushCompleted = 1; + } + if (flushCompleted && ((FUZ_rand(&randState) % 1024) == 3)) { /* add an empty block (requires uncompressed flag) */ op[0] = op[1] = op[2] = 0; op[3] = 0x80; /* 0x80000000U in little-endian format */ @@ -3630,28 +3710,24 @@ int fuzzerTests(U32 seed, unsigned nbTests, unsigned startTest, double compressi size_t const dstEndSize = tooSmallDstEnd ? dstEndTooSmallSize : dstEndSafeSize; BYTE const canaryByte = (BYTE)(FUZ_rand(&randState) & 255); size_t flushedSize; + size_t capacity = dstEndSize; DISPLAYLEVEL(7,"canaryByte at pos %u / %u \n", (unsigned)((size_t)(op - (BYTE*)compressedBuffer) + dstEndSize), (unsigned)compressedBufferSize); + (void)tooSmallDstEnd; assert(op + dstEndSize < (BYTE*)compressedBuffer + compressedBufferSize); - op[dstEndSize] = canaryByte; - flushedSize = LZ4F_compressEnd(cCtx, op, dstEndSize, &cOptions); - CHECK(op[dstEndSize] != canaryByte, "LZ4F_compressEnd writes beyond dstCapacity !"); - if (LZ4F_isError(flushedSize)) { - CHECK(!tooSmallDstEnd, "Compression completion failed (error %i : %s)", - (int)flushedSize, LZ4F_getErrorName(flushedSize)); - /* failure is allowed */ - CHECK(LZ4F_getErrorCode(flushedSize) != LZ4F_ERROR_dstCapacity_tooSmall, - "LZ4F_compressEnd into %u bytes : unexpected error %s", - (unsigned)dstEndSize, LZ4F_getErrorName(flushedSize)); - /* the frame can't be completed anymore */ - flushedSize = LZ4F_compressEnd(cCtx, op, dstEndSafeSize, &cOptions); - CHECK(LZ4F_getErrorCode(flushedSize) != LZ4F_ERROR_compressionState_uninitialized, - "LZ4F_compressEnd after a failed LZ4F_compressEnd : %s", - LZ4F_isError(flushedSize) ? LZ4F_getErrorName(flushedSize) : "should have failed"); - continue; + for (;;) { + /* LZ4F_compressEnd() fills dstBuffer, or completes the frame */ + op[capacity] = canaryByte; + flushedSize = LZ4F_compressEnd(cCtx, op, capacity, &cOptions); + CHECK(op[capacity] != canaryByte, "LZ4F_compressEnd writes beyond dstCapacity !"); + CHECK(LZ4F_isError(flushedSize), "Compression completion into %u bytes failed (error %i : %s)", + (unsigned)capacity, (int)flushedSize, LZ4F_getErrorName(flushedSize)); + op += flushedSize; + if (flushedSize < capacity) break; + capacity = (FUZ_rand(&randState) % dstEndSafeSize) + 1; + assert(op + capacity < (BYTE*)compressedBuffer + compressedBufferSize); } - op += flushedSize; } cSize = (size_t)(op - (BYTE*)compressedBuffer); DISPLAYLEVEL(5, "\nCompressed %u bytes into %u \n", (U32)srcSize, (U32)cSize);