ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(compress): move stream orchestration into Rust

Browse Source 
The high-level single-thread stream path previously combined stable-input
rewind, buffer fill, direct-versus-buffered output, pending flush, end-of-frame
reset, and progress hints in C. The multithreaded path likewise owned input
range selection, rsync and end-directive adjustments, job creation decisions,
and outer flush return policy in C.

Add explicit C/Rust projections and callbacks. Rust now drives the
single-thread state machine and the MT scheduling and flush decision sequence,
while C retains CCtx-sensitive block callbacks, reusable buffers, job
descriptors, synchronization, and worker lifecycle. Focused tests cover stable
and buffered stream behavior, errors, pending output, frame completion, and MT
scheduler branches. Sequence-store and dictionary boundaries remain unchanged.

Test Plan:
- Rust lib and CLI checks, tests, formatting, and clippy passed.
- Rust legacy feature matrix passed: 556 tests.
- Native lib and CLI builds passed; CLI tests passed: 41 tests.
- Native test-zstd, fuzzer, zstream, and decode-corpus gates passed.
This commit is contained in:
ddidderr
2026-07-19 00:14:18 +02:00
parent e3d8f61ec5
commit dbc6992f3b
4 changed files with 1621 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+128 -77
View File
@@ -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;
}