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

feat(compress): move stream2 fallback orchestration to Rust

Browse Source 
The public ZSTD_compressStream2_c entry point still contained the complete
fallback state machine even though its validation, initialization policy,
serial-versus-MT dispatch, and result accounting were already represented by
Rust policy helpers. That left the main streaming entry point as a large C
orchestration island and made the Rust rewrite boundary misleading.

Project the scalar stream state and private CCtx operations through a checked
C ABI, then let Rust own the fallback ordering. Rust now validates buffers and
the end directive, handles transparent one-shot completion and stable-input
deferral, performs the initialization and stability decisions, selects the
serial or MT path, and publishes the result policy. C retains the private
context layout, stream adapter, MT operations, checksum/epilogue side effects,
and trace/reset callbacks. ABI offset assertions and focused callback-order
and stable-input tests protect the projection without requiring a standalone
Rust test binary to link every C bridge symbol.

Test Plan:
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo +nightly fmt --manifest-path rust/Cargo.toml --all` -- passed.
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` -- passed before and after formatting.
- `git diff --cached --check` -- passed.
- Native `make -j1` and the original suite remain the next post-commit verification gates.
This commit is contained in:
ddidderr
2026-07-21 13:08:28 +02:00
parent 5c6302a3c3
commit 6cf4898ab1
3 changed files with 762 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+176 -145
View File
@@ -1042,6 +1042,109 @@ typedef char ZSTD_rust_compress_stream2_mt_coordinator_state_layout[
&& sizeof(ZSTD_rust_compressStream2MTCoordinatorState)
== 6 * sizeof(void*))
? 1 : -1];
typedef size_t (*ZSTD_rust_compressStream2Init_f)(
void* context, int endOp, size_t inputSize);
typedef size_t (*ZSTD_rust_compressStream2CheckBuffer_f)(
void* context, const void* output, const void* input);
typedef size_t (*ZSTD_rust_compressStream2Stream_f)(
void* context, ZSTD_outBuffer* output, ZSTD_inBuffer* input, int endOp);
typedef void (*ZSTD_rust_compressStream2UpdateMT_f)(void* context);
typedef struct {
void* callbackContext;
ZSTD_outBuffer* output;
ZSTD_inBuffer* input;
int* streamStage;
size_t* stableInNotConsumed;
ZSTD_inBuffer* expectedInBuffer;
unsigned* simpleCompress2Completed;
int* cParamsChanged;
const size_t* outBuffContentSize;
const size_t* outBuffFlushedSize;
unsigned long long pledgedSrcSizePlusOne;
int inBufferMode;
int outBufferMode;
int format;
int nbWorkers;
} ZSTD_rust_compressStream2Fields;
typedef struct {
ZSTD_rust_compressStream2Init_f initCompressStream2;
ZSTD_rust_compressStream2CheckBuffer_f checkBufferStability;
ZSTD_rust_compressStream2SetBufferExpectations_f setBufferExpectations;
ZSTD_rust_compressStream2Stream_f compressStream;
ZSTD_rust_compressStream2UpdateMT_f updateMTCParams;
ZSTD_rust_compressStream2MTStep_f mtStep;
ZSTD_rust_compressStream2MTTrace_f mtTrace;
ZSTD_rust_compressStream2MTReset_f mtReset;
} ZSTD_rust_compressStream2Callbacks;
typedef struct {
const ZSTD_rust_compressStream2Fields* fields;
const ZSTD_rust_compressStream2Callbacks* callbacks;
int endOp;
} ZSTD_rust_compressStream2State;
size_t ZSTD_rust_compressStream2(
const ZSTD_rust_compressStream2State* state);
typedef char ZSTD_rust_compress_stream2_fields_layout[
(offsetof(ZSTD_rust_compressStream2Fields, callbackContext) == 0
&& offsetof(ZSTD_rust_compressStream2Fields, output)
== sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, input)
== 2 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, streamStage)
== 3 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, stableInNotConsumed)
== 4 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, expectedInBuffer)
== 5 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, simpleCompress2Completed)
== 6 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, cParamsChanged)
== 7 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, outBuffContentSize)
== 8 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, outBuffFlushedSize)
== 9 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, pledgedSrcSizePlusOne)
== 10 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Fields, inBufferMode)
== 10 * sizeof(void*) + sizeof(unsigned long long)
&& offsetof(ZSTD_rust_compressStream2Fields, outBufferMode)
== 10 * sizeof(void*) + sizeof(unsigned long long) + sizeof(int)
&& offsetof(ZSTD_rust_compressStream2Fields, format)
== 10 * sizeof(void*) + sizeof(unsigned long long) + 2 * sizeof(int)
&& offsetof(ZSTD_rust_compressStream2Fields, nbWorkers)
== 10 * sizeof(void*) + sizeof(unsigned long long) + 3 * sizeof(int)
&& sizeof(ZSTD_rust_compressStream2Fields)
== ((10 * sizeof(void*) + sizeof(unsigned long long) + 4 * sizeof(int)
+ sizeof(void*) - 1) / sizeof(void*)) * sizeof(void*))
? 1 : -1];
typedef char ZSTD_rust_compress_stream2_callbacks_layout[
(offsetof(ZSTD_rust_compressStream2Callbacks, initCompressStream2) == 0
&& offsetof(ZSTD_rust_compressStream2Callbacks, checkBufferStability)
== sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, setBufferExpectations)
== 2 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, compressStream)
== 3 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, updateMTCParams)
== 4 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, mtStep)
== 5 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, mtTrace)
== 6 * sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2Callbacks, mtReset)
== 7 * sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2Callbacks) == 8 * sizeof(void*))
? 1 : -1];
typedef char ZSTD_rust_compress_stream2_state_layout[
(offsetof(ZSTD_rust_compressStream2State, fields) == 0
&& offsetof(ZSTD_rust_compressStream2State, callbacks)
== sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2State, endOp)
== 2 * sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2State)
== ((2 * sizeof(void*) + sizeof(int) + sizeof(void*) - 1)
/ sizeof(void*)) * sizeof(void*))
? 1 : -1];
int ZSTD_rust_simpleCompress2Level(const void* cctx, size_t srcSize);
typedef struct {
int format;
@@ -8769,157 +8872,85 @@ static size_t ZSTD_rust_compressStream2MT_step(
}
#endif /* ZSTD_MULTITHREAD */
static size_t ZSTD_rust_compressStream2_init(void* context, int endOp, size_t inputSize)
{
return ZSTD_CCtx_init_compressStream2(
(ZSTD_CCtx*)context, (ZSTD_EndDirective)endOp, inputSize);
}
static size_t ZSTD_rust_compressStream2_checkBuffer(
void* context, const void* output, const void* input)
{
return ZSTD_checkBufferStability(
(const ZSTD_CCtx*)context,
(const ZSTD_outBuffer*)output,
(const ZSTD_inBuffer*)input);
}
static size_t ZSTD_rust_compressStream2_compressStream(
void* context, ZSTD_outBuffer* output, ZSTD_inBuffer* input, int endOp)
{
return ZSTD_compressStream_generic(
(ZSTD_CStream*)context, output, input, (ZSTD_EndDirective)endOp);
}
#ifdef ZSTD_MULTITHREAD
static void ZSTD_rust_compressStream2_updateMTCParams(void* context)
{
ZSTD_CCtx* const cctx = (ZSTD_CCtx*)context;
ZSTDMT_updateCParams_whileCompressing(cctx->mtctx, &cctx->requestedParams);
}
#endif
size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
ZSTD_outBuffer* output,
ZSTD_inBuffer* input,
ZSTD_EndDirective endOp)
{
DEBUGLOG(5, "ZSTD_compressStream2, endOp=%u ", (unsigned)endOp);
/* check conditions */
{ ZSTD_rust_compressStream2BufferPolicyState const state = {
output->pos, output->size, input->pos, input->size};
int const bufferPolicy = ZSTD_rust_compressStream2BufferPolicy(&state);
RETURN_ERROR_IF(bufferPolicy == ZSTD_RUST_COMPRESS_STREAM2_BUFFER_OUTPUT_INVALID,
dstSize_tooSmall, "invalid output buffer");
RETURN_ERROR_IF(bufferPolicy == ZSTD_RUST_COMPRESS_STREAM2_BUFFER_INPUT_INVALID,
srcSize_wrong, "invalid input buffer");
}
{ ZSTD_rust_compressStream2PolicyState const state = {(int)endOp};
RETURN_ERROR_IF(!ZSTD_rust_compressStream2Policy(&state),
parameter_outOfBound, "invalid endDirective");
}
assert(cctx != NULL);
/* transparent initialization stage */
if (cctx->rustSimpleCompress2Completed) {
if (endOp == ZSTD_e_end
&& cctx->streamStage == zcss_init
&& cctx->pledgedSrcSizePlusOne == 0
&& ZSTD_rust_simpleCompress2Level(
cctx, input->size - input->pos) != (-2147483647 - 1)) {
void* const dst = output->dst ?
(char*)output->dst + output->pos : output->dst;
const void* const src = input->src ?
(const char*)input->src + input->pos : input->src;
size_t const result = ZSTD_compress2(
cctx, dst, output->size - output->pos,
src, input->size - input->pos);
cctx->rustSimpleCompress2Completed = 0;
if (!ZSTD_isError(result)) {
output->pos += result;
input->pos = input->size;
return 0;
}
}
/* A different streaming directive or an advanced parameter means the
* frame can no longer reuse the one-shot Rust result. */
cctx->rustSimpleCompress2Completed = 0;
}
if (cctx->streamStage == zcss_init) {
size_t const inputSize = input->size - input->pos; /* no obligation to start from pos==0 */
ZSTD_rust_compressStream2InitPolicyState const state = {
(int)cctx->requestedParams.inBufferMode,
(int)endOp,
inputSize,
cctx->stableIn_notConsumed,
input->src,
input->pos,
cctx->expectedInBuffer.src,
cctx->expectedInBuffer.size,
(int)cctx->requestedParams.format
};
int const initPolicy = ZSTD_rust_compressStream2InitPolicy(&state);
RETURN_ERROR_IF(
initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID,
stabilityCondition_notRespected,
"stableInBuffer condition not respected: wrong src pointer");
RETURN_ERROR_IF(
initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID,
stabilityCondition_notRespected,
"stableInBuffer condition not respected: externally modified pos");
if (initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS
|| initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD) {
/* pretend input was consumed, to give a sense forward progress */
input->pos = input->size;
/* save stable inBuffer, for later control, and flush/end */
cctx->expectedInBuffer = *input;
/* but actually input wasn't consumed, so keep track of position from where compression shall resume */
cctx->stableIn_notConsumed += inputSize;
/* don't initialize yet, wait for the first block of flush() order, for better parameters adaptation */
return (size_t)initPolicy; /* at least some header to produce */
}
FORWARD_IF_ERROR(
ZSTD_CCtx_init_compressStream2(
cctx, endOp,
inputSize + cctx->stableIn_notConsumed),
"compressStream2 initialization failed");
ZSTD_setBufferExpectations(cctx, output, input); /* Set initial buffer expectations now that we've initialized */
}
/* end of transparent initialization stage */
FORWARD_IF_ERROR(ZSTD_checkBufferStability(cctx, output, input), "invalid buffers");
/* compression stage */
ZSTD_rust_compressStream2Fields const fields = {
cctx,
output,
input,
(int*)&cctx->streamStage,
&cctx->stableIn_notConsumed,
&cctx->expectedInBuffer,
&cctx->rustSimpleCompress2Completed,
&cctx->cParamsChanged,
&cctx->outBuffContentSize,
&cctx->outBuffFlushedSize,
cctx->pledgedSrcSizePlusOne,
(int)cctx->requestedParams.inBufferMode,
(int)cctx->requestedParams.outBufferMode,
(int)cctx->requestedParams.format,
#ifdef ZSTD_MULTITHREAD
if (cctx->appliedParams.nbWorkers > 0) {
size_t flushMin;
if (cctx->cParamsChanged) {
ZSTDMT_updateCParams_whileCompressing(cctx->mtctx, &cctx->requestedParams);
cctx->cParamsChanged = 0;
}
if (cctx->stableIn_notConsumed) {
assert(cctx->appliedParams.inBufferMode == ZSTD_bm_stable);
/* some early data was skipped - make it available for consumption */
assert(input->pos >= cctx->stableIn_notConsumed);
input->pos -= cctx->stableIn_notConsumed;
cctx->stableIn_notConsumed = 0;
}
{ ZSTD_rust_compressStream2MTCompletionCallbacks const callbacks = {
cctx,
ZSTD_rust_compressStream2MT_trace,
ZSTD_rust_compressStream2MT_reset
};
ZSTD_rust_compressStream2MTCoordinatorState const state = {
cctx,
ZSTD_rust_compressStream2MT_step,
output,
input,
&callbacks,
(int)endOp
};
flushMin = ZSTD_rust_compressStream2MTCoordinator(&state);
}
FORWARD_IF_ERROR(flushMin, "ZSTDMT_compressStream_generic failed");
DEBUGLOG(5, "completed ZSTD_compressStream2 delegating to ZSTDMT_compressStream_generic");
/* Either we don't require maximum forward progress, we've finished the
* flush, or we are out of output space.
*/
assert(endOp == ZSTD_e_continue || flushMin == 0 || output->pos == output->size);
ZSTD_setBufferExpectations(cctx, output, input);
return flushMin;
}
#endif /* ZSTD_MULTITHREAD */
{ size_t const compressResult =
ZSTD_compressStream_generic(cctx, output, input, endOp);
if (!ERR_isError(compressResult)) {
DEBUGLOG(5, "completed ZSTD_compressStream2");
}
{ ZSTD_rust_compressStream2ResultPolicyState const state = {
cctx,
ZSTD_rust_compressStream2_setBufferExpectations,
output,
input,
compressResult,
cctx->outBuffContentSize,
cctx->outBuffFlushedSize
};
{ size_t const policyResult =
ZSTD_rust_compressStream2ResultPolicy(&state);
FORWARD_IF_ERROR(policyResult, "");
return policyResult;
}
}
}
(int)cctx->appliedParams.nbWorkers,
#else
0,
#endif
};
ZSTD_rust_compressStream2Callbacks const callbacks = {
ZSTD_rust_compressStream2_init,
ZSTD_rust_compressStream2_checkBuffer,
ZSTD_rust_compressStream2_setBufferExpectations,
ZSTD_rust_compressStream2_compressStream,
#ifdef ZSTD_MULTITHREAD
ZSTD_rust_compressStream2_updateMTCParams,
ZSTD_rust_compressStream2MT_step,
ZSTD_rust_compressStream2MT_trace,
ZSTD_rust_compressStream2MT_reset,
#else
NULL,
NULL,
NULL,
NULL,
#endif
};
ZSTD_rust_compressStream2State const state = {
&fields,
&callbacks,
(int)endOp,
};
return ZSTD_rust_compressStream2(&state);
}
size_t ZSTD_compressStream2_simpleArgs (