diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 6e8329440..05bcf8a6b 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -200,6 +200,78 @@ typedef char ZSTD_rust_compress_continue_state_layout[ && sizeof(ZSTD_rust_compressContinueState) == (sizeof(void*) == 8 ? 96 : 52)) ? 1 : -1]; + +/* Rust owns the single-threaded buffered/stable stream state machine. The + * projection contains only stream bookkeeping and callback slots; operations + * which still need the private CCtx layout remain C callbacks. */ +typedef size_t (*ZSTD_rust_compressStreamBlock_f)( + void* context, void* dst, size_t dstCapacity, + const void* src, size_t srcSize); +typedef size_t (*ZSTD_rust_compressStreamReset_f)(void* context); +typedef struct { + void* callbackContext; + int inBufferMode; + int outBufferMode; + ZSTD_cStreamStage* streamStage; + size_t blockSizeMax; + size_t* stableInNotConsumed; + void* inBuff; + size_t inBuffSize; + size_t* inToCompress; + size_t* inBuffPos; + size_t* inBuffTarget; + void* outBuff; + size_t outBuffSize; + size_t* outBuffContentSize; + size_t* outBuffFlushedSize; + U32* frameEnded; + ZSTD_rust_compressStreamBlock_f compressContinue; + ZSTD_rust_compressStreamBlock_f compressEnd; + ZSTD_rust_compressStreamReset_f resetSession; +} ZSTD_rust_compressStreamState; +size_t ZSTD_rust_compressStreamGeneric( + const ZSTD_rust_compressStreamState* state, + ZSTD_outBuffer* output, ZSTD_inBuffer* input, int flushMode); +typedef char ZSTD_rust_compress_stream_state_layout[ + (offsetof(ZSTD_rust_compressStreamState, callbackContext) == 0 + && offsetof(ZSTD_rust_compressStreamState, inBufferMode) == sizeof(void*) + && offsetof(ZSTD_rust_compressStreamState, outBufferMode) + == sizeof(void*) + sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, streamStage) + == sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, blockSizeMax) + == 2 * sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, stableInNotConsumed) + == 3 * sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, inBuff) + == 4 * sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, inBuffSize) + == 5 * sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressStreamState, inToCompress) + == 5 * sizeof(void*) + 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, inBuffPos) + == 6 * sizeof(void*) + 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, inBuffTarget) + == 7 * sizeof(void*) + 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, outBuff) + == 8 * sizeof(void*) + 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, outBuffSize) + == 9 * sizeof(void*) + 2 * sizeof(int) + sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, outBuffContentSize) + == 9 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, outBuffFlushedSize) + == 10 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, frameEnded) + == 11 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, compressContinue) + == 12 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, compressEnd) + == 13 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && offsetof(ZSTD_rust_compressStreamState, resetSession) + == 14 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) + && sizeof(ZSTD_rust_compressStreamState) + == 15 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t)) + ? 1 : -1]; /* The target-sized block body only needs this narrow projection of ZSTD_CCtx. * Matchfinder/window state, sequence-store construction, and outer repeat-mode * cleanup remain in C. */ @@ -4498,190 +4570,54 @@ static size_t ZSTD_nextInputSizeHint(const ZSTD_CCtx* cctx) /** ZSTD_compressStream_generic(): * internal function for all *compressStream*() variants * @return : hint size for next input to complete ongoing block */ +static size_t ZSTD_rust_compressStream_continue( + void* context, void* dst, size_t dstCapacity, + const void* src, size_t srcSize) +{ + return ZSTD_compressContinue_public( + (ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize); +} + +static size_t ZSTD_rust_compressStream_end( + void* context, void* dst, size_t dstCapacity, + const void* src, size_t srcSize) +{ + return ZSTD_compressEnd_public( + (ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize); +} + +static size_t ZSTD_rust_compressStream_reset(void* context) +{ + return ZSTD_CCtx_reset((ZSTD_CCtx*)context, ZSTD_reset_session_only); +} + static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs, ZSTD_outBuffer* output, ZSTD_inBuffer* input, ZSTD_EndDirective const flushMode) { - const char* const istart = (assert(input != NULL), (const char*)input->src); - const char* const iend = (istart != NULL) ? istart + input->size : istart; - const char* ip = (istart != NULL) ? istart + input->pos : istart; - char* const ostart = (assert(output != NULL), (char*)output->dst); - char* const oend = (ostart != NULL) ? ostart + output->size : ostart; - char* op = (ostart != NULL) ? ostart + output->pos : ostart; - U32 someMoreWork = 1; - - /* check expectations */ - DEBUGLOG(5, "ZSTD_compressStream_generic, flush=%i, srcSize = %zu", (int)flushMode, input->size - input->pos); - assert(zcs != NULL); - if (zcs->appliedParams.inBufferMode == ZSTD_bm_stable) { - assert(input->pos >= zcs->stableIn_notConsumed); - input->pos -= zcs->stableIn_notConsumed; - if (ip) ip -= zcs->stableIn_notConsumed; - zcs->stableIn_notConsumed = 0; - } - if (zcs->appliedParams.inBufferMode == ZSTD_bm_buffered) { - assert(zcs->inBuff != NULL); - assert(zcs->inBuffSize > 0); - } - if (zcs->appliedParams.outBufferMode == ZSTD_bm_buffered) { - assert(zcs->outBuff != NULL); - assert(zcs->outBuffSize > 0); - } - if (input->src == NULL) assert(input->size == 0); - assert(input->pos <= input->size); - if (output->dst == NULL) assert(output->size == 0); - assert(output->pos <= output->size); - assert((U32)flushMode <= (U32)ZSTD_e_end); - - while (someMoreWork) { - switch(zcs->streamStage) - { - case zcss_init: - RETURN_ERROR(init_missing, "call ZSTD_initCStream() first!"); - - case zcss_load: - if ( (flushMode == ZSTD_e_end) - && ( (size_t)(oend-op) >= ZSTD_compressBound((size_t)(iend-ip)) /* Enough output space */ - || zcs->appliedParams.outBufferMode == ZSTD_bm_stable) /* OR we are allowed to return dstSizeTooSmall */ - && (zcs->inBuffPos == 0) ) { - /* shortcut to compression pass directly into output buffer */ - size_t const cSize = ZSTD_compressEnd_public(zcs, - op, (size_t)(oend-op), - ip, (size_t)(iend-ip)); - DEBUGLOG(4, "ZSTD_compressEnd : cSize=%u", (unsigned)cSize); - FORWARD_IF_ERROR(cSize, "ZSTD_compressEnd failed"); - ip = iend; - op += cSize; - zcs->frameEnded = 1; - ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only); - someMoreWork = 0; break; - } - /* complete loading into inBuffer in buffered mode */ - if (zcs->appliedParams.inBufferMode == ZSTD_bm_buffered) { - size_t const toLoad = zcs->inBuffTarget - zcs->inBuffPos; - size_t const loaded = ZSTD_limitCopy( - zcs->inBuff + zcs->inBuffPos, toLoad, - ip, (size_t)(iend-ip)); - zcs->inBuffPos += loaded; - if (ip) ip += loaded; - if ( (flushMode == ZSTD_e_continue) - && (zcs->inBuffPos < zcs->inBuffTarget) ) { - /* not enough input to fill full block : stop here */ - someMoreWork = 0; break; - } - if ( (flushMode == ZSTD_e_flush) - && (zcs->inBuffPos == zcs->inToCompress) ) { - /* empty */ - someMoreWork = 0; break; - } - } else { - assert(zcs->appliedParams.inBufferMode == ZSTD_bm_stable); - if ( (flushMode == ZSTD_e_continue) - && ( (size_t)(iend - ip) < zcs->blockSizeMax) ) { - /* can't compress a full block : stop here */ - zcs->stableIn_notConsumed = (size_t)(iend - ip); - ip = iend; /* pretend to have consumed input */ - someMoreWork = 0; break; - } - if ( (flushMode == ZSTD_e_flush) - && (ip == iend) ) { - /* empty */ - someMoreWork = 0; break; - } - } - /* compress current block (note : this stage cannot be stopped in the middle) */ - DEBUGLOG(5, "stream compression stage (flushMode==%u)", flushMode); - { int const inputBuffered = (zcs->appliedParams.inBufferMode == ZSTD_bm_buffered); - void* cDst; - size_t cSize; - size_t oSize = (size_t)(oend-op); - size_t const iSize = inputBuffered ? zcs->inBuffPos - zcs->inToCompress - : MIN((size_t)(iend - ip), zcs->blockSizeMax); - if (oSize >= ZSTD_compressBound(iSize) || zcs->appliedParams.outBufferMode == ZSTD_bm_stable) - cDst = op; /* compress into output buffer, to skip flush stage */ - else - cDst = zcs->outBuff, oSize = zcs->outBuffSize; - if (inputBuffered) { - unsigned const lastBlock = (flushMode == ZSTD_e_end) && (ip==iend); - cSize = lastBlock ? - ZSTD_compressEnd_public(zcs, cDst, oSize, - zcs->inBuff + zcs->inToCompress, iSize) : - ZSTD_compressContinue_public(zcs, cDst, oSize, - zcs->inBuff + zcs->inToCompress, iSize); - FORWARD_IF_ERROR(cSize, "%s", lastBlock ? "ZSTD_compressEnd failed" : "ZSTD_compressContinue failed"); - zcs->frameEnded = lastBlock; - /* prepare next block */ - zcs->inBuffTarget = zcs->inBuffPos + zcs->blockSizeMax; - if (zcs->inBuffTarget > zcs->inBuffSize) - zcs->inBuffPos = 0, zcs->inBuffTarget = zcs->blockSizeMax; - DEBUGLOG(5, "inBuffTarget:%u / inBuffSize:%u", - (unsigned)zcs->inBuffTarget, (unsigned)zcs->inBuffSize); - if (!lastBlock) - assert(zcs->inBuffTarget <= zcs->inBuffSize); - zcs->inToCompress = zcs->inBuffPos; - } else { /* !inputBuffered, hence ZSTD_bm_stable */ - unsigned const lastBlock = (flushMode == ZSTD_e_end) && (ip + iSize == iend); - cSize = lastBlock ? - ZSTD_compressEnd_public(zcs, cDst, oSize, ip, iSize) : - ZSTD_compressContinue_public(zcs, cDst, oSize, ip, iSize); - /* Consume the input prior to error checking to mirror buffered mode. */ - if (ip) ip += iSize; - FORWARD_IF_ERROR(cSize, "%s", lastBlock ? "ZSTD_compressEnd failed" : "ZSTD_compressContinue failed"); - zcs->frameEnded = lastBlock; - if (lastBlock) assert(ip == iend); - } - if (cDst == op) { /* no need to flush */ - op += cSize; - if (zcs->frameEnded) { - DEBUGLOG(5, "Frame completed directly in outBuffer"); - someMoreWork = 0; - ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only); - } - break; - } - zcs->outBuffContentSize = cSize; - zcs->outBuffFlushedSize = 0; - zcs->streamStage = zcss_flush; /* pass-through to flush stage */ - } - ZSTD_FALLTHROUGH; - case zcss_flush: - DEBUGLOG(5, "flush stage"); - assert(zcs->appliedParams.outBufferMode == ZSTD_bm_buffered); - { size_t const toFlush = zcs->outBuffContentSize - zcs->outBuffFlushedSize; - size_t const flushed = ZSTD_limitCopy(op, (size_t)(oend-op), - zcs->outBuff + zcs->outBuffFlushedSize, toFlush); - DEBUGLOG(5, "toFlush: %u into %u ==> flushed: %u", - (unsigned)toFlush, (unsigned)(oend-op), (unsigned)flushed); - if (flushed) - op += flushed; - zcs->outBuffFlushedSize += flushed; - if (toFlush!=flushed) { - /* flush not fully completed, presumably because dst is too small */ - assert(op==oend); - someMoreWork = 0; - break; - } - zcs->outBuffContentSize = zcs->outBuffFlushedSize = 0; - if (zcs->frameEnded) { - DEBUGLOG(5, "Frame completed on flush"); - someMoreWork = 0; - ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only); - break; - } - zcs->streamStage = zcss_load; - break; - } - - default: /* impossible */ - assert(0); - } - } - - input->pos = (size_t)(ip - istart); - output->pos = (size_t)(op - ostart); - if (zcs->frameEnded) return 0; - return ZSTD_nextInputSizeHint(zcs); + ZSTD_rust_compressStreamState state; + state.callbackContext = zcs; + state.inBufferMode = (int)zcs->appliedParams.inBufferMode; + state.outBufferMode = (int)zcs->appliedParams.outBufferMode; + state.streamStage = &zcs->streamStage; + state.blockSizeMax = zcs->blockSizeMax; + state.stableInNotConsumed = &zcs->stableIn_notConsumed; + state.inBuff = zcs->inBuff; + state.inBuffSize = zcs->inBuffSize; + state.inToCompress = &zcs->inToCompress; + state.inBuffPos = &zcs->inBuffPos; + state.inBuffTarget = &zcs->inBuffTarget; + state.outBuff = zcs->outBuff; + state.outBuffSize = zcs->outBuffSize; + state.outBuffContentSize = &zcs->outBuffContentSize; + state.outBuffFlushedSize = &zcs->outBuffFlushedSize; + state.frameEnded = &zcs->frameEnded; + state.compressContinue = ZSTD_rust_compressStream_continue; + state.compressEnd = ZSTD_rust_compressStream_end; + state.resetSession = ZSTD_rust_compressStream_reset; + return ZSTD_rust_compressStreamGeneric( + &state, output, input, (int)flushMode); } static size_t ZSTD_nextInputSizeHint_MTorST(const ZSTD_CCtx* cctx) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 6cfd39637..6529e2301 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -184,6 +184,65 @@ ZSTDMT_RustFlushProducedResult ZSTDMT_rust_flushProduced( ZSTDMT_flushCompleteJobFn completeJob, ZSTDMT_flushErrorFn onError); +/* The Rust outer scheduler sees only this scalar snapshot. The MT context, + * reusable input buffer, worker pool, and all synchronization remain private + * to this translation unit. */ +typedef struct { + unsigned frameEnded; + unsigned jobReady; + void* inBuffStart; + size_t inBuffCapacity; + size_t inBuffFilled; + size_t targetSectionSize; + int rsyncable; + U64 rsyncPrimePower; + U64 rsyncHitMask; +} ZSTDMT_RustCompressStreamContextProjection; +typedef struct { + void* bufferStart; + size_t bufferCapacity; + size_t bufferFilled; +} ZSTDMT_RustStreamInputRangeProjection; +typedef struct { + const void* src; + size_t size; + size_t pos; +} ZSTDMT_RustStreamInputProjection; +typedef struct { + void* dst; + size_t size; + size_t pos; +} ZSTDMT_RustStreamOutputProjection; +typedef struct { + size_t toLoad; + int flush; +} ZSTDMT_RustSyncPointProjection; +typedef struct { + size_t result; + size_t outputPos; +} ZSTDMT_RustStreamFlushResult; +typedef struct { + size_t result; + size_t inputPos; + size_t outputPos; +} ZSTDMT_RustCompressStreamResult; +typedef int (*ZSTDMT_streamTryGetInputRangeFn)( + void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection); +typedef int (*ZSTDMT_streamLoadInputFn)(void* opaque, const void* src, size_t size); +typedef size_t (*ZSTDMT_streamCreateJobFn)(void* opaque, size_t srcSize, unsigned end); +typedef ZSTDMT_RustStreamFlushResult (*ZSTDMT_streamFlushProducedFn)( + void* opaque, void* outputDst, size_t outputSize, size_t outputPos, + unsigned blockToFlush, unsigned end); +ZSTDMT_RustCompressStreamResult ZSTDMT_rust_compressStreamGeneric( + const ZSTDMT_RustCompressStreamContextProjection* context, + const ZSTDMT_RustStreamInputProjection* input, + const ZSTDMT_RustStreamOutputProjection* output, + unsigned end, void* opaque, + ZSTDMT_streamTryGetInputRangeFn tryGetInputRange, + ZSTDMT_streamLoadInputFn loadInput, + ZSTDMT_streamCreateJobFn createJob, + ZSTDMT_streamFlushProducedFn flushProduced); + typedef struct { unsigned lastJob; size_t srcSize; @@ -1648,6 +1707,12 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* mtctx, size_t srcSize, ZS return result.returnCode; } +static size_t ZSTDMT_streamCreateJob(void* opaque, size_t srcSize, unsigned end) +{ + return ZSTDMT_createCompressionJob((ZSTDMT_CCtx*)opaque, srcSize, + (ZSTD_EndDirective)end); +} + /* The Rust flush state machine receives only this synchronized scalar view. * The descriptor, condition variable, serial checksum state, and buffer pool @@ -1763,6 +1828,16 @@ static size_t ZSTDMT_flushProduced(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, u return result.result; } +static ZSTDMT_RustStreamFlushResult ZSTDMT_streamFlushProduced( + void* opaque, void* outputDst, size_t outputSize, size_t outputPos, + unsigned blockToFlush, unsigned end) +{ + ZSTD_outBuffer output = { outputDst, outputSize, outputPos }; + size_t const result = ZSTDMT_flushProduced( + (ZSTDMT_CCtx*)opaque, &output, blockToFlush, (ZSTD_EndDirective)end); + return (ZSTDMT_RustStreamFlushResult){ result, output.pos }; +} + /** * Returns the range of data used by the earliest job that is not yet complete. * If the data of the first job is broken up into two segments, we cover both @@ -1875,29 +1950,35 @@ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx) return 1; } -typedef struct { - size_t toLoad; /* The number of bytes to load from the input. */ - int flush; /* Boolean declaring if we must flush because we found a synchronization point. */ -} SyncPoint; - /** * Searches through the input for a synchronization point. If one is found, we * will instruct the caller to flush, and return the number of bytes to load. * Otherwise, we will load as many bytes as possible and instruct the caller * to continue as normal. */ -static SyncPoint -findSynchronizationPoint(ZSTDMT_CCtx const* mtctx, ZSTD_inBuffer const input) +/* Adapter callback for the C-owned reusable input range. */ +static int ZSTDMT_streamTryGetInputRange( + void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection) { - SyncPoint syncPoint; - ZSTDMT_rust_findSynchronizationPoint( - input.src, input.size, input.pos, - mtctx->targetSectionSize, - mtctx->inBuff.buffer.start, mtctx->inBuff.filled, - mtctx->params.rsyncable, - mtctx->rsync.primePower, mtctx->rsync.hitMask, - &syncPoint.toLoad, &syncPoint.flush); - return syncPoint; + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + int const ready = ZSTDMT_tryGetInputRange(mtctx); + *projection = (ZSTDMT_RustStreamInputRangeProjection){ + mtctx->inBuff.buffer.start, + mtctx->inBuff.buffer.capacity, + mtctx->inBuff.filled + }; + return ready; +} + +static int ZSTDMT_streamLoadInput(void* opaque, const void* src, size_t size) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + assert(mtctx->inBuff.buffer.start != NULL); + assert(mtctx->inBuff.filled <= mtctx->inBuff.buffer.capacity); + assert(size <= mtctx->inBuff.buffer.capacity - mtctx->inBuff.filled); + ZSTD_memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, src, size); + mtctx->inBuff.filled += size; + return 1; } size_t ZSTDMT_nextInputSizeHint(const ZSTDMT_CCtx* mtctx) @@ -1915,69 +1996,39 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, ZSTD_inBuffer* input, ZSTD_EndDirective endOp) { - unsigned forwardInputProgress = 0; + ZSTDMT_RustCompressStreamContextProjection const context = { + mtctx->frameEnded, + (unsigned)mtctx->jobReady, + mtctx->inBuff.buffer.start, + mtctx->inBuff.buffer.capacity, + mtctx->inBuff.filled, + mtctx->targetSectionSize, + mtctx->params.rsyncable, + mtctx->rsync.primePower, + mtctx->rsync.hitMask + }; + ZSTDMT_RustStreamInputProjection const inputProjection = { + input->src, + input->size, + input->pos + }; + ZSTDMT_RustStreamOutputProjection const outputProjection = { + output->dst, + output->size, + output->pos + }; + ZSTDMT_RustCompressStreamResult const result = + ZSTDMT_rust_compressStreamGeneric( + &context, &inputProjection, &outputProjection, (unsigned)endOp, + mtctx, ZSTDMT_streamTryGetInputRange, ZSTDMT_streamLoadInput, + ZSTDMT_streamCreateJob, ZSTDMT_streamFlushProduced); + DEBUGLOG(5, "ZSTDMT_compressStream_generic (endOp=%u, srcSize=%u)", (U32)endOp, (U32)(input->size - input->pos)); assert(output->pos <= output->size); assert(input->pos <= input->size); - - if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) { - /* current frame being ended. Only flush/end are allowed */ - return ERROR(stage_wrong); - } - - /* fill input buffer */ - if ( (!mtctx->jobReady) - && (input->size > input->pos) ) { /* support NULL input */ - if (mtctx->inBuff.buffer.start == NULL) { - assert(mtctx->inBuff.filled == 0); /* Can't fill an empty buffer */ - if (!ZSTDMT_tryGetInputRange(mtctx)) { - /* It is only possible for this operation to fail if there are - * still compression jobs ongoing. - */ - DEBUGLOG(5, "ZSTDMT_tryGetInputRange failed"); - assert(mtctx->doneJobID != mtctx->nextJobID); - } else - DEBUGLOG(5, "ZSTDMT_tryGetInputRange completed successfully : mtctx->inBuff.buffer.start = %p", mtctx->inBuff.buffer.start); - } - if (mtctx->inBuff.buffer.start != NULL) { - SyncPoint const syncPoint = findSynchronizationPoint(mtctx, *input); - if (syncPoint.flush && endOp == ZSTD_e_continue) { - endOp = ZSTD_e_flush; - } - assert(mtctx->inBuff.buffer.capacity >= mtctx->targetSectionSize); - DEBUGLOG(5, "ZSTDMT_compressStream_generic: adding %u bytes on top of %u to buffer of size %u", - (U32)syncPoint.toLoad, (U32)mtctx->inBuff.filled, (U32)mtctx->targetSectionSize); - ZSTD_memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, syncPoint.toLoad); - input->pos += syncPoint.toLoad; - mtctx->inBuff.filled += syncPoint.toLoad; - forwardInputProgress = syncPoint.toLoad>0; - } - } - if ((input->pos < input->size) && (endOp == ZSTD_e_end)) { - /* Can't end yet because the input is not fully consumed. - * We are in one of these cases: - * - mtctx->inBuff is NULL & empty: we couldn't get an input buffer so don't create a new job. - * - We filled the input buffer: flush this job but don't end the frame. - * - We hit a synchronization point: flush this job but don't end the frame. - */ - assert(mtctx->inBuff.filled == 0 || mtctx->inBuff.filled == mtctx->targetSectionSize || mtctx->params.rsyncable); - endOp = ZSTD_e_flush; - } - - if ( (mtctx->jobReady) - || (mtctx->inBuff.filled >= mtctx->targetSectionSize) /* filled enough : let's compress */ - || ((endOp != ZSTD_e_continue) && (mtctx->inBuff.filled > 0)) /* something to flush : let's go */ - || ((endOp == ZSTD_e_end) && (!mtctx->frameEnded)) ) { /* must finish the frame with a zero-size block */ - size_t const jobSize = mtctx->inBuff.filled; - assert(mtctx->inBuff.filled <= mtctx->targetSectionSize); - FORWARD_IF_ERROR( ZSTDMT_createCompressionJob(mtctx, jobSize, endOp) , ""); - } - - /* check for potential compressed data ready to be flushed */ - { size_t const remainingToFlush = ZSTDMT_flushProduced(mtctx, output, !forwardInputProgress, endOp); /* block if there was no forward input progress */ - if (input->pos < input->size) return MAX(remainingToFlush, 1); /* input not consumed : do not end flush yet */ - DEBUGLOG(5, "end of ZSTDMT_compressStream_generic: remainingToFlush = %u", (U32)remainingToFlush); - return remainingToFlush; - } + input->pos = result.inputPos; + output->pos = result.outputPos; + DEBUGLOG(5, "end of ZSTDMT_compressStream_generic: remainingToFlush = %u", (U32)result.result); + return result.result; } diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 052f2d88f..d44a5dff8 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -83,7 +83,6 @@ unsafe extern "C" { const ZSTD_FAST: c_int = 1; const ZSTD_DFAST: c_int = 2; const ZSTD_REP_NUM: usize = 3; -#[cfg(test)] const ZSTD_BM_BUFFERED: c_int = 0; const ZSTD_BM_STABLE: c_int = 1; const ZSTD_SF_NO_BLOCK_DELIMITERS: c_int = 0; @@ -106,8 +105,11 @@ const ZSTD_CURRENT_MAX: usize = if size_of::() == 8 { 2000usize << 20 }; const ZSTD_CHUNKSIZE_MAX: usize = u32::MAX as usize - ZSTD_CURRENT_MAX; -#[cfg(not(test))] const ZSTD_E_END: c_int = 2; +const ZSTD_E_CONTINUE: c_int = 0; +const ZSTD_E_FLUSH: c_int = 1; +const ZSTD_CSTREAM_STAGE_LOAD: c_int = 1; +const ZSTD_CSTREAM_STAGE_FLUSH: c_int = 2; type FrameChunkPrepareFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize); type FrameChunkCompressFn = @@ -519,6 +521,436 @@ pub unsafe extern "C" fn ZSTD_rust_compressContinue( } } +type CompressStreamBlockFn = + unsafe extern "C" fn(*mut c_void, *mut c_void, usize, *const c_void, usize) -> usize; +type CompressStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize; + +/// Explicit projection of the single-threaded `ZSTD_compressStream_generic` +/// state machine. +/// +/// Rust owns buffering, output-drain policy, directive handling, and stream +/// stage transitions. The opaque callback context stays in C; callbacks +/// retain access to `ZSTD_CCtx`, frame construction, and session reset. +#[repr(C)] +pub struct ZSTD_rust_compressStreamState { + callback_context: *mut c_void, + in_buffer_mode: c_int, + out_buffer_mode: c_int, + stream_stage: *mut c_int, + block_size_max: usize, + stable_in_not_consumed: *mut usize, + in_buff: *mut c_void, + in_buff_size: usize, + in_to_compress: *mut usize, + in_buff_pos: *mut usize, + in_buff_target: *mut usize, + out_buff: *mut c_void, + out_buff_size: usize, + out_buff_content_size: *mut usize, + out_buff_flushed_size: *mut usize, + frame_ended: *mut c_uint, + compress_continue: CompressStreamBlockFn, + compress_end: CompressStreamBlockFn, + reset_session: CompressStreamResetFn, +} + +const _: () = { + assert!(offset_of!(ZSTD_rust_compressStreamState, callback_context) == 0); + assert!(offset_of!(ZSTD_rust_compressStreamState, in_buffer_mode) == size_of::()); + assert!( + offset_of!(ZSTD_rust_compressStreamState, out_buffer_mode) + == size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, stream_stage) + == size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, block_size_max) + == 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, stable_in_not_consumed) + == 3 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, in_buff) + == 4 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, in_buff_size) + == 5 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, in_to_compress) + == 5 * size_of::() + 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, in_buff_pos) + == 6 * size_of::() + 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, in_buff_target) + == 7 * size_of::() + 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, out_buff) + == 9 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, out_buff_size) + == 9 * size_of::() + 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, out_buff_content_size) + == 9 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, out_buff_flushed_size) + == 10 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, frame_ended) + == 11 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, compress_continue) + == 12 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, compress_end) + == 13 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStreamState, reset_session) + == 14 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); + assert!( + size_of::() + == 15 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); +}; + +#[inline] +unsafe fn stream_limit_copy( + dst: *mut c_void, + dst_capacity: usize, + src: *const c_void, + src_size: usize, +) -> usize { + let length = dst_capacity.min(src_size); + if length != 0 { + unsafe { + ptr::copy_nonoverlapping(src.cast::(), dst.cast::(), length); + } + } + length +} + +#[inline] +unsafe fn compress_stream_generic_body_with( + state: &ZSTD_rust_compressStreamState, + output: &mut ZSTD_outBuffer, + input: &mut ZSTD_inBuffer, + flush_mode: c_int, +) -> usize { + debug_assert!((ZSTD_E_CONTINUE..=ZSTD_E_END).contains(&flush_mode)); + debug_assert!(input.pos <= input.size); + debug_assert!(output.pos <= output.size); + debug_assert!(!input.src.is_null() || input.size == 0); + debug_assert!(!output.dst.is_null() || output.size == 0); + + if state.stream_stage.is_null() + || state.stable_in_not_consumed.is_null() + || state.in_to_compress.is_null() + || state.in_buff_pos.is_null() + || state.in_buff_target.is_null() + || state.out_buff_content_size.is_null() + || state.out_buff_flushed_size.is_null() + || state.frame_ended.is_null() + { + return ERROR(ZstdErrorCode::Generic); + } + + if state.in_buffer_mode == ZSTD_BM_STABLE { + let stable = unsafe { *state.stable_in_not_consumed }; + debug_assert!(input.pos >= stable); + input.pos = input.pos.wrapping_sub(stable); + *unsafe { state.stable_in_not_consumed.as_mut().unwrap_unchecked() } = 0; + } + + debug_assert!( + state.in_buffer_mode == ZSTD_BM_BUFFERED || state.in_buffer_mode == ZSTD_BM_STABLE + ); + if state.in_buffer_mode == ZSTD_BM_BUFFERED { + debug_assert!(!state.in_buff.is_null()); + debug_assert!(state.in_buff_size != 0); + } + if state.out_buffer_mode == ZSTD_BM_BUFFERED { + debug_assert!(!state.out_buff.is_null()); + debug_assert!(state.out_buff_size != 0); + } + + let mut some_more_work = true; + while some_more_work { + let stage = unsafe { *state.stream_stage }; + match stage { + ZSTD_CSTREAM_STAGE_INIT => return ERROR(ZstdErrorCode::InitMissing), + ZSTD_CSTREAM_STAGE_LOAD => { + let input_remaining = input.size - input.pos; + let output_remaining = output.size - output.pos; + let bound = crate::zstd_compress_api::ZSTD_compressBound(input_remaining); + if flush_mode == ZSTD_E_END + && (output_remaining >= bound || state.out_buffer_mode == ZSTD_BM_STABLE) + && unsafe { *state.in_buff_pos } == 0 + { + let dst = if output.dst.is_null() { + ptr::null_mut() + } else { + unsafe { output.dst.cast::().add(output.pos).cast() } + }; + let src = if input.src.is_null() { + ptr::null() + } else { + unsafe { input.src.cast::().add(input.pos).cast() } + }; + let c_size = unsafe { + (state.compress_end)( + state.callback_context, + dst, + output_remaining, + src, + input_remaining, + ) + }; + if ERR_isError(c_size) { + return c_size; + } + input.pos = input.size; + output.pos += c_size; + unsafe { *state.frame_ended = 1 }; + unsafe { (state.reset_session)(state.callback_context) }; + some_more_work = false; + continue; + } + + if state.in_buffer_mode == ZSTD_BM_BUFFERED { + let to_load = + unsafe { (*state.in_buff_target).wrapping_sub(*state.in_buff_pos) }; + let src = if input.src.is_null() { + ptr::null() + } else { + unsafe { input.src.cast::().add(input.pos).cast() } + }; + let dst = if state.in_buff.is_null() { + ptr::null_mut() + } else { + unsafe { state.in_buff.cast::().add(*state.in_buff_pos).cast() } + }; + let loaded = unsafe { stream_limit_copy(dst, to_load, src, input_remaining) }; + input.pos += loaded; + unsafe { *state.in_buff_pos += loaded }; + + if flush_mode == ZSTD_E_CONTINUE + && unsafe { *state.in_buff_pos < *state.in_buff_target } + { + some_more_work = false; + continue; + } + if flush_mode == ZSTD_E_FLUSH + && unsafe { *state.in_buff_pos == *state.in_to_compress } + { + some_more_work = false; + continue; + } + } else { + debug_assert_eq!(state.in_buffer_mode, ZSTD_BM_STABLE); + let input_remaining = input.size - input.pos; + if flush_mode == ZSTD_E_CONTINUE && input_remaining < state.block_size_max { + unsafe { *state.stable_in_not_consumed = input_remaining }; + input.pos = input.size; + some_more_work = false; + continue; + } + if flush_mode == ZSTD_E_FLUSH && input.pos == input.size { + some_more_work = false; + continue; + } + } + + let input_buffered = state.in_buffer_mode == ZSTD_BM_BUFFERED; + let mut output_size = output.size - output.pos; + let input_size = if input_buffered { + unsafe { (*state.in_buff_pos).wrapping_sub(*state.in_to_compress) } + } else { + (input.size - input.pos).min(state.block_size_max) + }; + let output_dst = if output_size + >= crate::zstd_compress_api::ZSTD_compressBound(input_size) + || state.out_buffer_mode == ZSTD_BM_STABLE + { + if output.dst.is_null() { + ptr::null_mut() + } else { + unsafe { output.dst.cast::().add(output.pos).cast() } + } + } else { + output_size = state.out_buff_size; + state.out_buff + }; + + let last_block = flush_mode == ZSTD_E_END + && if input_buffered { + input.pos == input.size + } else { + input.pos + input_size == input.size + }; + let source = if input_buffered { + if state.in_buff.is_null() { + ptr::null() + } else { + unsafe { state.in_buff.cast::().add(*state.in_to_compress).cast() } + } + } else if input.src.is_null() { + ptr::null() + } else { + unsafe { input.src.cast::().add(input.pos).cast() } + }; + let c_size = unsafe { + if last_block { + (state.compress_end)( + state.callback_context, + output_dst, + output_size, + source, + input_size, + ) + } else { + (state.compress_continue)( + state.callback_context, + output_dst, + output_size, + source, + input_size, + ) + } + }; + if !input_buffered { + /* Match C: stable input is consumed before forwarding a + * compression error. */ + input.pos += input_size; + } + if ERR_isError(c_size) { + return c_size; + } + + unsafe { *state.frame_ended = c_uint::from(last_block) }; + if input_buffered { + unsafe { + *state.in_buff_target = + (*state.in_buff_pos).wrapping_add(state.block_size_max); + if *state.in_buff_target > state.in_buff_size { + *state.in_buff_pos = 0; + *state.in_buff_target = state.block_size_max; + } + *state.in_to_compress = *state.in_buff_pos; + } + } + + let current_output = if output.dst.is_null() { + ptr::null_mut() + } else { + unsafe { output.dst.cast::().add(output.pos).cast() } + }; + if output_dst == current_output { + output.pos += c_size; + if unsafe { *state.frame_ended } != 0 { + unsafe { (state.reset_session)(state.callback_context) }; + some_more_work = false; + } + continue; + } + + unsafe { + *state.out_buff_content_size = c_size; + *state.out_buff_flushed_size = 0; + *state.stream_stage = ZSTD_CSTREAM_STAGE_FLUSH; + } + } + ZSTD_CSTREAM_STAGE_FLUSH => { + debug_assert_eq!(state.out_buffer_mode, ZSTD_BM_BUFFERED); + let to_flush = unsafe { + (*state.out_buff_content_size).wrapping_sub(*state.out_buff_flushed_size) + }; + let src = if state.out_buff.is_null() { + ptr::null() + } else { + unsafe { + state + .out_buff + .cast::() + .add(*state.out_buff_flushed_size) + .cast() + } + }; + let dst = if output.dst.is_null() { + ptr::null_mut() + } else { + unsafe { output.dst.cast::().add(output.pos).cast() } + }; + let flushed = + unsafe { stream_limit_copy(dst, output.size - output.pos, src, to_flush) }; + output.pos += flushed; + unsafe { *state.out_buff_flushed_size += flushed }; + if to_flush != flushed { + some_more_work = false; + continue; + } + + unsafe { + *state.out_buff_content_size = 0; + *state.out_buff_flushed_size = 0; + } + if unsafe { *state.frame_ended } != 0 { + unsafe { (state.reset_session)(state.callback_context) }; + some_more_work = false; + continue; + } + unsafe { *state.stream_stage = ZSTD_CSTREAM_STAGE_LOAD }; + } + _ => { + debug_assert!(false, "invalid C stream stage"); + return ERROR(ZstdErrorCode::StageWrong); + } + } + } + + if unsafe { *state.frame_ended } != 0 { + return 0; + } + next_input_size_hint( + state.in_buffer_mode, + state.block_size_max, + unsafe { *state.stable_in_not_consumed }, + unsafe { *state.in_buff_target }, + unsafe { *state.in_buff_pos }, + ) +} + +/// Drive the single-threaded stream state machine through a C-owned context. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_compressStreamGeneric( + state: *const ZSTD_rust_compressStreamState, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + flush_mode: c_int, +) -> usize { + if state.is_null() || output.is_null() || input.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + unsafe { compress_stream_generic_body_with(&*state, &mut *output, &mut *input, flush_mode) } +} + /// Explicit projection of the state used by `ZSTD_compressSequences_internal`. /// /// The C context and its function-pointer-bearing parameter structure remain @@ -1653,7 +2085,6 @@ pub struct ZSTD_inBuffer { pos: usize, } -#[cfg(not(test))] #[repr(C)] pub struct ZSTD_outBuffer { dst: *mut c_void, @@ -3901,6 +4332,404 @@ mod tests { assert_eq!(context.block_calls, 1); } + #[derive(Default)] + struct CompressStreamTestContext { + continue_calls: usize, + end_calls: usize, + reset_calls: usize, + continue_result: usize, + end_result: usize, + reset_result: usize, + } + + unsafe fn compress_stream_test_context( + context: *mut c_void, + ) -> &'static mut CompressStreamTestContext { + unsafe { &mut *context.cast::() } + } + + unsafe extern "C" fn compress_stream_test_block( + context: *mut c_void, + dst: *mut c_void, + dst_capacity: usize, + _src: *const c_void, + _src_size: usize, + ) -> usize { + let context = unsafe { compress_stream_test_context(context) }; + context.continue_calls += 1; + if !ERR_isError(context.continue_result) && context.continue_result != 0 { + assert!(context.continue_result <= dst_capacity); + unsafe { ptr::write_bytes(dst.cast::(), 0x5a, context.continue_result) }; + } + context.continue_result + } + + unsafe extern "C" fn compress_stream_test_end( + context: *mut c_void, + dst: *mut c_void, + dst_capacity: usize, + _src: *const c_void, + _src_size: usize, + ) -> usize { + let context = unsafe { compress_stream_test_context(context) }; + context.end_calls += 1; + if !ERR_isError(context.end_result) && context.end_result != 0 { + assert!(context.end_result <= dst_capacity); + unsafe { ptr::write_bytes(dst.cast::(), 0x3c, context.end_result) }; + } + context.end_result + } + + unsafe extern "C" fn compress_stream_test_reset(context: *mut c_void) -> usize { + let context = unsafe { compress_stream_test_context(context) }; + context.reset_calls += 1; + context.reset_result + } + + #[allow(clippy::too_many_arguments)] + fn compress_stream_test_state( + context: &mut CompressStreamTestContext, + in_buffer_mode: c_int, + out_buffer_mode: c_int, + stage: &mut c_int, + block_size_max: usize, + stable_in_not_consumed: &mut usize, + in_buff: *mut c_void, + in_buff_size: usize, + in_to_compress: &mut usize, + in_buff_pos: &mut usize, + in_buff_target: &mut usize, + out_buff: *mut c_void, + out_buff_size: usize, + out_buff_content_size: &mut usize, + out_buff_flushed_size: &mut usize, + frame_ended: &mut c_uint, + ) -> ZSTD_rust_compressStreamState { + ZSTD_rust_compressStreamState { + callback_context: (context as *mut CompressStreamTestContext).cast(), + in_buffer_mode, + out_buffer_mode, + stream_stage: stage, + block_size_max, + stable_in_not_consumed, + in_buff, + in_buff_size, + in_to_compress, + in_buff_pos, + in_buff_target, + out_buff, + out_buff_size, + out_buff_content_size, + out_buff_flushed_size, + frame_ended, + compress_continue: compress_stream_test_block, + compress_end: compress_stream_test_end, + reset_session: compress_stream_test_reset, + } + } + + #[test] + fn compress_stream_stable_short_input_is_deferred_with_a_progress_hint() { + let mut context = CompressStreamTestContext::default(); + let mut stage = ZSTD_CSTREAM_STAGE_LOAD; + let mut stable_in_not_consumed = 0; + let mut in_to_compress = 0; + let mut in_buff_pos = 0; + let mut in_buff_target = 0; + let mut out_buff_content_size = 0; + let mut out_buff_flushed_size = 0; + let mut frame_ended = 0; + let state = compress_stream_test_state( + &mut context, + ZSTD_BM_STABLE, + ZSTD_BM_STABLE, + &mut stage, + 8, + &mut stable_in_not_consumed, + ptr::null_mut(), + 0, + &mut in_to_compress, + &mut in_buff_pos, + &mut in_buff_target, + ptr::null_mut(), + 0, + &mut out_buff_content_size, + &mut out_buff_flushed_size, + &mut frame_ended, + ); + let source = [1u8, 2, 3]; + let mut output = [0xa5u8; 8]; + let mut input = ZSTD_inBuffer { + src: source.as_ptr().cast(), + size: source.len(), + pos: 0, + }; + let mut output_buffer = ZSTD_outBuffer { + dst: output.as_mut_ptr().cast(), + size: output.len(), + pos: 0, + }; + + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output_buffer, &mut input, ZSTD_E_CONTINUE) + }; + + assert_eq!(result, 5); + assert_eq!(input.pos, source.len()); + assert_eq!(stable_in_not_consumed, source.len()); + assert_eq!(output_buffer.pos, 0); + assert_eq!(context.continue_calls, 0); + assert_eq!(context.end_calls, 0); + assert_eq!(frame_ended, 0); + } + + #[test] + fn compress_stream_buffered_output_preserves_pending_flush_and_drains_it() { + let mut context = CompressStreamTestContext { + continue_result: 3, + ..CompressStreamTestContext::default() + }; + let mut stage = ZSTD_CSTREAM_STAGE_LOAD; + let mut stable_in_not_consumed = 0; + let mut in_to_compress = 0; + let mut in_buff_pos = 0; + let mut in_buff_target = 4; + let mut out_buff_content_size = 0; + let mut out_buff_flushed_size = 0; + let mut frame_ended = 0; + let mut input_buffer = [0u8; 8]; + let mut internal_output = [0xa5u8; 8]; + let state = compress_stream_test_state( + &mut context, + ZSTD_BM_BUFFERED, + ZSTD_BM_BUFFERED, + &mut stage, + 4, + &mut stable_in_not_consumed, + input_buffer.as_mut_ptr().cast(), + input_buffer.len(), + &mut in_to_compress, + &mut in_buff_pos, + &mut in_buff_target, + internal_output.as_mut_ptr().cast(), + internal_output.len(), + &mut out_buff_content_size, + &mut out_buff_flushed_size, + &mut frame_ended, + ); + let source = [1u8, 2, 3, 4]; + let mut output = [0xa5u8; 2]; + let mut input = ZSTD_inBuffer { + src: source.as_ptr().cast(), + size: source.len(), + pos: 0, + }; + let mut output_buffer = ZSTD_outBuffer { + dst: output.as_mut_ptr().cast(), + size: output.len(), + pos: 0, + }; + + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output_buffer, &mut input, ZSTD_E_CONTINUE) + }; + + assert_eq!(result, 4); + assert_eq!(input.pos, source.len()); + assert_eq!(output_buffer.pos, output.len()); + assert_eq!(output, [0x5a; 2]); + assert_eq!(out_buff_content_size, 3); + assert_eq!(out_buff_flushed_size, 2); + assert_eq!(stage, ZSTD_CSTREAM_STAGE_FLUSH); + assert_eq!(context.continue_calls, 1); + + let mut next_output = [0xa5u8; 2]; + output_buffer = ZSTD_outBuffer { + dst: next_output.as_mut_ptr().cast(), + size: next_output.len(), + pos: 0, + }; + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output_buffer, &mut input, ZSTD_E_FLUSH) + }; + + assert_eq!(result, 4); + assert_eq!(next_output[0], 0x5a); + assert_eq!(output_buffer.pos, 1); + assert_eq!((out_buff_content_size, out_buff_flushed_size), (0, 0)); + assert_eq!(stage, ZSTD_CSTREAM_STAGE_LOAD); + assert_eq!(context.continue_calls, 1); + } + + #[test] + fn compress_stream_end_shortcut_resets_after_direct_completion() { + let mut context = CompressStreamTestContext { + end_result: 5, + ..CompressStreamTestContext::default() + }; + let mut stage = ZSTD_CSTREAM_STAGE_LOAD; + let mut stable_in_not_consumed = 0; + let mut in_to_compress = 0; + let mut in_buff_pos = 0; + let mut in_buff_target = 4; + let mut out_buff_content_size = 0; + let mut out_buff_flushed_size = 0; + let mut frame_ended = 0; + let mut input_buffer = [0u8; 8]; + let mut internal_output = [0u8; 8]; + let state = compress_stream_test_state( + &mut context, + ZSTD_BM_BUFFERED, + ZSTD_BM_STABLE, + &mut stage, + 4, + &mut stable_in_not_consumed, + input_buffer.as_mut_ptr().cast(), + input_buffer.len(), + &mut in_to_compress, + &mut in_buff_pos, + &mut in_buff_target, + internal_output.as_mut_ptr().cast(), + internal_output.len(), + &mut out_buff_content_size, + &mut out_buff_flushed_size, + &mut frame_ended, + ); + let source = [7u8, 8, 9]; + let mut output = [0xa5u8; 8]; + let mut input = ZSTD_inBuffer { + src: source.as_ptr().cast(), + size: source.len(), + pos: 0, + }; + let mut output_buffer = ZSTD_outBuffer { + dst: output.as_mut_ptr().cast(), + size: output.len(), + pos: 0, + }; + + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output_buffer, &mut input, ZSTD_E_END) + }; + + assert_eq!(result, 0); + assert_eq!(input.pos, source.len()); + assert_eq!(output_buffer.pos, 5); + assert_eq!(&output[..5], &[0x3c; 5]); + assert_eq!(frame_ended, 1); + assert_eq!(context.end_calls, 1); + assert_eq!(context.continue_calls, 0); + assert_eq!(context.reset_calls, 1); + } + + #[test] + fn compress_stream_stable_block_error_consumes_input_before_returning() { + let mut context = CompressStreamTestContext { + continue_result: ERROR(ZstdErrorCode::DstSizeTooSmall), + ..CompressStreamTestContext::default() + }; + let mut stage = ZSTD_CSTREAM_STAGE_LOAD; + let mut stable_in_not_consumed = 0; + let mut in_to_compress = 0; + let mut in_buff_pos = 0; + let mut in_buff_target = 0; + let mut out_buff_content_size = 0; + let mut out_buff_flushed_size = 0; + let mut frame_ended = 0; + let state = compress_stream_test_state( + &mut context, + ZSTD_BM_STABLE, + ZSTD_BM_STABLE, + &mut stage, + 4, + &mut stable_in_not_consumed, + ptr::null_mut(), + 0, + &mut in_to_compress, + &mut in_buff_pos, + &mut in_buff_target, + ptr::null_mut(), + 0, + &mut out_buff_content_size, + &mut out_buff_flushed_size, + &mut frame_ended, + ); + let source = [0x11u8; 4]; + let mut output = [0xa5u8; 8]; + let mut input = ZSTD_inBuffer { + src: source.as_ptr().cast(), + size: source.len(), + pos: 0, + }; + let mut output_buffer = ZSTD_outBuffer { + dst: output.as_mut_ptr().cast(), + size: output.len(), + pos: 0, + }; + + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output_buffer, &mut input, ZSTD_E_CONTINUE) + }; + + assert_eq!(result, ERROR(ZstdErrorCode::DstSizeTooSmall)); + assert_eq!(input.pos, source.len()); + assert_eq!(output_buffer.pos, 0); + assert_eq!(context.continue_calls, 1); + assert_eq!(context.end_calls, 0); + assert_eq!(context.reset_calls, 0); + assert_eq!(frame_ended, 0); + } + + #[test] + fn compress_stream_rejects_missing_initialization_before_callbacks() { + let mut context = CompressStreamTestContext::default(); + let mut stage = ZSTD_CSTREAM_STAGE_INIT; + let mut stable_in_not_consumed = 0; + let mut in_to_compress = 0; + let mut in_buff_pos = 0; + let mut in_buff_target = 0; + let mut out_buff_content_size = 0; + let mut out_buff_flushed_size = 0; + let mut frame_ended = 0; + let state = compress_stream_test_state( + &mut context, + ZSTD_BM_STABLE, + ZSTD_BM_STABLE, + &mut stage, + 4, + &mut stable_in_not_consumed, + ptr::null_mut(), + 0, + &mut in_to_compress, + &mut in_buff_pos, + &mut in_buff_target, + ptr::null_mut(), + 0, + &mut out_buff_content_size, + &mut out_buff_flushed_size, + &mut frame_ended, + ); + let mut input = ZSTD_inBuffer { + src: ptr::null(), + size: 0, + pos: 0, + }; + let mut output = ZSTD_outBuffer { + dst: ptr::null_mut(), + size: 0, + pos: 0, + }; + + let result = unsafe { + ZSTD_rust_compressStreamGeneric(&state, &mut output, &mut input, ZSTD_E_FLUSH) + }; + + assert_eq!(result, ERROR(ZstdErrorCode::InitMissing)); + assert_eq!(context.continue_calls, 0); + assert_eq!(context.end_calls, 0); + assert_eq!(context.reset_calls, 0); + } + #[test] fn simple_strategy_follows_source_size_tiers() { assert_eq!(ZSTD_rust_compressCCtxStrategy(0, 1), ZSTD_FAST); diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index abc5f5510..56f3d090c 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -120,7 +120,97 @@ pub type ZSTDMT_flushCompleteJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint, src_size: usize, c_size: usize); pub type ZSTDMT_flushErrorFn = unsafe extern "C" fn(opaque: *mut c_void); +/// Scalar MT state used by the outer streaming scheduler. The C context, +/// input-range ownership, and synchronization objects remain private to C; +/// Rust only reads the fields needed to choose the next operation. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressStreamContextProjection { + pub frameEnded: c_uint, + pub jobReady: c_uint, + pub inBuffStart: *mut c_void, + pub inBuffCapacity: usize, + pub inBuffFilled: usize, + pub targetSectionSize: usize, + pub rsyncable: c_int, + pub rsyncPrimePower: u64, + pub rsyncHitMask: u64, +} + +/// Scalar view of the reusable C-owned input range returned by the existing +/// MT input-range seam. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_streamInputRangeProjection { + pub bufferStart: *mut c_void, + pub bufferCapacity: usize, + pub bufferFilled: usize, +} + +/// Explicit projections of the public stream buffers. These deliberately do +/// not use the private C context layout or rely on a Rust view of `ZSTD_inBuffer` +/// and `ZSTD_outBuffer`. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_streamInputProjection { + pub src: *const c_void, + pub size: usize, + pub pos: usize, +} + +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_streamOutputProjection { + pub dst: *mut c_void, + pub size: usize, + pub pos: usize, +} + +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_syncPointProjection { + pub toLoad: usize, + pub flush: c_int, +} + +/// Result returned by the C-owned pending-output seam after it has updated a +/// temporary output projection. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_streamFlushResult { + pub result: usize, + pub outputPos: usize, +} + +/// Result of one outer MT stream scheduling pass. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressStreamResult { + pub result: usize, + pub inputPos: usize, + pub outputPos: usize, +} + +pub type ZSTDMT_streamTryGetInputRangeFn = unsafe extern "C" fn( + opaque: *mut c_void, + projection: *mut ZSTDMT_streamInputRangeProjection, +) -> c_int; +pub type ZSTDMT_streamLoadInputFn = + unsafe extern "C" fn(opaque: *mut c_void, src: *const c_void, size: usize) -> c_int; +pub type ZSTDMT_streamCreateJobFn = + unsafe extern "C" fn(opaque: *mut c_void, src_size: usize, end: c_uint) -> usize; +pub type ZSTDMT_streamFlushProducedFn = unsafe extern "C" fn( + opaque: *mut c_void, + output_dst: *mut c_void, + output_size: usize, + output_pos: usize, + block_to_flush: c_uint, + end: c_uint, +) -> ZSTDMT_streamFlushResult; + const ZSTD_E_END: c_uint = 2; +const ZSTD_E_FLUSH: c_uint = 1; +const ZSTD_E_CONTINUE: c_uint = 0; /// Scalar view of the C-owned job state needed to finish a streaming frame /// with an empty last block. The job descriptor, buffer pool, and mutex stay @@ -857,6 +947,226 @@ pub unsafe extern "C" fn ZSTDMT_rust_flushProduced( } } +#[inline] +fn compress_stream_result( + result: usize, + input_pos: usize, + output_pos: usize, +) -> ZSTDMT_compressStreamResult { + ZSTDMT_compressStreamResult { + result, + inputPos: input_pos, + outputPos: output_pos, + } +} + +/// Run one outer MT streaming pass while C retains the context and worker +/// implementation details. +/// +/// The callbacks are intentionally narrow: input-range acquisition and input +/// copying can update C-owned buffers, job creation can post the prepared job, +/// and flushing can wait on C-owned job synchronization. Rust owns only the +/// ordering of those operations, the end-directive adjustments, and the +/// public progress/return policy. +#[allow(clippy::too_many_arguments)] +unsafe fn compress_stream_generic_with( + context: ZSTDMT_compressStreamContextProjection, + input: ZSTDMT_streamInputProjection, + output: ZSTDMT_streamOutputProjection, + end_op: c_uint, + mut try_get_input_range: T, + mut load_input: L, + mut create_job: J, + mut flush_produced: F, + mut find_sync_point: S, +) -> ZSTDMT_compressStreamResult +where + T: FnMut(&mut ZSTDMT_streamInputRangeProjection) -> c_int, + L: FnMut(*const c_void, usize) -> c_int, + J: FnMut(usize, c_uint) -> usize, + F: FnMut(*mut c_void, usize, usize, c_uint, c_uint) -> ZSTDMT_streamFlushResult, + S: FnMut( + &ZSTDMT_compressStreamContextProjection, + &ZSTDMT_streamInputProjection, + &ZSTDMT_streamInputRangeProjection, + ) -> ZSTDMT_syncPointProjection, +{ + let mut input_pos = input.pos; + let output_pos = output.pos; + if input_pos > input.size + || output_pos > output.size + || context.targetSectionSize == 0 + || context.inBuffFilled > context.inBuffCapacity + || (context.inBuffStart.is_null() && context.inBuffFilled != 0) + { + return compress_stream_result(ERROR(ZstdErrorCode::Generic), input_pos, output_pos); + } + + if context.frameEnded != 0 && end_op == ZSTD_E_CONTINUE { + return compress_stream_result(ERROR(ZstdErrorCode::StageWrong), input_pos, output_pos); + } + + let mut end_op = end_op; + let mut input_range = ZSTDMT_streamInputRangeProjection { + bufferStart: context.inBuffStart, + bufferCapacity: context.inBuffCapacity, + bufferFilled: context.inBuffFilled, + }; + let mut forward_input_progress = false; + + /* Fill the current round-buffer section unless a prepared job is waiting + * for a worker. This is the same ordering as the C implementation: a + * blocked range acquisition is allowed to fall through to the flush pass. + */ + if context.jobReady == 0 && input.size > input_pos { + if input_range.bufferStart.is_null() { + debug_assert_eq!(input_range.bufferFilled, 0); + let _ = try_get_input_range(&mut input_range); + } + + if !input_range.bufferStart.is_null() { + let sync_point = find_sync_point(&context, &input, &input_range); + if sync_point.flush != 0 && end_op == ZSTD_E_CONTINUE { + end_op = ZSTD_E_FLUSH; + } + + let available = input.size - input_pos; + let Some(buffer_left) = input_range + .bufferCapacity + .checked_sub(input_range.bufferFilled) + else { + return compress_stream_result( + ERROR(ZstdErrorCode::Generic), + input_pos, + output_pos, + ); + }; + if sync_point.toLoad > available + || sync_point.toLoad > buffer_left + || (sync_point.toLoad != 0 && input.src.is_null()) + { + return compress_stream_result( + ERROR(ZstdErrorCode::Generic), + input_pos, + output_pos, + ); + } + + if sync_point.toLoad != 0 { + let source = input + .src + .cast::() + .wrapping_add(input_pos) + .cast::(); + if load_input(source, sync_point.toLoad) == 0 { + return compress_stream_result( + ERROR(ZstdErrorCode::Generic), + input_pos, + output_pos, + ); + } + input_pos += sync_point.toLoad; + input_range.bufferFilled += sync_point.toLoad; + forward_input_progress = true; + } + } + } + + if input_pos < input.size && end_op == ZSTD_E_END { + /* The caller cannot end a frame while input remains outside the + * current job. The current pass must flush this job first. */ + end_op = ZSTD_E_FLUSH; + } + + if context.jobReady != 0 + || input_range.bufferFilled >= context.targetSectionSize + || (end_op != ZSTD_E_CONTINUE && input_range.bufferFilled > 0) + || (end_op == ZSTD_E_END && context.frameEnded == 0) + { + debug_assert!(input_range.bufferFilled <= context.targetSectionSize); + let create_result = create_job(input_range.bufferFilled, end_op); + if ERR_isError(create_result) { + return compress_stream_result(create_result, input_pos, output_pos); + } + } + + let flush_result = flush_produced( + output.dst, + output.size, + output_pos, + u32::from(!forward_input_progress), + end_op, + ); + let remaining_to_flush = flush_result.result; + let result = if input_pos < input.size { + remaining_to_flush.max(1) + } else { + remaining_to_flush + }; + compress_stream_result(result, input_pos, flush_result.outputPos) +} + +/// C ABI entry point for the MT outer scheduling/flush policy. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_compressStreamGeneric( + context: *const ZSTDMT_compressStreamContextProjection, + input: *const ZSTDMT_streamInputProjection, + output: *const ZSTDMT_streamOutputProjection, + end_op: c_uint, + opaque: *mut c_void, + tryGetInputRange: Option, + loadInput: Option, + createJob: Option, + flushProduced: Option, +) -> ZSTDMT_compressStreamResult { + let Some(context) = (unsafe { context.as_ref() }).copied() else { + return compress_stream_result(ERROR(ZstdErrorCode::Generic), 0, 0); + }; + let Some(input) = (unsafe { input.as_ref() }).copied() else { + return compress_stream_result(ERROR(ZstdErrorCode::Generic), 0, 0); + }; + let Some(output) = (unsafe { output.as_ref() }).copied() else { + return compress_stream_result(ERROR(ZstdErrorCode::Generic), input.pos, 0); + }; + let (Some(try_get_input_range), Some(load_input), Some(create_job), Some(flush_produced)) = + (tryGetInputRange, loadInput, createJob, flushProduced) + else { + return compress_stream_result(ERROR(ZstdErrorCode::Generic), input.pos, output.pos); + }; + + unsafe { + compress_stream_generic_with( + context, + input, + output, + end_op, + |projection| try_get_input_range(opaque, projection), + |src, size| load_input(opaque, src, size), + |src_size, end| create_job(opaque, src_size, end), + |dst, size, pos, block_to_flush, end| { + flush_produced(opaque, dst, size, pos, block_to_flush, end) + }, + |context, input, input_range| { + let (to_load, flush) = find_synchronization_point( + input.src, + input.size, + input.pos, + context.targetSectionSize, + input_range.bufferStart.cast_const(), + input_range.bufferFilled, + context.rsyncable, + context.rsyncPrimePower, + context.rsyncHitMask, + ); + ZSTDMT_syncPointProjection { + toLoad: to_load, + flush, + } + }, + ) + } +} + #[inline] fn get_input_data_in_use_with( first_job_id: c_uint, @@ -2849,6 +3159,242 @@ mod tests { assert_eq!(ended.allJobsCompleted, 1); } + #[test] + fn outer_scheduler_loads_input_posts_job_and_does_not_block_after_progress() { + let input_bytes = [1u8, 2, 3, 4, 5]; + let mut round_buffer = [0xa5u8; 4]; + let mut output_buffer = [0xb6u8; 8]; + let context = ZSTDMT_compressStreamContextProjection { + frameEnded: 0, + jobReady: 0, + inBuffStart: ptr::null_mut(), + inBuffCapacity: 0, + inBuffFilled: 0, + targetSectionSize: round_buffer.len(), + rsyncable: 0, + ..ZSTDMT_compressStreamContextProjection::default() + }; + let input = ZSTDMT_streamInputProjection { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let output = ZSTDMT_streamOutputProjection { + dst: output_buffer.as_mut_ptr().cast(), + size: output_buffer.len(), + pos: 1, + }; + let mut range_calls = 0; + let mut load_calls = 0; + let mut create_calls = Vec::new(); + let mut flush_calls = Vec::new(); + let round_buffer_ptr = round_buffer.as_mut_ptr(); + let round_capacity = round_buffer.len(); + + let result = unsafe { + compress_stream_generic_with( + context, + input, + output, + ZSTD_E_CONTINUE, + |projection| { + range_calls += 1; + projection.bufferStart = round_buffer_ptr.cast(); + projection.bufferCapacity = round_capacity; + projection.bufferFilled = 0; + 1 + }, + |src, size| { + load_calls += 1; + assert_eq!(size, round_capacity); + ptr::copy_nonoverlapping(src.cast::(), round_buffer_ptr, size); + 1 + }, + |src_size, end| { + create_calls.push((src_size, end)); + 0 + }, + |_dst, _size, output_pos, block_to_flush, end| { + flush_calls.push((block_to_flush, end)); + ZSTDMT_streamFlushResult { + result: 0, + outputPos: output_pos + 2, + } + }, + |_context, input, projection| ZSTDMT_syncPointProjection { + toLoad: (input.size - input.pos).min(round_capacity - projection.bufferFilled), + flush: 0, + }, + ) + }; + + assert_eq!(range_calls, 1); + assert_eq!(load_calls, 1); + assert_eq!(round_buffer, input_bytes[..round_buffer.len()]); + assert_eq!(create_calls, vec![(round_buffer.len(), ZSTD_E_CONTINUE)]); + assert_eq!(flush_calls, vec![(0, ZSTD_E_CONTINUE)]); + assert_eq!(result.result, 1); + assert_eq!(result.inputPos, round_buffer.len()); + assert_eq!(result.outputPos, 3); + assert_eq!(output_buffer[0], 0xb6); + } + + #[test] + fn outer_scheduler_forces_flush_before_end_when_input_remains() { + let input_bytes = [7u8, 8, 9, 10, 11]; + let mut round_buffer = [0xa5u8; 3]; + let context = ZSTDMT_compressStreamContextProjection { + inBuffStart: round_buffer.as_mut_ptr().cast(), + inBuffCapacity: round_buffer.len(), + targetSectionSize: round_buffer.len(), + rsyncable: 0, + ..ZSTDMT_compressStreamContextProjection::default() + }; + let input = ZSTDMT_streamInputProjection { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let output = ZSTDMT_streamOutputProjection::default(); + let mut create_end = None; + let mut flush_end = None; + let mut flush_block = None; + + let result = unsafe { + compress_stream_generic_with( + context, + input, + output, + ZSTD_E_END, + |_projection| panic!("input range should already be available"), + |_src, size| { + assert_eq!(size, round_buffer.len()); + 1 + }, + |src_size, end| { + create_end = Some((src_size, end)); + 0 + }, + |_dst, _size, output_pos, block_to_flush, end| { + flush_block = Some(block_to_flush); + flush_end = Some(end); + ZSTDMT_streamFlushResult { + result: 0, + outputPos: output_pos, + } + }, + |_context, input, projection| ZSTDMT_syncPointProjection { + toLoad: (input.size - input.pos) + .min(round_buffer.len() - projection.bufferFilled), + flush: 0, + }, + ) + }; + + assert_eq!(create_end, Some((round_buffer.len(), ZSTD_E_FLUSH))); + assert_eq!(flush_end, Some(ZSTD_E_FLUSH)); + assert_eq!(flush_block, Some(0)); + assert_eq!(result.result, 1); + assert_eq!(result.inputPos, round_buffer.len()); + } + + #[test] + fn outer_scheduler_resubmits_ready_job_and_blocks_for_flush_progress() { + let context = ZSTDMT_compressStreamContextProjection { + jobReady: 1, + targetSectionSize: 4, + ..ZSTDMT_compressStreamContextProjection::default() + }; + let input_bytes = [0x31u8, 0x32]; + let input = ZSTDMT_streamInputProjection { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let output = ZSTDMT_streamOutputProjection::default(); + let mut create_call = None; + let mut flush_call = None; + + let result = unsafe { + compress_stream_generic_with( + context, + input, + output, + ZSTD_E_CONTINUE, + |_projection| panic!("a ready job must skip input acquisition"), + |_src, _size| panic!("a ready job must skip input copying"), + |src_size, end| { + create_call = Some((src_size, end)); + 0 + }, + |_dst, _size, output_pos, block_to_flush, end| { + flush_call = Some((block_to_flush, end)); + ZSTDMT_streamFlushResult { + result: 1, + outputPos: output_pos, + } + }, + |_context, _input, _projection| panic!("ready job must skip sync scan"), + ) + }; + + assert_eq!(create_call, Some((0, ZSTD_E_CONTINUE))); + assert_eq!(flush_call, Some((1, ZSTD_E_CONTINUE))); + assert_eq!(result.result, 1); + assert_eq!(result.inputPos, 0); + } + + #[test] + fn outer_scheduler_stops_before_flush_on_job_error() { + let mut round_buffer = [0xa5u8; 2]; + let input_bytes = [0x41u8, 0x42]; + let context = ZSTDMT_compressStreamContextProjection { + inBuffStart: round_buffer.as_mut_ptr().cast(), + inBuffCapacity: round_buffer.len(), + targetSectionSize: round_buffer.len(), + ..ZSTDMT_compressStreamContextProjection::default() + }; + let input = ZSTDMT_streamInputProjection { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let output = ZSTDMT_streamOutputProjection::default(); + let mut flush_called = false; + + let result = unsafe { + compress_stream_generic_with( + context, + input, + output, + ZSTD_E_CONTINUE, + |_projection| panic!("input range should already be available"), + |_src, size| { + assert_eq!(size, round_buffer.len()); + 1 + }, + |_src_size, _end| ERROR(ZstdErrorCode::DstSizeTooSmall), + |_dst, _size, output_pos, _block_to_flush, _end| { + flush_called = true; + ZSTDMT_streamFlushResult { + result: 0, + outputPos: output_pos, + } + }, + |_context, input, projection| ZSTDMT_syncPointProjection { + toLoad: (input.size - input.pos) + .min(round_buffer.len() - projection.bufferFilled), + flush: 0, + }, + ) + }; + + assert_eq!(result.result, ERROR(ZstdErrorCode::DstSizeTooSmall)); + assert_eq!(result.inputPos, round_buffer.len()); + assert_eq!(result.outputPos, 0); + assert!(!flush_called); + } + #[test] fn empty_last_block_serializes_after_buffer_callback() { let mut storage = vec![0xa5u8; 3];