diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 21f393efe..095292b00 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -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 ( diff --git a/rust/README.md b/rust/README.md index dbee6ac76..5ec857d0b 100644 --- a/rust/README.md +++ b/rust/README.md @@ -65,9 +65,13 @@ zstd ABI: Rust while C retains the private contexts and mutation callbacks. The public sequence APIs likewise use Rust-owned validation, frame-header, checksum, and output-accounting orchestration around C-owned block state. - The public end-of-frame path and the `ZSTD_compress2_c` fallback also use - Rust-owned orchestration boundaries while C retains the context reset, - stream adapter, checksum/epilogue, and trace callbacks. + The public end-of-frame path, the `ZSTD_compress2_c` fallback, and the + `ZSTD_compressStream2_c` fallback also use Rust-owned orchestration + boundaries while C retains the context reset, stream adapter, + checksum/epilogue, and trace callbacks. The stream2 fallback keeps + transparent initialization/defer, buffer validation, serial-versus-MT + dispatch, and result accounting in Rust; C supplies only the private CCtx + callbacks and projected state. Public advanced one-shot compression now uses the same Rust-owned begin-then-end ordering boundary while C retains the private begin/end callbacks. @@ -151,8 +155,8 @@ zstd ABI: destination-name construction, and the private file/resource/format callbacks for these scheduler boundaries. Rust already owns the file preference policy, filename decisions, - source/destination opening, dictionary buffers, asynchronous I/O pools, - and pass-through copy leaf. + source/destination opening and destination retry ordering, dictionary + buffers, asynchronous I/O pools, and pass-through copy leaf. - `timefn` provides the monotonic nanosecond clock behind `UTIL_time_t`, while `benchfn` owns the benchmark run/timing loop (`BMK_benchFunction`, `BMK_benchTimedFn`) and `benchzstd` owns benchmark orchestration and diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 9111f2362..554e6a42f 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -895,6 +895,373 @@ pub unsafe extern "C" fn ZSTD_rust_compressStream2MTCoordinator( unsafe { compress_stream2_mt_coordinator(state) } } +type CompressStream2InitFn = unsafe extern "C" fn(*mut c_void, c_int, usize) -> usize; +type CompressStream2CheckBufferFn = + unsafe extern "C" fn(*mut c_void, *const c_void, *const c_void) -> usize; +type CompressStream2StreamFn = + unsafe extern "C" fn(*mut c_void, *mut ZSTD_outBuffer, *mut ZSTD_inBuffer, c_int) -> usize; +type CompressStream2UpdateMTFn = unsafe extern "C" fn(*mut c_void); + +/// Scalar and private-state projections for the fallback half of the public +/// `ZSTD_compressStream2` entry point. The C shim retains context layout and +/// codec calls behind callbacks; Rust owns validation, initialization staging, +/// and the MT-versus-ST dispatch order. +#[repr(C)] +pub struct ZSTD_rust_compressStream2Fields { + callback_context: *mut c_void, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + stream_stage: *mut c_int, + stable_in_not_consumed: *mut usize, + expected_in_buffer: *mut ZSTD_inBuffer, + simple_compress2_completed: *mut c_uint, + c_params_changed: *mut c_int, + out_buff_content_size: *const usize, + out_buff_flushed_size: *const usize, + pledged_src_size_plus_one: u64, + in_buffer_mode: c_int, + out_buffer_mode: c_int, + format: c_int, + nb_workers: c_int, +} + +#[repr(C)] +pub struct ZSTD_rust_compressStream2Callbacks { + init_compress_stream2: Option, + check_buffer_stability: Option, + set_buffer_expectations: Option, + compress_stream: Option, + update_mt_cparams: Option, + mt_step: Option, + mt_trace: Option, + mt_reset: Option, +} + +#[repr(C)] +pub struct ZSTD_rust_compressStream2State { + fields: *const ZSTD_rust_compressStream2Fields, + callbacks: *const ZSTD_rust_compressStream2Callbacks, + end_op: c_int, +} + +const _: () = { + assert!(offset_of!(ZSTD_rust_compressStream2Fields, callback_context) == 0); + assert!(offset_of!(ZSTD_rust_compressStream2Fields, output) == size_of::()); + assert!(offset_of!(ZSTD_rust_compressStream2Fields, input) == 2 * size_of::()); + assert!(offset_of!(ZSTD_rust_compressStream2Fields, stream_stage) == 3 * size_of::()); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, stable_in_not_consumed) + == 4 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, expected_in_buffer) == 5 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, simple_compress2_completed) + == 6 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, c_params_changed) == 7 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, out_buff_content_size) + == size_of::<[usize; 8]>() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, out_buff_flushed_size) + == 9 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, pledged_src_size_plus_one) + == 10 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, in_buffer_mode) + == 10 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, out_buffer_mode) + == 10 * size_of::() + size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, format) + == 10 * size_of::() + size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2Fields, nb_workers) + == 10 * size_of::() + size_of::() + 3 * size_of::() + ); + assert!( + size_of::() + == (10 * size_of::() + size_of::() + 4 * size_of::()) + .div_ceil(size_of::()) + * size_of::() + ); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(offset_of!(ZSTD_rust_compressStream2State, fields) == 0); + assert!(offset_of!(ZSTD_rust_compressStream2State, callbacks) == size_of::()); + assert!(offset_of!(ZSTD_rust_compressStream2State, end_op) == 2 * size_of::()); + assert!( + size_of::() + == (2 * size_of::() + size_of::()).div_ceil(size_of::()) + * size_of::() + ); +}; + +#[cfg(not(test))] +#[inline] +unsafe fn simple_compress2_level(cctx: *const c_void, src_size: usize) -> c_int { + unsafe { ZSTD_rust_simpleCompress2Level(cctx, src_size) } +} + +#[cfg(test)] +#[inline] +unsafe fn simple_compress2_level(_cctx: *const c_void, _src_size: usize) -> c_int { + c_int::MIN +} + +#[inline] +unsafe fn compress_stream2_fallback(state: &ZSTD_rust_compressStream2State) -> usize { + let Some(fields) = (unsafe { state.fields.as_ref() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(callbacks) = (unsafe { state.callbacks.as_ref() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(output) = (unsafe { fields.output.as_mut() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(input) = (unsafe { fields.input.as_mut() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(stream_stage) = (unsafe { fields.stream_stage.as_mut() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(stable_in_not_consumed) = (unsafe { fields.stable_in_not_consumed.as_mut() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(expected_in_buffer) = (unsafe { fields.expected_in_buffer.as_mut() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(simple_compress2_completed) = (unsafe { fields.simple_compress2_completed.as_mut() }) + else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(out_buff_content_size) = (unsafe { fields.out_buff_content_size.as_ref() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(out_buff_flushed_size) = (unsafe { fields.out_buff_flushed_size.as_ref() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(init_compress_stream2) = callbacks.init_compress_stream2 else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(check_buffer_stability) = callbacks.check_buffer_stability else { + return ERROR(ZstdErrorCode::Generic); + }; + let Some(set_buffer_expectations) = callbacks.set_buffer_expectations else { + return ERROR(ZstdErrorCode::Generic); + }; + if fields.callback_context.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + if output.pos > output.size { + return ERROR(ZstdErrorCode::DstSizeTooSmall); + } + if input.pos > input.size { + return ERROR(ZstdErrorCode::SrcSizeWrong); + } + if compress_stream2_policy(state.end_op) == 0 { + return ERROR(ZstdErrorCode::ParameterOutOfBound); + } + + if *simple_compress2_completed != 0 { + if state.end_op == ZSTD_E_END + && *stream_stage == ZSTD_CSTREAM_STAGE_INIT + && fields.pledged_src_size_plus_one == 0 + { + let src_size = input.size - input.pos; + let level = + unsafe { simple_compress2_level(fields.callback_context.cast_const(), src_size) }; + if level != c_int::MIN { + 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 result = unsafe { + ZSTD_compress2( + fields.callback_context, + dst, + output.size - output.pos, + src, + src_size, + ) + }; + *simple_compress2_completed = 0; + if !ERR_isError(result) { + output.pos += result; + input.pos = input.size; + return 0; + } + } + } + *simple_compress2_completed = 0; + } + + if *stream_stage == ZSTD_CSTREAM_STAGE_INIT { + let input_size = input.size - input.pos; + let init_policy = compress_stream2_init_policy(&ZSTD_rust_compressStream2InitPolicyState { + in_buffer_mode: fields.in_buffer_mode, + end_op: state.end_op, + input_size, + stable_in_not_consumed: *stable_in_not_consumed, + input_src: input.src, + input_pos: input.pos, + expected_input_src: expected_in_buffer.src, + expected_input_size: expected_in_buffer.size, + format: fields.format, + }); + match init_policy { + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID => { + return ERROR(ZstdErrorCode::StabilityConditionNotRespected) + } + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID => { + return ERROR(ZstdErrorCode::StabilityConditionNotRespected) + } + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS + | ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD => { + input.pos = input.size; + expected_in_buffer.src = input.src; + expected_in_buffer.size = input.size; + expected_in_buffer.pos = input.pos; + *stable_in_not_consumed = stable_in_not_consumed.wrapping_add(input_size); + return init_policy as usize; + } + _ => {} + } + + let init_result = unsafe { + init_compress_stream2( + fields.callback_context, + state.end_op, + input_size.wrapping_add(*stable_in_not_consumed), + ) + }; + if ERR_isError(init_result) { + return init_result; + } + unsafe { + set_buffer_expectations( + fields.callback_context, + output as *const ZSTD_outBuffer as *const c_void, + input as *const ZSTD_inBuffer as *const c_void, + ) + }; + } + + let stability_result = unsafe { + check_buffer_stability( + fields.callback_context, + output as *const ZSTD_outBuffer as *const c_void, + input as *const ZSTD_inBuffer as *const c_void, + ) + }; + if ERR_isError(stability_result) { + return stability_result; + } + + if fields.nb_workers > 0 { + if !fields.c_params_changed.is_null() { + let c_params_changed = unsafe { &mut *fields.c_params_changed }; + if *c_params_changed != 0 { + let Some(update_mt_cparams) = callbacks.update_mt_cparams else { + return ERROR(ZstdErrorCode::Generic); + }; + unsafe { update_mt_cparams(fields.callback_context) }; + *c_params_changed = 0; + } + } + if *stable_in_not_consumed != 0 { + if input.pos < *stable_in_not_consumed { + return ERROR(ZstdErrorCode::Generic); + } + input.pos -= *stable_in_not_consumed; + *stable_in_not_consumed = 0; + } + let (Some(mt_step), Some(mt_trace), Some(mt_reset)) = + (callbacks.mt_step, callbacks.mt_trace, callbacks.mt_reset) + else { + return ERROR(ZstdErrorCode::Generic); + }; + let completion_callbacks = ZSTD_rust_compressStream2MTCompletionCallbacks { + callback_context: fields.callback_context, + trace: Some(mt_trace), + reset: Some(mt_reset), + }; + let coordinator_state = ZSTD_rust_compressStream2MTCoordinatorState { + callback_context: fields.callback_context, + step: Some(mt_step), + output, + input, + completion_callbacks: &completion_callbacks, + end_op: state.end_op, + }; + let flush_min = unsafe { ZSTD_rust_compressStream2MTCoordinator(&coordinator_state) }; + if ERR_isError(flush_min) { + return flush_min; + } + unsafe { + set_buffer_expectations( + fields.callback_context, + output as *const ZSTD_outBuffer as *const c_void, + input as *const ZSTD_inBuffer as *const c_void, + ) + }; + return flush_min; + } + + let Some(compress_stream) = callbacks.compress_stream else { + return ERROR(ZstdErrorCode::Generic); + }; + let compress_result = + unsafe { compress_stream(fields.callback_context, output, input, state.end_op) }; + let result_state = ZSTD_rust_compressStream2ResultPolicyState { + callback_context: fields.callback_context, + set_buffer_expectations: Some(set_buffer_expectations), + output: output as *const ZSTD_outBuffer as *const c_void, + input: input as *const ZSTD_inBuffer as *const c_void, + compress_result, + out_buff_content_size: *out_buff_content_size, + out_buff_flushed_size: *out_buff_flushed_size, + }; + unsafe { compress_stream2_result_policy(&result_state) } +} + +/// Run the public stream fallback while keeping private CCtx operations behind +/// the C projection callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_compressStream2( + state: *const ZSTD_rust_compressStream2State, +) -> usize { + let Some(state) = (unsafe { state.as_ref() }) else { + return ERROR(ZstdErrorCode::Generic); + }; + unsafe { compress_stream2_fallback(state) } +} + const ZSTD_C_WINDOW_LOG: c_int = 101; const ZSTD_C_HASH_LOG: c_int = 102; const ZSTD_C_CHAIN_LOG: c_int = 103; @@ -22557,4 +22924,214 @@ mod tests { ERROR(ZstdErrorCode::SrcSizeWrong) ); } + + struct Stream2FallbackProbe { + events: Vec<&'static str>, + compress_result: usize, + } + + unsafe extern "C" fn stream2_test_init( + context: *mut c_void, + _end_op: c_int, + _input_size: usize, + ) -> usize { + unsafe { &mut *context.cast::() } + .events + .push("init"); + 0 + } + + unsafe extern "C" fn stream2_test_check( + context: *mut c_void, + _output: *const c_void, + _input: *const c_void, + ) -> usize { + unsafe { &mut *context.cast::() } + .events + .push("check"); + 0 + } + + unsafe extern "C" fn stream2_test_set( + context: *mut c_void, + _output: *const c_void, + _input: *const c_void, + ) { + unsafe { &mut *context.cast::() } + .events + .push("set"); + } + + unsafe extern "C" fn stream2_test_compress( + context: *mut c_void, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + _end_op: c_int, + ) -> usize { + let probe = unsafe { &mut *context.cast::() }; + probe.events.push("compress"); + unsafe { + (*input).pos = (*input).size; + (*output).pos += 5; + } + probe.compress_result + } + + fn stream2_test_state<'a>( + probe: &'a mut Stream2FallbackProbe, + output: &'a mut ZSTD_outBuffer, + input: &'a mut ZSTD_inBuffer, + stage: &'a mut c_int, + stable_in_not_consumed: &'a mut usize, + expected_in_buffer: &'a mut ZSTD_inBuffer, + simple_compress2_completed: &'a mut c_uint, + out_buff_content_size: &'a usize, + out_buff_flushed_size: &'a usize, + ) -> ZSTD_rust_compressStream2State { + let fields = Box::new(ZSTD_rust_compressStream2Fields { + callback_context: probe as *mut Stream2FallbackProbe as *mut c_void, + output, + input, + stream_stage: stage, + stable_in_not_consumed, + expected_in_buffer, + simple_compress2_completed, + c_params_changed: ptr::null_mut(), + out_buff_content_size, + out_buff_flushed_size, + pledged_src_size_plus_one: 0, + in_buffer_mode: ZSTD_BM_BUFFERED, + out_buffer_mode: ZSTD_BM_BUFFERED, + format: ZSTD_F_ZSTD1, + nb_workers: 0, + }); + let callbacks = Box::new(ZSTD_rust_compressStream2Callbacks { + init_compress_stream2: Some(stream2_test_init), + check_buffer_stability: Some(stream2_test_check), + set_buffer_expectations: Some(stream2_test_set), + compress_stream: Some(stream2_test_compress), + update_mt_cparams: None, + mt_step: None, + mt_trace: None, + mt_reset: None, + }); + ZSTD_rust_compressStream2State { + fields: Box::into_raw(fields), + callbacks: Box::into_raw(callbacks), + end_op: ZSTD_E_CONTINUE, + } + } + + #[test] + fn stream2_fallback_preserves_serial_callback_order_and_result_policy() { + let mut probe = Stream2FallbackProbe { + events: Vec::new(), + compress_result: 7, + }; + let mut output_bytes = [0u8; 32]; + let input_bytes = [1u8; 8]; + let mut output = ZSTD_outBuffer { + dst: output_bytes.as_mut_ptr().cast(), + size: output_bytes.len(), + pos: 0, + }; + let mut input = ZSTD_inBuffer { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let mut stage = ZSTD_CSTREAM_STAGE_LOAD; + let mut stable_in_not_consumed = 0; + let mut expected_in_buffer = ZSTD_inBuffer { + src: ptr::null(), + size: 0, + pos: 0, + }; + let mut simple_compress2_completed = 0; + let out_buff_content_size = 19; + let out_buff_flushed_size = 4; + let state = stream2_test_state( + &mut probe, + &mut output, + &mut input, + &mut stage, + &mut stable_in_not_consumed, + &mut expected_in_buffer, + &mut simple_compress2_completed, + &out_buff_content_size, + &out_buff_flushed_size, + ); + + let result = unsafe { ZSTD_rust_compressStream2(&state) }; + + assert_eq!(result, 15); + assert_eq!(probe.events, ["check", "compress", "set"]); + assert_eq!(input.pos, input.size); + assert_eq!(output.pos, 5); + unsafe { + drop(Box::from_raw(state.fields.cast_mut())); + drop(Box::from_raw(state.callbacks.cast_mut())); + } + } + + #[test] + fn stream2_fallback_defers_small_stable_input_without_callbacks() { + let mut probe = Stream2FallbackProbe { + events: Vec::new(), + compress_result: 0, + }; + let input_bytes = [1u8; 16]; + let mut output = ZSTD_outBuffer { + dst: ptr::null_mut(), + size: 0, + pos: 0, + }; + let mut input = ZSTD_inBuffer { + src: input_bytes.as_ptr().cast(), + size: input_bytes.len(), + pos: 0, + }; + let mut stage = ZSTD_CSTREAM_STAGE_INIT; + let mut stable_in_not_consumed = 0; + let mut expected_in_buffer = ZSTD_inBuffer { + src: ptr::null(), + size: 0, + pos: 0, + }; + let mut simple_compress2_completed = 0; + let out_buff_content_size = 0; + let out_buff_flushed_size = 0; + let state = stream2_test_state( + &mut probe, + &mut output, + &mut input, + &mut stage, + &mut stable_in_not_consumed, + &mut expected_in_buffer, + &mut simple_compress2_completed, + &out_buff_content_size, + &out_buff_flushed_size, + ); + unsafe { + (*state.fields.cast_mut()).in_buffer_mode = ZSTD_BM_STABLE; + (*state.fields.cast_mut()).format = ZSTD_F_ZSTD1; + } + + let result = unsafe { ZSTD_rust_compressStream2(&state) }; + + assert_eq!( + result, + ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD as usize + ); + assert_eq!(input.pos, input.size); + assert_eq!(stable_in_not_consumed, input.size); + assert_eq!(expected_in_buffer.src, input.src); + assert_eq!(expected_in_buffer.size, input.size); + assert_eq!(expected_in_buffer.pos, input.size); + assert!(probe.events.is_empty()); + unsafe { + drop(Box::from_raw(state.fields.cast_mut())); + drop(Box::from_raw(state.callbacks.cast_mut())); + } + } }