make stableSrc compatible with regular streaming API

including flushStream().

Now the only condition is for `input.size` to continuously grow.
This commit is contained in:
Yann Collet
2022-01-26 10:31:24 -08:00
parent c0c5ffa973
commit 37b87add7a
3 changed files with 48 additions and 44 deletions
+9 -8
View File
@@ -5338,7 +5338,7 @@ static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs,
U32 someMoreWork = 1; U32 someMoreWork = 1;
/* check expectations */ /* check expectations */
DEBUGLOG(5, "ZSTD_compressStream_generic, flush=%u", (unsigned)flushMode); DEBUGLOG(5, "ZSTD_compressStream_generic, flush=%i", (int)flushMode);
if (zcs->appliedParams.inBufferMode == ZSTD_bm_buffered) { if (zcs->appliedParams.inBufferMode == ZSTD_bm_buffered) {
assert(zcs->inBuff != NULL); assert(zcs->inBuff != NULL);
assert(zcs->inBuffSize > 0); assert(zcs->inBuffSize > 0);
@@ -5424,9 +5424,8 @@ static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs,
if (!lastBlock) if (!lastBlock)
assert(zcs->inBuffTarget <= zcs->inBuffSize); assert(zcs->inBuffTarget <= zcs->inBuffSize);
zcs->inToCompress = zcs->inBuffPos; zcs->inToCompress = zcs->inBuffPos;
} else { } else { /* !inputBuffered, hence ZSTD_bm_stable */
unsigned const lastBlock = (ip + iSize == iend); unsigned const lastBlock = (ip + iSize == iend);
assert(flushMode == ZSTD_e_end /* Already validated */);
cSize = lastBlock ? cSize = lastBlock ?
ZSTD_compressEnd(zcs, cDst, oSize, ip, iSize) : ZSTD_compressEnd(zcs, cDst, oSize, ip, iSize) :
ZSTD_compressContinue(zcs, cDst, oSize, ip, iSize); ZSTD_compressContinue(zcs, cDst, oSize, ip, iSize);
@@ -5533,11 +5532,10 @@ static size_t ZSTD_checkBufferStability(ZSTD_CCtx const* cctx,
{ {
if (cctx->appliedParams.inBufferMode == ZSTD_bm_stable) { if (cctx->appliedParams.inBufferMode == ZSTD_bm_stable) {
ZSTD_inBuffer const expect = cctx->expectedInBuffer; ZSTD_inBuffer const expect = cctx->expectedInBuffer;
if (expect.src != input->src || expect.pos != input->pos || expect.size != input->size) if (expect.src != input->src || expect.pos != input->pos)
RETURN_ERROR(srcBuffer_wrong, "ZSTD_c_stableInBuffer enabled but input differs!"); RETURN_ERROR(srcBuffer_wrong, "ZSTD_c_stableInBuffer enabled but input differs!");
if (endOp != ZSTD_e_end)
RETURN_ERROR(srcBuffer_wrong, "ZSTD_c_stableInBuffer can only be used with ZSTD_e_end!");
} }
(void)endOp;
if (cctx->appliedParams.outBufferMode == ZSTD_bm_stable) { if (cctx->appliedParams.outBufferMode == ZSTD_bm_stable) {
size_t const outBufferSize = output->size - output->pos; size_t const outBufferSize = output->size - output->pos;
if (cctx->expectedOutBufferSize != outBufferSize) if (cctx->expectedOutBufferSize != outBufferSize)
@@ -5650,7 +5648,7 @@ size_t ZSTD_compressStream2( ZSTD_CCtx* cctx,
&& (endOp == ZSTD_e_continue) /* more to come */ && (endOp == ZSTD_e_continue) /* more to come */
&& (input->pos < ZSTD_BLOCKSIZE_MAX) ) { /* not even reached one block yet */ && (input->pos < ZSTD_BLOCKSIZE_MAX) ) { /* not even reached one block yet */
cctx->expectedInBuffer = *input; cctx->expectedInBuffer = *input;
return (ZSTD_BLOCKSIZE_MAX - input->pos); /* don't do anything : allows lazy compression parameters adaptation */ return (ZSTD_BLOCKSIZE_MAX - input->pos); /* don't do anything : allows lazy adaptation of compression parameters */
} }
FORWARD_IF_ERROR(ZSTD_CCtx_init_compressStream2(cctx, endOp, input->size), "CompressStream2 initialization failed"); FORWARD_IF_ERROR(ZSTD_CCtx_init_compressStream2(cctx, endOp, input->size), "CompressStream2 initialization failed");
ZSTD_setBufferExpectations(cctx, output, input); /* Set initial buffer expectations now that we've initialized */ ZSTD_setBufferExpectations(cctx, output, input); /* Set initial buffer expectations now that we've initialized */
@@ -6167,7 +6165,10 @@ size_t ZSTD_compressSequences(ZSTD_CCtx* const cctx, void* dst, size_t dstCapaci
* @return : amount of data remaining to flush */ * @return : amount of data remaining to flush */
size_t ZSTD_flushStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output) size_t ZSTD_flushStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output)
{ {
ZSTD_inBuffer input = { NULL, 0, 0 }; ZSTD_inBuffer const nullInput = { NULL, 0, 0 };
int const stableInput = (zcs->appliedParams.inBufferMode == ZSTD_bm_stable);
ZSTD_inBuffer input = stableInput ? zcs->expectedInBuffer : nullInput;
input.size = input.pos; /* do not ingest more input during flush */
return ZSTD_compressStream2(zcs, output, &input, ZSTD_e_flush); return ZSTD_compressStream2(zcs, output, &input, ZSTD_e_flush);
} }
+14 -14
View File
@@ -1829,13 +1829,16 @@ ZSTDLIB_STATIC_API size_t ZSTD_CCtx_refPrefix_advanced(ZSTD_CCtx* cctx, const vo
* Experimental parameter. * Experimental parameter.
* Default is 0 == disabled. Set to 1 to enable. * Default is 0 == disabled. Set to 1 to enable.
* *
* Tells the compressor that the ZSTD_inBuffer will ALWAYS be the same * Tells the compressor that input data presented with ZSTD_inBuffer
* between calls, except for the modifications that zstd makes to pos (the * will ALWAYS be the same between calls.
* caller must not modify pos). This is checked by the compressor, and * Technically, the @src pointer must never be changed,
* compression will fail if it ever changes. This means the only flush * and the @pos field can only be updated by zstd.
* mode that makes sense is ZSTD_e_end, so zstd will error if ZSTD_e_end * However, it's possible to increase the @size field,
* is not used. The data in the ZSTD_inBuffer in the range [src, src + pos) * allowing scenarios where more data can be appended after compressions starts.
* MUST not be modified during compression or you will get data corruption. * These conditions are checked by the compressor,
* and compression will fail if they are not respected.
* Also, data in the ZSTD_inBuffer within the range [src, src + pos)
* MUST not be modified during compression or it will result in data corruption.
* *
* When this flag is enabled zstd won't allocate an input window buffer, * When this flag is enabled zstd won't allocate an input window buffer,
* because the user guarantees it can reference the ZSTD_inBuffer until * because the user guarantees it can reference the ZSTD_inBuffer until
@@ -1843,18 +1846,15 @@ ZSTDLIB_STATIC_API size_t ZSTD_CCtx_refPrefix_advanced(ZSTD_CCtx* cctx, const vo
* large enough to fit a block (see ZSTD_c_stableOutBuffer). This will also * large enough to fit a block (see ZSTD_c_stableOutBuffer). This will also
* avoid the memcpy() from the input buffer to the input window buffer. * avoid the memcpy() from the input buffer to the input window buffer.
* *
* NOTE: ZSTD_compressStream2() will error if ZSTD_e_flush is used.
* That means this flag cannot be used with ZSTD_flushStream().
*
* NOTE: So long as the ZSTD_inBuffer always points to valid memory, using * NOTE: So long as the ZSTD_inBuffer always points to valid memory, using
* this flag is ALWAYS memory safe, and will never access out-of-bounds * this flag is ALWAYS memory safe, and will never access out-of-bounds
* memory. However, compression WILL fail if you violate the preconditions. * memory. However, compression WILL fail if conditions are not respected.
* *
* WARNING: The data in the ZSTD_inBuffer in the range [src, src + pos) MUST * WARNING: The data in the ZSTD_inBuffer in the range [src, src + pos) MUST
* not be modified during compression or you will get data corruption. This * not be modified during compression or it sill result in data corruption.
* is because zstd needs to reference data in the ZSTD_inBuffer to find * This is because zstd needs to reference data in the ZSTD_inBuffer to find
* matches. Normally zstd maintains its own window buffer for this purpose, * matches. Normally zstd maintains its own window buffer for this purpose,
* but passing this flag tells zstd to use the user provided buffer. * but passing this flag tells zstd to rely on user provided buffer instead.
*/ */
#define ZSTD_c_stableInBuffer ZSTD_c_experimentalParam9 #define ZSTD_c_stableInBuffer ZSTD_c_experimentalParam9
+25 -22
View File
@@ -613,7 +613,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(3, "OK (error detected : %s) \n", ZSTD_getErrorName(r)); DISPLAYLEVEL(3, "OK (error detected : %s) \n", ZSTD_getErrorName(r));
} } } }
/* Complex context re-use scenario */ /* Compression state re-use scenario */
DISPLAYLEVEL(3, "test%3i : context re-use : ", testNb++); DISPLAYLEVEL(3, "test%3i : context re-use : ", testNb++);
ZSTD_freeCStream(zc); ZSTD_freeCStream(zc);
zc = ZSTD_createCStream(); zc = ZSTD_createCStream();
@@ -634,8 +634,7 @@ static int basicUnitTests(U32 seed, double compressibility)
CHECK_Z( ZSTD_compressStream(zc, &outBuff, &inBuff) ); CHECK_Z( ZSTD_compressStream(zc, &outBuff, &inBuff) );
if (inBuff.pos != inBuff.size) goto _output_error; /* entire input should be consumed */ if (inBuff.pos != inBuff.size) goto _output_error; /* entire input should be consumed */
DISPLAYLEVEL(5, "end1 "); DISPLAYLEVEL(5, "end1 ");
{ size_t const r = ZSTD_endStream(zc, &outBuff); if (ZSTD_endStream(zc, &outBuff) != 0) goto _output_error; /* error, or some data not flushed */
if (r != 0) goto _output_error; } /* error, or some data not flushed */
} }
/* use 2 */ /* use 2 */
{ size_t const inSize = 1025; /* will not continue, because tables auto-adjust and are therefore different size */ { size_t const inSize = 1025; /* will not continue, because tables auto-adjust and are therefore different size */
@@ -653,8 +652,7 @@ static int basicUnitTests(U32 seed, double compressibility)
CHECK_Z( ZSTD_compressStream(zc, &outBuff, &inBuff) ); CHECK_Z( ZSTD_compressStream(zc, &outBuff, &inBuff) );
if (inBuff.pos != inBuff.size) goto _output_error; /* entire input should be consumed */ if (inBuff.pos != inBuff.size) goto _output_error; /* entire input should be consumed */
DISPLAYLEVEL(5, "end2 "); DISPLAYLEVEL(5, "end2 ");
{ size_t const r = ZSTD_endStream(zc, &outBuff); if (ZSTD_endStream(zc, &outBuff) != 0) goto _output_error; /* error, or some data not flushed */
if (r != 0) goto _output_error; } /* error, or some data not flushed */
} }
DISPLAYLEVEL(3, "OK \n"); DISPLAYLEVEL(3, "OK \n");
@@ -786,16 +784,18 @@ static int basicUnitTests(U32 seed, double compressibility)
CHECK(!(cSize < ZSTD_compressBound(CNBufferSize)), "cSize too large for test"); CHECK(!(cSize < ZSTD_compressBound(CNBufferSize)), "cSize too large for test");
CHECK_Z(cSize = ZSTD_compress2(cctx, compressedBuffer, cSize + 4, CNBuffer, CNBufferSize)); CHECK_Z(cSize = ZSTD_compress2(cctx, compressedBuffer, cSize + 4, CNBuffer, CNBufferSize));
CHECK_Z(cctxSize1 = ZSTD_sizeof_CCtx(cctx)); CHECK_Z(cctxSize1 = ZSTD_sizeof_CCtx(cctx));
{ ZSTD_CCtx* cctx2 = ZSTD_createCCtx(); { ZSTD_CCtx* const cctx2 = ZSTD_createCCtx();
assert(cctx2 != NULL);
in.pos = out.pos = 0; in.pos = out.pos = 0;
CHECK_Z(ZSTD_compressStream2(cctx2, &out, &in, ZSTD_e_continue)); CHECK_Z(ZSTD_compressStream2(cctx2, &out, &in, ZSTD_e_continue));
CHECK(!(ZSTD_compressStream2(cctx2, &out, &in, ZSTD_e_end) == 0), "Not finished"); CHECK(!(ZSTD_compressStream2(cctx2, &out, &in, ZSTD_e_end) == 0), "Not finished");
CHECK_Z(cctxSize2 = ZSTD_sizeof_CCtx(cctx2)); CHECK_Z(cctxSize2 = ZSTD_sizeof_CCtx(cctx2));
ZSTD_freeCCtx(cctx2); ZSTD_freeCCtx(cctx2);
} }
{ ZSTD_CCtx* cctx3 = ZSTD_createCCtx(); { ZSTD_CCtx* const cctx3 = ZSTD_createCCtx();
ZSTD_parameters params = ZSTD_getParams(0, CNBufferSize, 0); ZSTD_parameters params = ZSTD_getParams(0, CNBufferSize, 0);
size_t cSize3; size_t cSize3;
assert(cctx3 != NULL);
params.fParams.checksumFlag = 1; params.fParams.checksumFlag = 1;
cSize3 = ZSTD_compress_advanced(cctx3, compressedBuffer, compressedBufferSize, CNBuffer, CNBufferSize, NULL, 0, params); cSize3 = ZSTD_compress_advanced(cctx3, compressedBuffer, compressedBufferSize, CNBuffer, CNBufferSize, NULL, 0, params);
CHECK_Z(cSize3); CHECK_Z(cSize3);
@@ -808,8 +808,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(3, "OK \n"); DISPLAYLEVEL(3, "OK \n");
DISPLAYLEVEL(3, "test%3i : ZSTD_compress2() doesn't modify user parameters : ", testNb++); DISPLAYLEVEL(3, "test%3i : ZSTD_compress2() doesn't modify user parameters : ", testNb++);
{ { int stableInBuffer;
int stableInBuffer;
int stableOutBuffer; int stableOutBuffer;
CHECK_Z(ZSTD_CCtx_getParameter(cctx, ZSTD_c_stableInBuffer, &stableInBuffer)); CHECK_Z(ZSTD_CCtx_getParameter(cctx, ZSTD_c_stableInBuffer, &stableInBuffer));
CHECK_Z(ZSTD_CCtx_getParameter(cctx, ZSTD_c_stableOutBuffer, &stableOutBuffer)); CHECK_Z(ZSTD_CCtx_getParameter(cctx, ZSTD_c_stableOutBuffer, &stableOutBuffer));
@@ -870,21 +869,24 @@ static int basicUnitTests(U32 seed, double compressibility)
} }
DISPLAYLEVEL(3, "OK \n"); DISPLAYLEVEL(3, "OK \n");
DISPLAYLEVEL(3, "test%3i : ZSTD_compressStream2() ZSTD_c_stableInBuffer with continue and flush : ", testNb++); /* stableSrc + streaming */
DISPLAYLEVEL(3, "test%3i : ZSTD_c_stableInBuffer compatibility with compressStream, flushStream and endStream : ", testNb++);
CHECK_Z( ZSTD_initCStream(cctx, 1) );
CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_stableInBuffer, 1) );
in.src = CNBuffer; in.src = CNBuffer;
in.size = CNBufferSize; in.size = 100;
in.pos = 0; in.pos = 0;
out.pos = 0; { ZSTD_outBuffer outBuf;
out.size = compressedBufferSize; outBuf.dst = (char*)(compressedBuffer)+cSize;
CHECK_Z(ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only)); outBuf.size = ZSTD_compressBound(500);
{ size_t const ret = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue); outBuf.pos = 0;
CHECK(!ZSTD_isError(ret), "Must error"); CHECK_Z( ZSTD_compressStream(cctx, &outBuf, &in) );
CHECK(!(ZSTD_getErrorCode(ret) == ZSTD_error_srcBuffer_wrong), "Must be this error"); in.size = 200;
} CHECK_Z( ZSTD_compressStream(cctx, &outBuf, &in) );
CHECK_Z(ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only)); CHECK_Z( ZSTD_flushStream(cctx, &outBuf) );
{ size_t const ret = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush); in.size = 300;
CHECK(!ZSTD_isError(ret), "Must error"); CHECK_Z( ZSTD_compressStream(cctx, &outBuf, &in) );
CHECK(!(ZSTD_getErrorCode(ret) == ZSTD_error_srcBuffer_wrong), "Must be this error"); if (ZSTD_endStream(cctx, &outBuf) != 0) goto _output_error; /* error, or some data not flushed */
} }
DISPLAYLEVEL(3, "OK \n"); DISPLAYLEVEL(3, "OK \n");
@@ -902,6 +904,7 @@ static int basicUnitTests(U32 seed, double compressibility)
CHECK_Z(ZSTD_CCtx_setParameter(cctx, ZSTD_c_stableOutBuffer, 1)); CHECK_Z(ZSTD_CCtx_setParameter(cctx, ZSTD_c_stableOutBuffer, 1));
in.pos = out.pos = 0; in.pos = out.pos = 0;
in.size = MIN(CNBufferSize, 10); in.size = MIN(CNBufferSize, 10);
out.size = compressedBufferSize;
CHECK_Z(ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush)); CHECK_Z(ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_flush));
in.pos = 0; in.pos = 0;
in.size = CNBufferSize - in.size; in.size = CNBufferSize - in.size;