diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index b93870e58..4c0df881f 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -46,7 +46,12 @@ void ZSTD_rust_markSimpleCompression2Complete(void* cctx); size_t ZSTD_compress2_c(ZSTD_CCtx* cctx, void* dst, size_t dstCapacity, const void* src, size_t srcSize); +size_t ZSTD_compressStream2_c(ZSTD_CCtx* cctx, + ZSTD_outBuffer* output, + ZSTD_inBuffer* input, + ZSTD_EndDirective endOp); int ZSTD_rust_simpleCompress2Level(const void* cctx); +int ZSTD_rust_simpleCompressStream2Level(const void* cctx); /* Context-free compression-parameter selection and sizing leaves live in * Rust (rust/src/zstd_compress_params.rs). This file retains @@ -949,7 +954,12 @@ size_t ZSTD_CCtx_setParameter(ZSTD_CCtx* cctx, ZSTD_cParameter param, int value) default: RETURN_ERROR(parameter_unsupported, "unknown parameter"); } - return ZSTD_CCtxParams_setParameter(&cctx->requestedParams, param, value); + { size_t const result = ZSTD_CCtxParams_setParameter(&cctx->requestedParams, param, value); + if (!ZSTD_isError(result) && param == ZSTD_c_maxBlockSize) { + cctx->rustSimpleCompress2MaxBlockSizeSet = 1; + } + return result; + } } size_t ZSTD_CCtx_getParameter(ZSTD_CCtx const* cctx, ZSTD_cParameter param, int* value) @@ -1166,6 +1176,7 @@ size_t ZSTD_CCtx_reset(ZSTD_CCtx* cctx, ZSTD_ResetDirective reset) RETURN_ERROR_IF(cctx->streamStage != zcss_init, stage_wrong, "Reset parameters is only possible during init stage."); ZSTD_clearAllDicts(cctx); + cctx->rustSimpleCompress2MaxBlockSizeSet = 0; return ZSTD_CCtxParams_reset(&cctx->requestedParams); } return 0; @@ -4532,7 +4543,7 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx) || params->targetCBlockSize != 0 || params->srcSizeHint != 0 || params->attachDictPref != ZSTD_dictDefaultAttach - || params->literalCompressionMode != ZSTD_lcm_auto + || params->literalCompressionMode != (ZSTD_ParamSwitch_e)ZSTD_lcm_auto || params->nbWorkers != 0 || params->jobSize != 0 || params->overlapLog != 0 @@ -4553,6 +4564,7 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx) || params->postBlockSplitter != ZSTD_ps_auto || params->preBlockSplitter_level != 0 || params->maxBlockSize != 0 + || cctx->rustSimpleCompress2MaxBlockSizeSet != 0 || params->useRowMatchFinder != ZSTD_ps_auto || params->deterministicRefPrefix != 0 || params->prefetchCDictTables != ZSTD_ps_auto @@ -4570,6 +4582,22 @@ int ZSTD_rust_simpleCompress2Level(const void* opaqueCctx) return params->compressionLevel; } +/* Return the simple level only while a new stream frame can be initialized. + * Rust handles this complete-input case; all other stream states stay on the + * original implementation so partial output and advanced buffering remain + * governed by the C state machine. */ +int ZSTD_rust_simpleCompressStream2Level(const void* opaqueCctx) +{ + ZSTD_CCtx const* const cctx = (ZSTD_CCtx const*)opaqueCctx; + if (cctx == NULL + || cctx->streamStage != zcss_init + || cctx->pledgedSrcSizePlusOne != 0 + || cctx->rustSimpleCompress2Completed != 0) { + return (-2147483647 - 1); + } + return ZSTD_rust_simpleCompress2Level(opaqueCctx); +} + /* ZSTD_compress() is implemented by rust/src/zstd_compress.rs. */ @@ -5503,10 +5531,10 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx, /* @return provides a minimum amount of data remaining to be flushed from internal buffers */ -size_t ZSTD_compressStream2( ZSTD_CCtx* cctx, - ZSTD_outBuffer* output, - ZSTD_inBuffer* input, - ZSTD_EndDirective endOp) +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 */ diff --git a/lib/compress/zstd_compress_internal.h b/lib/compress/zstd_compress_internal.h index a81ae184e..b31037875 100644 --- a/lib/compress/zstd_compress_internal.h +++ b/lib/compress/zstd_compress_internal.h @@ -518,6 +518,10 @@ struct ZSTD_CCtx_s { U32 frameEnded; /* The next one-shot stream call may reuse a Rust-compressed frame. */ unsigned rustSimpleCompress2Completed; + /* A caller may explicitly set maxBlockSize to zero. Keep that request + * on the C path so it remains byte-identical to the explicit default + * maxBlockSize value. */ + unsigned rustSimpleCompress2MaxBlockSizeSet; /* Stable in/out buffer verification */ ZSTD_inBuffer expectedInBuffer; diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index a2ea3dce5..2f6f63755 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -8,14 +8,15 @@ //! The layout-independent one-shot entry point in this module drives the //! already migrated compression leaves. `ZSTD_compressCCtx` uses the same //! path after a narrow C-owned context reset; the public context lifecycle -//! and streaming APIs remain C-owned because they share the -//! configuration-dependent private `ZSTD_CCtx_s` layout. `ZSTD_compress2` -//! dispatches simple, dictionary-free contexts through the same Rust frame -//! path and retains the C implementation for advanced contexts. A following -//! one-shot `ZSTD_compressStream2(..., ZSTD_e_end)` call can reuse that path. +//! remains C-owned because it shares the configuration-dependent private +//! `ZSTD_CCtx_s` layout. `ZSTD_compress2` and the complete-input simple +//! `ZSTD_compressStream2(..., ZSTD_e_end)` path dispatch through Rust while +//! retaining the C implementation for advanced and partial-stream cases. use crate::errors::{ERR_isError, ZstdErrorCode, ERROR}; use crate::mem::MEM_writeLE24; +#[cfg(not(test))] +use crate::zstd_compress_api::ZSTD_compressBound; use crate::zstd_compress_frame::{ZSTD_rust_writeFrameHeader, ZSTD_writeLastEmptyBlock}; use crate::zstd_compress_params::{ ZSTD_rust_params_adjustCParams, ZSTD_rust_params_maxNbSeq, ZSTD_rust_params_selectCParams, @@ -40,6 +41,7 @@ unsafe extern "C" { fn ZSTD_rust_resetCCtxForSimpleCompressionSession(cctx: *mut c_void) -> usize; fn ZSTD_rust_markSimpleCompression2Complete(cctx: *mut c_void); fn ZSTD_rust_simpleCompress2Level(cctx: *const c_void) -> c_int; + fn ZSTD_rust_simpleCompressStream2Level(cctx: *const c_void) -> c_int; fn ZSTD_compress2_c( cctx: *mut c_void, dst: *mut c_void, @@ -47,12 +49,36 @@ unsafe extern "C" { src: *const c_void, src_size: usize, ) -> usize; + fn ZSTD_compressStream2_c( + cctx: *mut c_void, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + end_op: c_int, + ) -> usize; } const ZSTD_FAST: c_int = 1; const ZSTD_DFAST: c_int = 2; const ZSTD_BLOCKSIZE_MAX: usize = 1 << 17; const ZSTD_CONTENTSIZE_UNKNOWN: u64 = u64::MAX; +#[cfg(not(test))] +const ZSTD_E_END: c_int = 2; + +#[cfg(not(test))] +#[repr(C)] +pub struct ZSTD_inBuffer { + src: *const c_void, + size: usize, + pos: usize, +} + +#[cfg(not(test))] +#[repr(C)] +pub struct ZSTD_outBuffer { + dst: *mut c_void, + size: usize, + pos: usize, +} /* HUF_WORKSPACE_SIZE + (MaxSeq + 2) * sizeof(unsigned), rounded up. The * superblock leaf also accepts the larger pre-split workspace, so a fixed @@ -433,6 +459,70 @@ pub unsafe extern "C" fn ZSTD_compress2( } } +/// Compress a streaming call when the complete input and a full output bound +/// are already available for the ordinary context configuration. +/// +/// The C implementation remains the fallback for partial-output streaming, +/// `ZSTD_e_continue`/`ZSTD_e_flush`, dictionaries, and every advanced context +/// configuration. The Rust path resets the C-owned session after emitting a +/// complete frame so the same context can immediately start another frame. +#[cfg(not(test))] +#[no_mangle] +pub unsafe extern "C" fn ZSTD_compressStream2( + cctx: *mut c_void, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + end_op: c_int, +) -> usize { + if cctx.is_null() || output.is_null() || input.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + + let output_ref = unsafe { &mut *output }; + let input_ref = unsafe { &mut *input }; + if output_ref.pos > output_ref.size { + return ERROR(ZstdErrorCode::DstSizeTooSmall); + } + if input_ref.pos > input_ref.size { + return ERROR(ZstdErrorCode::SrcSizeWrong); + } + + if end_op == ZSTD_E_END { + let level = unsafe { ZSTD_rust_simpleCompressStream2Level(cctx.cast_const()) }; + if level != c_int::MIN { + let src_size = input_ref.size - input_ref.pos; + let dst_capacity = output_ref.size - output_ref.pos; + let bound = ZSTD_compressBound(src_size); + if !ERR_isError(bound) && dst_capacity >= bound { + if dst_capacity != 0 && output_ref.dst.is_null() { + return ERROR(ZstdErrorCode::DstBufferNull); + } + if src_size != 0 && input_ref.src.is_null() { + return ERROR(ZstdErrorCode::SrcSizeWrong); + } + let dst = if output_ref.dst.is_null() { + ptr::null_mut() + } else { + unsafe { output_ref.dst.cast::().add(output_ref.pos).cast() } + }; + let src = if input_ref.src.is_null() { + ptr::null() + } else { + unsafe { input_ref.src.cast::().add(input_ref.pos).cast() } + }; + let result = unsafe { compress_frame(dst, dst_capacity, src, src_size, level) }; + if !ERR_isError(result) { + output_ref.pos += result; + input_ref.pos = input_ref.size; + return unsafe { ZSTD_rust_resetCCtxForSimpleCompressionSession(cctx) }; + } + } + } + } + + unsafe { ZSTD_compressStream2_c(cctx, output, input, end_op) } +} + #[cfg(test)] mod tests { use super::*; diff --git a/tests/rustLibSmoke.c b/tests/rustLibSmoke.c index 5a246d3b7..fb24dbe64 100644 --- a/tests/rustLibSmoke.c +++ b/tests/rustLibSmoke.c @@ -24,6 +24,7 @@ int main(void) void* restored = malloc(sourceSize); size_t compressedSize; size_t restoredSize; + ZSTD_CCtx* cctx = NULL; ZSTD_DDict* ddict; size_t index; @@ -48,6 +49,103 @@ int main(void) goto cleanup; } + cctx = ZSTD_createCCtx(); + if (cctx == NULL) { + fputs("CCtx creation failed\n", stderr); + goto cleanup; + } + { + ZSTD_inBuffer input = { source, sourceSize, 0 }; + ZSTD_outBuffer output = { compressed, compressedCapacity, 0 }; + size_t const remaining = ZSTD_compressStream2( + cctx, &output, &input, ZSTD_e_end); + if (checkError(remaining, "ZSTD_compressStream2") + || remaining != 0 + || input.pos != input.size) { + fputs("streaming end call did not complete\n", stderr); + goto cleanup; + } + restoredSize = ZSTD_decompress( + restored, sourceSize, compressed, output.pos); + if (checkError(restoredSize, "ZSTD_decompress(stream)") + || restoredSize != sourceSize + || memcmp(source, restored, sourceSize) != 0) { + fputs("streaming round trip mismatch\n", stderr); + goto cleanup; + } + } + /* The Rust end path resets the C-owned session before returning. */ + { + ZSTD_inBuffer input = { source, sourceSize, 0 }; + ZSTD_outBuffer output = { compressed, compressedCapacity, 0 }; + size_t const remaining = ZSTD_compressStream2( + cctx, &output, &input, ZSTD_e_end); + if (checkError(remaining, "ZSTD_compressStream2(reuse)") + || remaining != 0 + || input.pos != input.size) { + fputs("streaming context reuse failed\n", stderr); + goto cleanup; + } + } + if (ZSTD_isError(ZSTD_CCtx_setParameter(cctx, ZSTD_c_checksumFlag, 1))) { + fputs("advanced parameter setup failed\n", stderr); + goto cleanup; + } + { + ZSTD_inBuffer input = { source, sourceSize, 0 }; + ZSTD_outBuffer output = { compressed, compressedCapacity, 0 }; + size_t const remaining = ZSTD_compressStream2( + cctx, &output, &input, ZSTD_e_end); + if (checkError(remaining, "ZSTD_compressStream2(checksum)") + || remaining != 0 + || input.pos != input.size) { + fputs("advanced stream fallback did not complete\n", stderr); + goto cleanup; + } + restoredSize = ZSTD_decompress( + restored, sourceSize, compressed, output.pos); + if (checkError(restoredSize, "ZSTD_decompress(checksum)") + || restoredSize != sourceSize + || memcmp(source, restored, sourceSize) != 0) { + fputs("advanced stream fallback mismatch\n", stderr); + goto cleanup; + } + } + if (ZSTD_CCtx_reset(cctx, ZSTD_reset_session_and_parameters) != 0) { + fputs("CCtx parameter reset failed\n", stderr); + goto cleanup; + } + if (ZSTD_CCtx_reset(cctx, ZSTD_reset_session_and_parameters) != 0) { + fputs("CCtx parameter reset failed\n", stderr); + goto cleanup; + } + { + ZSTD_inBuffer input = { source, sourceSize, 0 }; + ZSTD_outBuffer output = { compressed, compressedCapacity - 1, 0 }; + size_t const remaining = ZSTD_compressStream2( + cctx, &output, &input, ZSTD_e_end); + if (checkError(remaining, "ZSTD_compressStream2(short output)") + || remaining != 0 + || input.pos != input.size) { + fputs("short-output stream fallback did not complete\n", stderr); + goto cleanup; + } + restoredSize = ZSTD_decompress( + restored, sourceSize, compressed, output.pos); + if (checkError(restoredSize, "ZSTD_decompress(short output)") + || restoredSize != sourceSize + || memcmp(source, restored, sourceSize) != 0) { + fputs("short-output stream fallback mismatch\n", stderr); + goto cleanup; + } + } + if (ZSTD_freeCCtx(cctx) != 0) { + fputs("CCtx cleanup failed\n", stderr); + cctx = NULL; + goto cleanup; + } + cctx = NULL; + /* This must pull the Rust DDict object and resolve its C helper cycle. */ ddict = ZSTD_createDDict(NULL, 0); if (ddict == NULL || ZSTD_freeDDict(ddict) != 0) { @@ -60,6 +158,7 @@ int main(void) return 0; cleanup: + ZSTD_freeCCtx(cctx); free(restored); free(compressed); return 1;