lib/lz4
1
0
Fork 0
mirror of https://github.com/lz4/lz4.git synced 2026-10-24 01:48:57 +03:00
Code Activity

lz4frame: LZ4F_flush() and LZ4F_compressEnd() fill dstBuffer or complete ; CLI releaseLag from LZ4F_compressBound(0) (checkpoint)

Browse Source 
LZ4F_flush() and LZ4F_compressEnd() : each invocation fills dstBuffer, or completes :
while it returns dstCapacity, output may remain, and the operation is invoked again, with any dstCapacity.
Output remains strictly below LZ4F_compressBound(0) (+1 for autoFlush), so that callers sizing
dstBuffer with the bound see no change : a single invocation completes.
- the first invocation sets the flush point : buffered input is compressed (all jobs waited for),
  so input is no longer read after it returns (LZ4F_srcStableUntilFlush lifetime unchanged) ;
- single-threaded : a block not fitting is compressed into stageBuff (allocated on first need), same output ;
  multi-threaded : the oldest job can be emitted partially (mt->emitted) ; frame footer staged (8 bytes) ;
- compressUpdate() / uncompressedUpdate() may follow an incomplete flush : remaining output comes first ;
  until an end completes, only LZ4F_compressEnd() continues it (LZ4F_compressBegin() abandons it) ;
- dstCapacity 0 is rejected when output remains, without modifying the frame ;
- contentSize mismatch is detected when the end starts, before writing anything ;
- LZ4F_compressFrame*() keep failing with dstCapacity_tooSmall when the frame doesn't fit.
lz4file : LZ4F_writeClose() loops on LZ4F_compressEnd().
lz4 : releaseLag = LZ4F_compressBound(0) + 64 KB (bound of input retained within cctx).
tests : frametest (#1817 streaming probes, fuzzer), frameMT (validatePartialFlush, small dst, strict bound),
stream fuzzer : flush and end in pieces.
This commit is contained in:
Yann Collet
2026-10-20 18:16:24 -07:00
parent a88e3a7311
commit fa086fd0d6
8 changed files with 737 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
NEWS
View File

@@ -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

22
lib/lz4file.c
View File

@@ -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);

381
lib/lz4frame.c
View File

@@ -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;
}

27
lib/lz4frame.h
View File

@@ -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,

55
ossfuzz/compress_frame_stream_fuzzer.c
View File

@@ -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);

8
programs/lz4io.c
View File

@@ -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);

262
tests/frameMT.c
View File

@@ -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();

120
tests/frametest.c
View File

@@ -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);