From 7282ec40466066255f7b13e216afccd3637fe61d Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 08:20:46 +0200 Subject: [PATCH] feat(cli): move optional decompression loops into Rust Move gzip, xz/LZMA, and LZ4 frame I/O loops into Rust projections while keeping zlib, liblzma, LZ4F, asynchronous pool operations, diagnostics, and private C handles in C callbacks. Preserve trailing-input accounting and format-dispatch behavior, and stop faulty LZ4 callbacks from producing an unbounded no-progress output loop while retaining the codec error value. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --lib lz4_decompress_projection_reports_errors_and_unfinished_input -- --test-threads=1 (1 passed) - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --lib -- -D warnings (passed) - ulimit -v 41943040; make -B -C tests -j1 test-cli-tests (41 passed) - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1 (507 passed before final error-value assertion) --- programs/fileio.c | 683 ++++++++++++-------- rust/src/fileio_asyncio.rs | 1221 ++++++++++++++++++++++++++++++++++++ 2 files changed, 1643 insertions(+), 261 deletions(-) diff --git a/programs/fileio.c b/programs/fileio.c index 394d8ba5f..f75b75024 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -466,6 +466,107 @@ int FIO_rust_finishDecompressFrames(int status, const char* srcFileName, U64 decodedSize, const FIO_rust_decompress_callbacks_t* callbacks); + +typedef void (*FIO_rust_decompress_read_fill_fn)( + void* opaque, size_t requested, const unsigned char** buffer, size_t* loaded); +typedef void (*FIO_rust_decompress_read_consume_fn)(void* opaque, size_t consumed); +typedef void (*FIO_rust_decompress_write_acquire_fn)( + void* opaque, void** job, unsigned char** buffer, size_t* bufferSize); +typedef void (*FIO_rust_decompress_write_enqueue_fn)( + void* opaque, void** job, size_t usedBufferSize, + unsigned char** buffer, size_t* bufferSize); +typedef void (*FIO_rust_decompress_write_release_fn)(void* opaque, void* job); +typedef void (*FIO_rust_decompress_sparse_write_end_fn)(void* opaque); + +typedef struct { + void* readOpaque; + void* writeOpaque; + size_t readBufferSize; + FIO_rust_decompress_read_fill_fn readFill; + FIO_rust_decompress_read_consume_fn readConsume; + FIO_rust_decompress_write_acquire_fn writeAcquire; + FIO_rust_decompress_write_enqueue_fn writeEnqueue; + FIO_rust_decompress_write_release_fn writeRelease; + FIO_rust_decompress_sparse_write_end_fn sparseWriteEnd; +} FIO_rust_decompress_io_projection_t; + +enum { + FIO_RUST_GZIP_DECOMPRESS_OK = 0, + FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR = 1, + FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR = 2, + FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR = 3, + FIO_RUST_GZIP_DECOMPRESS_END_ERROR = 4, + FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION = 5 +}; +typedef int (*FIO_rust_gzip_decompress_init_fn)(void* opaque); +typedef int (*FIO_rust_gzip_decompress_inflate_fn)( + void* opaque, const unsigned char* input, size_t inputSize, + unsigned char* output, size_t outputSize, int flush, + size_t* consumed, size_t* produced); +typedef int (*FIO_rust_gzip_decompress_end_fn)(void* opaque); +typedef struct { + FIO_rust_decompress_io_projection_t io; + void* zlibOpaque; + FIO_rust_gzip_decompress_init_fn zlibInit; + FIO_rust_gzip_decompress_inflate_fn zlibInflate; + FIO_rust_gzip_decompress_end_fn zlibEnd; +} FIO_rust_gzip_decompress_projection_t; +int FIO_rust_decompressGzipFrame( + const FIO_rust_gzip_decompress_projection_t* projection, + U64* frameSize, int* zlibResult); + +enum { + FIO_RUST_LZMA_DECOMPRESS_OK = 0, + FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR = 1, + FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR = 2, + FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR = 3, + FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION = 4 +}; +typedef int (*FIO_rust_lzma_decompress_init_fn)(void* opaque, int plainLzma); +typedef int (*FIO_rust_lzma_decompress_code_fn)( + void* opaque, const unsigned char* input, size_t inputSize, + unsigned char* output, size_t outputSize, int action, + size_t* consumed, size_t* produced); +typedef void (*FIO_rust_lzma_decompress_end_fn)(void* opaque); +typedef struct { + FIO_rust_decompress_io_projection_t io; + void* lzmaOpaque; + FIO_rust_lzma_decompress_init_fn lzmaInit; + FIO_rust_lzma_decompress_code_fn lzmaCode; + FIO_rust_lzma_decompress_end_fn lzmaEnd; +} FIO_rust_lzma_decompress_projection_t; +int FIO_rust_decompressLzmaFrame( + const FIO_rust_lzma_decompress_projection_t* projection, + int plainLzma, U64* frameSize, int* lzmaResult); + +enum { + FIO_RUST_LZ4_DECOMPRESS_OK = 0, + FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR = 1, + FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR = 2, + FIO_RUST_LZ4_DECOMPRESS_UNFINISHED = 3, + FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION = 4 +}; +typedef int (*FIO_rust_lz4_decompress_create_fn)( + void* opaque, unsigned version, size_t* result); +typedef int (*FIO_rust_lz4_decompress_code_fn)( + void* opaque, unsigned char* output, size_t* outputSize, + const unsigned char* input, size_t* inputSize, size_t* nextToLoad); +typedef void (*FIO_rust_lz4_decompress_free_fn)(void* opaque); +typedef void (*FIO_rust_lz4_decompress_progress_fn)(void* opaque, U64 decodedSize); +typedef struct { + FIO_rust_decompress_io_projection_t io; + void* codecOpaque; + void* progressOpaque; + unsigned version; + FIO_rust_lz4_decompress_create_fn create; + FIO_rust_lz4_decompress_code_fn code; + FIO_rust_lz4_decompress_free_fn freeContext; + FIO_rust_lz4_decompress_progress_fn progress; +} FIO_rust_lz4_decompress_projection_t; +int FIO_rust_decompressLz4Frame( + const FIO_rust_lz4_decompress_projection_t* projection, + U64* frameSize, size_t* lz4Result); + void FIO_rust_displayCompressionParameters(const FIO_prefs_t* prefs); #ifdef ZSTD_LZ4COMPRESS int FIO_rust_LZ4_GetBlockSize_FromBlockId(int id); @@ -2822,214 +2923,6 @@ FIO_decompressZstdFrames(FIO_ctx_t* const fCtx, dRess_t* ress, } -#ifdef ZSTD_GZDECOMPRESS -static unsigned long long -FIO_decompressGzFrame(dRess_t* ress, const char* srcFileName) -{ - unsigned long long outFileSize = 0; - z_stream strm; - int flush = Z_NO_FLUSH; - int decodingError = 0; - IOJob_t *writeJob = NULL; - - strm.zalloc = Z_NULL; - strm.zfree = Z_NULL; - strm.opaque = Z_NULL; - strm.next_in = 0; - strm.avail_in = 0; - /* see https://www.zlib.net/manual.html */ - if (inflateInit2(&strm, 15 /* maxWindowLogSize */ + 16 /* gzip only */) != Z_OK) - return FIO_ERROR_FRAME_DECODING; - - writeJob = AIO_WritePool_acquireJob(ress->writeCtx); - strm.next_out = (Bytef*)writeJob->buffer; - strm.avail_out = (uInt)writeJob->bufferSize; - strm.avail_in = (uInt)ress->readCtx->srcBufferLoaded; - strm.next_in = (z_const unsigned char*)ress->readCtx->srcBuffer; - - for ( ; ; ) { - int ret; - if (strm.avail_in == 0) { - AIO_ReadPool_consumeAndRefill(ress->readCtx); - if (ress->readCtx->srcBufferLoaded == 0) flush = Z_FINISH; - strm.next_in = (z_const unsigned char*)ress->readCtx->srcBuffer; - strm.avail_in = (uInt)ress->readCtx->srcBufferLoaded; - } - ret = inflate(&strm, flush); - if (ret == Z_BUF_ERROR) { - DISPLAYLEVEL(1, "zstd: %s: premature gz end \n", srcFileName); - decodingError = 1; break; - } - if (ret != Z_OK && ret != Z_STREAM_END) { - DISPLAYLEVEL(1, "zstd: %s: inflate error %d \n", srcFileName, ret); - decodingError = 1; break; - } - { size_t const decompBytes = writeJob->bufferSize - strm.avail_out; - if (decompBytes) { - writeJob->usedBufferSize = decompBytes; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - outFileSize += decompBytes; - strm.next_out = (Bytef*)writeJob->buffer; - strm.avail_out = (uInt)writeJob->bufferSize; - } - } - if (ret == Z_STREAM_END) break; - } - - AIO_ReadPool_consumeBytes(ress->readCtx, ress->readCtx->srcBufferLoaded - strm.avail_in); - - if ( (inflateEnd(&strm) != Z_OK) /* release resources ; error detected */ - && (decodingError==0) ) { - DISPLAYLEVEL(1, "zstd: %s: inflateEnd error \n", srcFileName); - decodingError = 1; - } - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - return decodingError ? FIO_ERROR_FRAME_DECODING : outFileSize; -} -#endif - -#ifdef ZSTD_LZMADECOMPRESS -static unsigned long long -FIO_decompressLzmaFrame(dRess_t* ress, - const char* srcFileName, int plain_lzma) -{ - unsigned long long outFileSize = 0; - lzma_stream strm = LZMA_STREAM_INIT; - lzma_action action = LZMA_RUN; - lzma_ret initRet; - int decodingError = 0; - IOJob_t *writeJob = NULL; - - strm.next_in = 0; - strm.avail_in = 0; - if (plain_lzma) { - initRet = lzma_alone_decoder(&strm, UINT64_MAX); /* LZMA */ - } else { - initRet = lzma_stream_decoder(&strm, UINT64_MAX, 0); /* XZ */ - } - - if (initRet != LZMA_OK) { - DISPLAYLEVEL(1, "zstd: %s: %s error %d \n", - plain_lzma ? "lzma_alone_decoder" : "lzma_stream_decoder", - srcFileName, initRet); - return FIO_ERROR_FRAME_DECODING; - } - - writeJob = AIO_WritePool_acquireJob(ress->writeCtx); - strm.next_out = (BYTE*)writeJob->buffer; - strm.avail_out = writeJob->bufferSize; - strm.next_in = (BYTE const*)ress->readCtx->srcBuffer; - strm.avail_in = ress->readCtx->srcBufferLoaded; - - for ( ; ; ) { - lzma_ret ret; - if (strm.avail_in == 0) { - AIO_ReadPool_consumeAndRefill(ress->readCtx); - if (ress->readCtx->srcBufferLoaded == 0) action = LZMA_FINISH; - strm.next_in = (BYTE const*)ress->readCtx->srcBuffer; - strm.avail_in = ress->readCtx->srcBufferLoaded; - } - ret = lzma_code(&strm, action); - - if (ret == LZMA_BUF_ERROR) { - DISPLAYLEVEL(1, "zstd: %s: premature lzma end \n", srcFileName); - decodingError = 1; break; - } - if (ret != LZMA_OK && ret != LZMA_STREAM_END) { - DISPLAYLEVEL(1, "zstd: %s: lzma_code decoding error %d \n", - srcFileName, ret); - decodingError = 1; break; - } - { size_t const decompBytes = writeJob->bufferSize - strm.avail_out; - if (decompBytes) { - writeJob->usedBufferSize = decompBytes; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - outFileSize += decompBytes; - strm.next_out = (BYTE*)writeJob->buffer; - strm.avail_out = writeJob->bufferSize; - } } - if (ret == LZMA_STREAM_END) break; - } - - AIO_ReadPool_consumeBytes(ress->readCtx, ress->readCtx->srcBufferLoaded - strm.avail_in); - lzma_end(&strm); - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - return decodingError ? FIO_ERROR_FRAME_DECODING : outFileSize; -} -#endif - -#ifdef ZSTD_LZ4DECOMPRESS -static unsigned long long -FIO_decompressLz4Frame(dRess_t* ress, const char* srcFileName) -{ - unsigned long long filesize = 0; - LZ4F_errorCode_t nextToLoad = 4; - LZ4F_decompressionContext_t dCtx; - LZ4F_errorCode_t const errorCode = LZ4F_createDecompressionContext(&dCtx, LZ4F_VERSION); - int decodingError = 0; - IOJob_t *writeJob = NULL; - - if (LZ4F_isError(errorCode)) { - DISPLAYLEVEL(1, "zstd: failed to create lz4 decompression context \n"); - return FIO_ERROR_FRAME_DECODING; - } - - writeJob = AIO_WritePool_acquireJob(ress->writeCtx); - - /* Main Loop */ - for (;nextToLoad;) { - size_t pos = 0; - size_t decodedBytes = writeJob->bufferSize; - int fullBufferDecoded = 0; - - /* Read input */ - AIO_ReadPool_fillBuffer(ress->readCtx, nextToLoad); - if(!ress->readCtx->srcBufferLoaded) break; /* reached end of file */ - - while ((pos < ress->readCtx->srcBufferLoaded) || fullBufferDecoded) { /* still to read, or still to flush */ - /* Decode Input (at least partially) */ - size_t remaining = ress->readCtx->srcBufferLoaded - pos; - decodedBytes = writeJob->bufferSize; - nextToLoad = LZ4F_decompress(dCtx, writeJob->buffer, &decodedBytes, (char*)(ress->readCtx->srcBuffer)+pos, - &remaining, NULL); - if (LZ4F_isError(nextToLoad)) { - DISPLAYLEVEL(1, "zstd: %s: lz4 decompression error : %s \n", - srcFileName, LZ4F_getErrorName(nextToLoad)); - decodingError = 1; nextToLoad = 0; break; - } - pos += remaining; - assert(pos <= ress->readCtx->srcBufferLoaded); - fullBufferDecoded = decodedBytes == writeJob->bufferSize; - - /* Write Block */ - if (decodedBytes) { - UTIL_HumanReadableSize_t hrs; - writeJob->usedBufferSize = decodedBytes; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - filesize += decodedBytes; - hrs = UTIL_makeHumanReadableSize(filesize); - DISPLAYUPDATE_PROGRESS("\rDecompressed : %.*f%s ", hrs.precision, hrs.value, hrs.suffix); - } - - if (!nextToLoad) break; - } - AIO_ReadPool_consumeBytes(ress->readCtx, pos); - } - if (nextToLoad!=0) { - DISPLAYLEVEL(1, "zstd: %s: unfinished lz4 stream \n", srcFileName); - decodingError=1; - } - - LZ4F_freeDecompressionContext(dCtx); - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - - return decodingError ? FIO_ERROR_FRAME_DECODING : filesize; -} -#endif - /* Rust owns the mixed-format loop, while these adapters keep each codec and * the private dRess_t layout in this C translation unit. */ typedef struct { @@ -3038,6 +2931,327 @@ typedef struct { const FIO_prefs_t* prefs; } FIO_rust_decompression_projection_t; +#if defined(ZSTD_GZDECOMPRESS) || defined(ZSTD_LZMADECOMPRESS) || defined(ZSTD_LZ4DECOMPRESS) +static void FIO_rust_decompressReadFill(void* opaque, size_t requested, + const unsigned char** buffer, size_t* loaded) +{ + ReadPoolCtx_t* const readCtx = (ReadPoolCtx_t*)opaque; + AIO_ReadPool_fillBuffer(readCtx, requested); + *buffer = (const unsigned char*)readCtx->srcBuffer; + *loaded = readCtx->srcBufferLoaded; +} + +static void FIO_rust_decompressReadConsume(void* opaque, size_t consumed) +{ + AIO_ReadPool_consumeBytes((ReadPoolCtx_t*)opaque, consumed); +} + +static void FIO_rust_decompressWriteAcquire(void* opaque, void** job, + unsigned char** buffer, size_t* bufferSize) +{ + IOJob_t* const writeJob = AIO_WritePool_acquireJob((WritePoolCtx_t*)opaque); + *job = writeJob; + *buffer = (unsigned char*)writeJob->buffer; + *bufferSize = writeJob->bufferSize; +} + +static void FIO_rust_decompressWriteEnqueue(void* opaque, void** job, + size_t usedBufferSize, + unsigned char** buffer, size_t* bufferSize) +{ + IOJob_t* writeJob = (IOJob_t*)*job; + writeJob->usedBufferSize = usedBufferSize; + AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); + *job = writeJob; + *buffer = (unsigned char*)writeJob->buffer; + *bufferSize = writeJob->bufferSize; + (void)opaque; +} + +static void FIO_rust_decompressWriteRelease(void* opaque, void* job) +{ + AIO_WritePool_releaseIoJob((IOJob_t*)job); + (void)opaque; +} + +static void FIO_rust_decompressSparseWriteEnd(void* opaque) +{ + AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque); +} + +static void FIO_rust_decompressInitIo(FIO_rust_decompress_io_projection_t* io, + const dRess_t* ress) +{ + memset(io, 0, sizeof(*io)); + io->readOpaque = (void*)ress->readCtx; + io->writeOpaque = (void*)ress->writeCtx; + io->readBufferSize = ZSTD_DStreamInSize(); + io->readFill = FIO_rust_decompressReadFill; + io->readConsume = FIO_rust_decompressReadConsume; + io->writeAcquire = FIO_rust_decompressWriteAcquire; + io->writeEnqueue = FIO_rust_decompressWriteEnqueue; + io->writeRelease = FIO_rust_decompressWriteRelease; + io->sparseWriteEnd = FIO_rust_decompressSparseWriteEnd; +} +#endif + +#ifdef ZSTD_GZDECOMPRESS +static int FIO_rust_gzip_decompressInit(void* opaque) +{ + z_stream* const strm = (z_stream*)opaque; + strm->zalloc = Z_NULL; + strm->zfree = Z_NULL; + strm->opaque = Z_NULL; + return inflateInit2(strm, 15 /* maxWindowLogSize */ + 16 /* gzip only */); +} + +static int FIO_rust_gzip_decompressInflate(void* opaque, + const unsigned char* input, size_t inputSize, + unsigned char* output, size_t outputSize, + int flush, size_t* consumed, size_t* produced) +{ + z_stream* const strm = (z_stream*)opaque; + size_t const inputBefore = inputSize; + size_t const outputBefore = outputSize; + int ret; + + assert(inputSize <= UINT_MAX); + assert(outputSize <= UINT_MAX); + strm->next_in = (z_const unsigned char*)input; + strm->avail_in = (uInt)inputSize; + strm->next_out = (Bytef*)output; + strm->avail_out = (uInt)outputSize; + ret = inflate(strm, flush); + *consumed = inputBefore - strm->avail_in; + *produced = outputBefore - strm->avail_out; + return ret; +} + +static int FIO_rust_gzip_decompressEnd(void* opaque) +{ + return inflateEnd((z_stream*)opaque); +} + +static int FIO_rust_decompressGzFrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + FIO_rust_gzip_decompress_projection_t codecProjection; + z_stream strm; + U64 decodedSize = 0; + int zlibResult = Z_OK; + int status; + + (void)alreadyDecoded; + (void)mode; + memset(&strm, 0, sizeof(strm)); + memset(&codecProjection, 0, sizeof(codecProjection)); + FIO_rust_decompressInitIo(&codecProjection.io, projection->ress); + codecProjection.zlibOpaque = &strm; + codecProjection.zlibInit = FIO_rust_gzip_decompressInit; + codecProjection.zlibInflate = FIO_rust_gzip_decompressInflate; + codecProjection.zlibEnd = FIO_rust_gzip_decompressEnd; + + status = FIO_rust_decompressGzipFrame(&codecProjection, &decodedSize, &zlibResult); + *errorCode = 0; + *frameSize = status == FIO_RUST_GZIP_DECOMPRESS_OK + ? decodedSize : FIO_ERROR_FRAME_DECODING; + switch (status) { + case FIO_RUST_GZIP_DECOMPRESS_OK: + break; + case FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR: + DISPLAYLEVEL(1, "zstd: %s: premature gz end \n", srcFileName); + break; + case FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR: + DISPLAYLEVEL(1, "zstd: %s: inflate error %d \n", srcFileName, zlibResult); + break; + case FIO_RUST_GZIP_DECOMPRESS_END_ERROR: + DISPLAYLEVEL(1, "zstd: %s: inflateEnd error \n", srcFileName); + break; + default: + assert(status == FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR + || status == FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION); + break; + } + return status != FIO_RUST_GZIP_DECOMPRESS_OK; +} +#endif + +#ifdef ZSTD_LZMADECOMPRESS +static int FIO_rust_lzma_decompressInit(void* opaque, int plainLzma) +{ + lzma_stream* const strm = (lzma_stream*)opaque; + if (plainLzma) + return (int)lzma_alone_decoder(strm, UINT64_MAX); /* LZMA */ + return (int)lzma_stream_decoder(strm, UINT64_MAX, 0); /* XZ */ +} + +static int FIO_rust_lzma_decompressCode(void* opaque, + const unsigned char* input, size_t inputSize, + unsigned char* output, size_t outputSize, + int action, size_t* consumed, size_t* produced) +{ + lzma_stream* const strm = (lzma_stream*)opaque; + size_t const inputBefore = inputSize; + size_t const outputBefore = outputSize; + lzma_ret ret; + + strm->next_in = (const uint8_t*)input; + strm->avail_in = inputSize; + strm->next_out = (uint8_t*)output; + strm->avail_out = outputSize; + ret = lzma_code(strm, (lzma_action)action); + *consumed = inputBefore - strm->avail_in; + *produced = outputBefore - strm->avail_out; + return (int)ret; +} + +static void FIO_rust_lzma_decompressEnd(void* opaque) +{ + lzma_end((lzma_stream*)opaque); +} + +static int FIO_rust_decompressLzmaFrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + FIO_rust_lzma_decompress_projection_t codecProjection; + lzma_stream strm = LZMA_STREAM_INIT; + U64 decodedSize = 0; + int lzmaResult = (int)LZMA_OK; + int status; + + (void)alreadyDecoded; + memset(&codecProjection, 0, sizeof(codecProjection)); + FIO_rust_decompressInitIo(&codecProjection.io, projection->ress); + codecProjection.lzmaOpaque = &strm; + codecProjection.lzmaInit = FIO_rust_lzma_decompressInit; + codecProjection.lzmaCode = FIO_rust_lzma_decompressCode; + codecProjection.lzmaEnd = FIO_rust_lzma_decompressEnd; + + status = FIO_rust_decompressLzmaFrame( + &codecProjection, mode, &decodedSize, &lzmaResult); + *errorCode = 0; + *frameSize = status == FIO_RUST_LZMA_DECOMPRESS_OK + ? decodedSize : FIO_ERROR_FRAME_DECODING; + switch (status) { + case FIO_RUST_LZMA_DECOMPRESS_OK: + break; + case FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR: + DISPLAYLEVEL(1, "zstd: %s: %s error %d \n", + mode ? "lzma_alone_decoder" : "lzma_stream_decoder", + srcFileName, lzmaResult); + break; + case FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR: + DISPLAYLEVEL(1, "zstd: %s: premature lzma end \n", srcFileName); + break; + case FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR: + DISPLAYLEVEL(1, "zstd: %s: lzma_code decoding error %d \n", + srcFileName, lzmaResult); + break; + default: + assert(status == FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION); + break; + } + return status != FIO_RUST_LZMA_DECOMPRESS_OK; +} +#endif + +#ifdef ZSTD_LZ4DECOMPRESS +static int FIO_rust_lz4_decompressCreate(void* opaque, unsigned version, size_t* result) +{ + LZ4F_errorCode_t const errorCode = LZ4F_createDecompressionContext( + (LZ4F_decompressionContext_t*)opaque, version); + *result = (size_t)errorCode; + return LZ4F_isError(errorCode); +} + +static int FIO_rust_lz4_decompressCode(void* opaque, unsigned char* output, + size_t* outputSize, const unsigned char* input, + size_t* inputSize, size_t* nextToLoad) +{ + LZ4F_errorCode_t const result = LZ4F_decompress( + *(LZ4F_decompressionContext_t*)opaque, output, outputSize, + input, inputSize, NULL); + *nextToLoad = (size_t)result; + return LZ4F_isError(result); +} + +static void FIO_rust_lz4_decompressFree(void* opaque) +{ + LZ4F_freeDecompressionContext(*(LZ4F_decompressionContext_t*)opaque); +} + +static void FIO_rust_lz4_decompressProgress(void* opaque, U64 decodedSize) +{ + UTIL_HumanReadableSize_t const hrs = UTIL_makeHumanReadableSize(decodedSize); + (void)opaque; + DISPLAYUPDATE_PROGRESS("\rDecompressed : %.*f%s ", + hrs.precision, hrs.value, hrs.suffix); +} + +static int FIO_rust_decompressLz4FrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + FIO_rust_lz4_decompress_projection_t codecProjection; + LZ4F_decompressionContext_t dCtx; + U64 decodedSize = 0; + size_t lz4Result = 0; + int status; + + (void)alreadyDecoded; + (void)mode; + memset(&codecProjection, 0, sizeof(codecProjection)); + FIO_rust_decompressInitIo(&codecProjection.io, projection->ress); + codecProjection.codecOpaque = &dCtx; + codecProjection.progressOpaque = NULL; + codecProjection.version = LZ4F_VERSION; + codecProjection.create = FIO_rust_lz4_decompressCreate; + codecProjection.code = FIO_rust_lz4_decompressCode; + codecProjection.freeContext = FIO_rust_lz4_decompressFree; + codecProjection.progress = FIO_rust_lz4_decompressProgress; + + status = FIO_rust_decompressLz4Frame(&codecProjection, &decodedSize, &lz4Result); + *errorCode = 0; + *frameSize = status == FIO_RUST_LZ4_DECOMPRESS_OK + ? decodedSize : FIO_ERROR_FRAME_DECODING; + switch (status) { + case FIO_RUST_LZ4_DECOMPRESS_OK: + break; + case FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR: + DISPLAYLEVEL(1, "zstd: failed to create lz4 decompression context \n"); + break; + case FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR: + DISPLAYLEVEL(1, "zstd: %s: lz4 decompression error : %s \n", + srcFileName, LZ4F_getErrorName(lz4Result)); + break; + case FIO_RUST_LZ4_DECOMPRESS_UNFINISHED: + DISPLAYLEVEL(1, "zstd: %s: unfinished lz4 stream \n", srcFileName); + break; + default: + assert(status == FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION); + break; + } + return status != FIO_RUST_LZ4_DECOMPRESS_OK; +} +#endif + static int FIO_rust_decompressZstdFrameCallback(void* opaque, const char* srcFileName, U64 alreadyDecoded, @@ -3057,59 +3271,6 @@ static int FIO_rust_decompressZstdFrameCallback(void* opaque, return *frameSize == FIO_ERROR_FRAME_DECODING; } -#ifdef ZSTD_GZDECOMPRESS -static int FIO_rust_decompressGzFrameCallback(void* opaque, - const char* srcFileName, - U64 alreadyDecoded, - U64* frameSize, - size_t* errorCode, - int mode) -{ - FIO_rust_decompression_projection_t* const projection = - (FIO_rust_decompression_projection_t*)opaque; - (void)alreadyDecoded; - (void)mode; - *errorCode = 0; - *frameSize = FIO_decompressGzFrame(projection->ress, srcFileName); - return *frameSize == FIO_ERROR_FRAME_DECODING; -} -#endif - -#ifdef ZSTD_LZMADECOMPRESS -static int FIO_rust_decompressLzmaFrameCallback(void* opaque, - const char* srcFileName, - U64 alreadyDecoded, - U64* frameSize, - size_t* errorCode, - int mode) -{ - FIO_rust_decompression_projection_t* const projection = - (FIO_rust_decompression_projection_t*)opaque; - (void)alreadyDecoded; - *errorCode = 0; - *frameSize = FIO_decompressLzmaFrame(projection->ress, srcFileName, mode); - return *frameSize == FIO_ERROR_FRAME_DECODING; -} -#endif - -#ifdef ZSTD_LZ4DECOMPRESS -static int FIO_rust_decompressLz4FrameCallback(void* opaque, - const char* srcFileName, - U64 alreadyDecoded, - U64* frameSize, - size_t* errorCode, - int mode) -{ - FIO_rust_decompression_projection_t* const projection = - (FIO_rust_decompression_projection_t*)opaque; - (void)alreadyDecoded; - (void)mode; - *errorCode = 0; - *frameSize = FIO_decompressLz4Frame(projection->ress, srcFileName); - return *frameSize == FIO_ERROR_FRAME_DECODING; -} -#endif - static int FIO_rust_decompressPassThroughCallback(void* opaque) { FIO_rust_decompression_projection_t* const projection = diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 3eb74de33..cb5242ae4 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -79,6 +79,140 @@ pub struct FIO_rust_decompress_callbacks_t { pub finish: Option, } +pub type FIO_rust_decompress_read_fill_fn = + unsafe extern "C" fn(*mut c_void, usize, *mut *const u8, *mut usize); +pub type FIO_rust_decompress_read_consume_fn = unsafe extern "C" fn(*mut c_void, usize); +pub type FIO_rust_decompress_write_acquire_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, *mut *mut u8, *mut usize); +pub type FIO_rust_decompress_write_enqueue_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, usize, *mut *mut u8, *mut usize); +pub type FIO_rust_decompress_write_release_fn = unsafe extern "C" fn(*mut c_void, *mut c_void); +pub type FIO_rust_decompress_sparse_write_end_fn = unsafe extern "C" fn(*mut c_void); + +/// The optional-format leaves use the same opaque asynchronous I/O projection. +/// `read_fill(0, ...)` returns the bytes already staged by the mixed-format +/// dispatcher; non-zero requests refill the C-owned read pool after the Rust +/// loop has consumed the current input. +#[repr(C)] +pub struct FIO_rust_decompress_io_projection_t { + pub read_opaque: *mut c_void, + pub write_opaque: *mut c_void, + pub read_buffer_size: usize, + pub read_fill: Option, + pub read_consume: Option, + pub write_acquire: Option, + pub write_enqueue: Option, + pub write_release: Option, + pub sparse_write_end: Option, +} + +pub const FIO_RUST_GZIP_DECOMPRESS_OK: c_int = 0; +pub const FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR: c_int = 1; +pub const FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR: c_int = 2; +pub const FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR: c_int = 3; +pub const FIO_RUST_GZIP_DECOMPRESS_END_ERROR: c_int = 4; +pub const FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION: c_int = 5; + +const FIO_RUST_GZIP_DECOMPRESS_Z_OK: c_int = 0; +const FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END: c_int = 1; +const FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR: c_int = -5; +const FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH: c_int = 0; +const FIO_RUST_GZIP_DECOMPRESS_Z_FINISH: c_int = 4; + +pub type FIO_rust_gzip_decompress_init_fn = unsafe extern "C" fn(*mut c_void) -> c_int; +pub type FIO_rust_gzip_decompress_inflate_fn = unsafe extern "C" fn( + *mut c_void, + *const u8, + usize, + *mut u8, + usize, + c_int, + *mut usize, + *mut usize, +) -> c_int; +pub type FIO_rust_gzip_decompress_end_fn = unsafe extern "C" fn(*mut c_void) -> c_int; + +/// Rust owns the gzip read/inflate/output loop. The C side retains only the +/// zlib stream and its calls, plus the opaque asynchronous pool operations. +#[repr(C)] +pub struct FIO_rust_gzip_decompress_projection_t { + pub io: FIO_rust_decompress_io_projection_t, + pub zlib_opaque: *mut c_void, + pub zlib_init: Option, + pub zlib_inflate: Option, + pub zlib_end: Option, +} + +pub const FIO_RUST_LZMA_DECOMPRESS_OK: c_int = 0; +pub const FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR: c_int = 1; +pub const FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR: c_int = 2; +pub const FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR: c_int = 3; +pub const FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION: c_int = 4; + +const FIO_RUST_LZMA_DECOMPRESS_OK_CODE: c_int = 0; +const FIO_RUST_LZMA_DECOMPRESS_STREAM_END: c_int = 1; +const FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR_CODE: c_int = 10; +const FIO_RUST_LZMA_DECOMPRESS_RUN: c_int = 0; +const FIO_RUST_LZMA_DECOMPRESS_FINISH: c_int = 3; + +pub type FIO_rust_lzma_decompress_init_fn = unsafe extern "C" fn(*mut c_void, c_int) -> c_int; +pub type FIO_rust_lzma_decompress_code_fn = unsafe extern "C" fn( + *mut c_void, + *const u8, + usize, + *mut u8, + usize, + c_int, + *mut usize, + *mut usize, +) -> c_int; +pub type FIO_rust_lzma_decompress_end_fn = unsafe extern "C" fn(*mut c_void); + +/// Rust owns the xz/LZMA read/code/output loop while liblzma's stream remains +/// an opaque C-owned handle. +#[repr(C)] +pub struct FIO_rust_lzma_decompress_projection_t { + pub io: FIO_rust_decompress_io_projection_t, + pub lzma_opaque: *mut c_void, + pub lzma_init: Option, + pub lzma_code: Option, + pub lzma_end: Option, +} + +pub const FIO_RUST_LZ4_DECOMPRESS_OK: c_int = 0; +pub const FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR: c_int = 1; +pub const FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR: c_int = 2; +pub const FIO_RUST_LZ4_DECOMPRESS_UNFINISHED: c_int = 3; +pub const FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION: c_int = 4; + +pub type FIO_rust_lz4_decompress_create_fn = + unsafe extern "C" fn(*mut c_void, c_uint, *mut usize) -> c_int; +pub type FIO_rust_lz4_decompress_code_fn = unsafe extern "C" fn( + *mut c_void, + *mut u8, + *mut usize, + *const u8, + *mut usize, + *mut usize, +) -> c_int; +pub type FIO_rust_lz4_decompress_free_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_lz4_decompress_progress_fn = unsafe extern "C" fn(*mut c_void, u64); + +/// Rust owns the LZ4 frame loop. LZ4F's context, error values, and calls are +/// kept behind callbacks so no optional-library or private-handle layout is +/// part of the Rust/C projection. +#[repr(C)] +pub struct FIO_rust_lz4_decompress_projection_t { + pub io: FIO_rust_decompress_io_projection_t, + pub codec_opaque: *mut c_void, + pub progress_opaque: *mut c_void, + pub version: c_uint, + pub create: Option, + pub code: Option, + pub free_context: Option, + pub progress: Option, +} + pub const FIO_RUST_COMPRESS_OK: c_int = 0; pub const FIO_RUST_COMPRESS_GZIP_UNSUPPORTED: c_int = 1; pub const FIO_RUST_COMPRESS_LZMA_UNSUPPORTED: c_int = 2; @@ -2686,6 +2820,512 @@ pub unsafe extern "C" fn FIO_rust_decompressZstdFrames( } } +/// Decompresses one gzip member through a C-owned zlib stream and the common +/// asynchronous I/O projection. Input is consumed immediately after each +/// inflate call; this is equivalent to the original C leaf's final +/// `avail_in` accounting while keeping any following member or format in the +/// read pool. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_decompressGzipFrame( + projection: *const FIO_rust_gzip_decompress_projection_t, + frame_size: *mut u64, + zlib_result: *mut c_int, +) -> c_int { + assert!(!projection.is_null()); + assert!(!frame_size.is_null()); + assert!(!zlib_result.is_null()); + + unsafe { + *frame_size = 0; + *zlib_result = FIO_RUST_GZIP_DECOMPRESS_Z_OK; + } + + let projection = unsafe { &*projection }; + if projection.io.read_buffer_size == 0 { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + } + let Some(read_fill) = projection.io.read_fill else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(read_consume) = projection.io.read_consume else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_acquire) = projection.io.write_acquire else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_enqueue) = projection.io.write_enqueue else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_release) = projection.io.write_release else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(sparse_write_end) = projection.io.sparse_write_end else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(zlib_init) = projection.zlib_init else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(zlib_inflate) = projection.zlib_inflate else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(zlib_end) = projection.zlib_end else { + return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION; + }; + + let init_result = unsafe { zlib_init(projection.zlib_opaque) }; + if init_result != FIO_RUST_GZIP_DECOMPRESS_Z_OK { + unsafe { *zlib_result = init_result }; + return FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR; + } + + let mut job = ptr::null_mut::(); + let mut output = ptr::null_mut::(); + let mut output_size = 0_usize; + unsafe { + write_acquire( + projection.io.write_opaque, + &mut job, + &mut output, + &mut output_size, + ); + } + assert!(!job.is_null()); + assert!(!output.is_null() || output_size == 0); + + let mut input = ptr::null(); + let mut input_size = 0_usize; + unsafe { + read_fill(projection.io.read_opaque, 0, &mut input, &mut input_size); + } + let mut flush = FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH; + let mut decoded = 0_u64; + let mut status = FIO_RUST_GZIP_DECOMPRESS_OK; + + loop { + if input_size == 0 { + unsafe { + read_fill( + projection.io.read_opaque, + projection.io.read_buffer_size, + &mut input, + &mut input_size, + ); + } + if input_size == 0 { + flush = FIO_RUST_GZIP_DECOMPRESS_Z_FINISH; + } + } + + let mut consumed = 0_usize; + let mut produced = 0_usize; + let result = unsafe { + zlib_inflate( + projection.zlib_opaque, + input, + input_size, + output, + output_size, + flush, + &mut consumed, + &mut produced, + ) + }; + assert!(consumed <= input_size); + assert!(produced <= output_size); + unsafe { read_consume(projection.io.read_opaque, consumed) }; + if consumed != 0 { + input = unsafe { input.add(consumed) }; + } + input_size -= consumed; + + if result == FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR { + unsafe { *zlib_result = result }; + status = FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR; + break; + } + if result != FIO_RUST_GZIP_DECOMPRESS_Z_OK + && result != FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END + { + unsafe { *zlib_result = result }; + status = FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR; + break; + } + + if produced != 0 { + unsafe { + write_enqueue( + projection.io.write_opaque, + &mut job, + produced, + &mut output, + &mut output_size, + ); + } + decoded = decoded.wrapping_add(produced as u64); + } + if result == FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END { + break; + } + } + + let end_result = unsafe { zlib_end(projection.zlib_opaque) }; + if status == FIO_RUST_GZIP_DECOMPRESS_OK && end_result != FIO_RUST_GZIP_DECOMPRESS_Z_OK { + unsafe { *zlib_result = end_result }; + status = FIO_RUST_GZIP_DECOMPRESS_END_ERROR; + } + + unsafe { + write_release(projection.io.write_opaque, job); + sparse_write_end(projection.io.write_opaque); + } + if status == FIO_RUST_GZIP_DECOMPRESS_OK { + unsafe { *frame_size = decoded }; + } + status +} + +/// Decompresses one xz or plain-LZMA stream through a C-owned liblzma +/// stream. The action switches to `LZMA_FINISH` only after the read pool +/// reaches EOF, matching the legacy C loop and preserving unread trailing +/// bytes after `LZMA_STREAM_END`. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_decompressLzmaFrame( + projection: *const FIO_rust_lzma_decompress_projection_t, + plain_lzma: c_int, + frame_size: *mut u64, + lzma_result: *mut c_int, +) -> c_int { + assert!(!projection.is_null()); + assert!(!frame_size.is_null()); + assert!(!lzma_result.is_null()); + + unsafe { + *frame_size = 0; + *lzma_result = FIO_RUST_LZMA_DECOMPRESS_OK_CODE; + } + + let projection = unsafe { &*projection }; + if projection.io.read_buffer_size == 0 { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + } + let Some(read_fill) = projection.io.read_fill else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(read_consume) = projection.io.read_consume else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_acquire) = projection.io.write_acquire else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_enqueue) = projection.io.write_enqueue else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_release) = projection.io.write_release else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(sparse_write_end) = projection.io.sparse_write_end else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(lzma_init) = projection.lzma_init else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(lzma_code) = projection.lzma_code else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(lzma_end) = projection.lzma_end else { + return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION; + }; + + let init_result = unsafe { lzma_init(projection.lzma_opaque, plain_lzma) }; + if init_result != FIO_RUST_LZMA_DECOMPRESS_OK_CODE { + unsafe { *lzma_result = init_result }; + return FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR; + } + + let mut job = ptr::null_mut::(); + let mut output = ptr::null_mut::(); + let mut output_size = 0_usize; + unsafe { + write_acquire( + projection.io.write_opaque, + &mut job, + &mut output, + &mut output_size, + ); + } + assert!(!job.is_null()); + assert!(!output.is_null() || output_size == 0); + + let mut input = ptr::null(); + let mut input_size = 0_usize; + unsafe { + read_fill(projection.io.read_opaque, 0, &mut input, &mut input_size); + } + let mut action = FIO_RUST_LZMA_DECOMPRESS_RUN; + let mut decoded = 0_u64; + let mut status = FIO_RUST_LZMA_DECOMPRESS_OK; + + loop { + if input_size == 0 { + unsafe { + read_fill( + projection.io.read_opaque, + projection.io.read_buffer_size, + &mut input, + &mut input_size, + ); + } + if input_size == 0 { + action = FIO_RUST_LZMA_DECOMPRESS_FINISH; + } + } + + let mut consumed = 0_usize; + let mut produced = 0_usize; + let result = unsafe { + lzma_code( + projection.lzma_opaque, + input, + input_size, + output, + output_size, + action, + &mut consumed, + &mut produced, + ) + }; + assert!(consumed <= input_size); + assert!(produced <= output_size); + unsafe { read_consume(projection.io.read_opaque, consumed) }; + if consumed != 0 { + input = unsafe { input.add(consumed) }; + } + input_size -= consumed; + + if result == FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR_CODE { + unsafe { *lzma_result = result }; + status = FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR; + break; + } + if result != FIO_RUST_LZMA_DECOMPRESS_OK_CODE + && result != FIO_RUST_LZMA_DECOMPRESS_STREAM_END + { + unsafe { *lzma_result = result }; + status = FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR; + break; + } + + if produced != 0 { + unsafe { + write_enqueue( + projection.io.write_opaque, + &mut job, + produced, + &mut output, + &mut output_size, + ); + } + decoded = decoded.wrapping_add(produced as u64); + } + if result == FIO_RUST_LZMA_DECOMPRESS_STREAM_END { + break; + } + } + + unsafe { lzma_end(projection.lzma_opaque) }; + unsafe { + write_release(projection.io.write_opaque, job); + sparse_write_end(projection.io.write_opaque); + } + if status == FIO_RUST_LZMA_DECOMPRESS_OK { + unsafe { *frame_size = decoded }; + } + status +} + +/// Decompresses one LZ4 frame while preserving LZ4F's recommended input +/// request and full-output-buffer retry protocol. A successful frame ends at +/// the first zero `next_to_load`; any EOF while that hint remains non-zero is +/// reported as an unfinished stream. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_decompressLz4Frame( + projection: *const FIO_rust_lz4_decompress_projection_t, + frame_size: *mut u64, + lz4_result: *mut usize, +) -> c_int { + assert!(!projection.is_null()); + assert!(!frame_size.is_null()); + assert!(!lz4_result.is_null()); + + unsafe { + *frame_size = 0; + *lz4_result = 0; + } + + let projection = unsafe { &*projection }; + let Some(read_fill) = projection.io.read_fill else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(read_consume) = projection.io.read_consume else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_acquire) = projection.io.write_acquire else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_enqueue) = projection.io.write_enqueue else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(write_release) = projection.io.write_release else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(sparse_write_end) = projection.io.sparse_write_end else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(create) = projection.create else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(code) = projection.code else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(free_context) = projection.free_context else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + let Some(progress) = projection.progress else { + return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION; + }; + + let mut create_result = 0_usize; + let create_status = unsafe { + create( + projection.codec_opaque, + projection.version, + &mut create_result, + ) + }; + if create_status != 0 { + unsafe { *lz4_result = create_result }; + return FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR; + } + + let mut job = ptr::null_mut::(); + let mut output = ptr::null_mut::(); + let mut output_size = 0_usize; + unsafe { + write_acquire( + projection.io.write_opaque, + &mut job, + &mut output, + &mut output_size, + ); + } + assert!(!job.is_null()); + assert!(!output.is_null() || output_size == 0); + + let mut next_to_load = 4_usize; + let mut decoded = 0_u64; + let mut status = FIO_RUST_LZ4_DECOMPRESS_OK; + + while next_to_load != 0 { + let mut input = ptr::null(); + let mut loaded = 0_usize; + unsafe { + read_fill( + projection.io.read_opaque, + next_to_load, + &mut input, + &mut loaded, + ); + } + if loaded == 0 { + break; + } + + let mut pos = 0_usize; + let mut full_buffer_decoded = false; + let mut unchanged_input_calls = 0_usize; + while pos < loaded || full_buffer_decoded { + let previous_next_to_load = next_to_load; + let mut produced = output_size; + let mut remaining = loaded - pos; + let mut next = 0_usize; + let result = unsafe { + code( + projection.codec_opaque, + output, + &mut produced, + input.add(pos), + &mut remaining, + &mut next, + ) + }; + if result != 0 { + /* The C adapter returns a boolean failure status but keeps + * LZ4F's actual error code in the next-input slot. */ + unsafe { *lz4_result = next }; + status = FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR; + next_to_load = 0; + break; + } + assert!(produced <= output_size); + assert!(remaining <= loaded - pos); + pos += remaining; + next_to_load = next; + full_buffer_decoded = produced == output_size; + + if produced != 0 { + unsafe { + write_enqueue( + projection.io.write_opaque, + &mut job, + produced, + &mut output, + &mut output_size, + ); + } + decoded = decoded.wrapping_add(produced as u64); + unsafe { progress(projection.progress_opaque, decoded) }; + } + + if next_to_load == 0 { + break; + } + + /* A codec must eventually consume input or change its next-input + * request. Without this guard, a faulty callback can keep + * producing output while receiving an empty input slice forever, + * which would turn an ordinary decode error into an unbounded + * output/memory loop. */ + if remaining == 0 && next == previous_next_to_load { + unchanged_input_calls += 1; + if unchanged_input_calls >= 2 { + status = FIO_RUST_LZ4_DECOMPRESS_UNFINISHED; + break; + } + } else { + unchanged_input_calls = 0; + } + } + unsafe { read_consume(projection.io.read_opaque, pos) }; + if status != FIO_RUST_LZ4_DECOMPRESS_OK { + break; + } + } + + if status == FIO_RUST_LZ4_DECOMPRESS_OK && next_to_load != 0 { + status = FIO_RUST_LZ4_DECOMPRESS_UNFINISHED; + } + + unsafe { + free_context(projection.codec_opaque); + write_release(projection.io.write_opaque, job); + sparse_write_end(projection.io.write_opaque); + } + if status == FIO_RUST_LZ4_DECOMPRESS_OK { + unsafe { *frame_size = decoded }; + } + status +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum DecompressionFormat { Zstd, @@ -4183,6 +4823,587 @@ mod tests { } } + #[derive(Default)] + struct OptionalDecompressIoState { + input: Vec, + input_pos: usize, + output: [u8; 4], + fill_requests: Vec, + consumed: Vec, + enqueue_sizes: Vec, + output_chunks: Vec>, + acquire_calls: usize, + release_calls: usize, + sparse_end_calls: usize, + } + + impl OptionalDecompressIoState { + fn new(input: &[u8]) -> Self { + Self { + input: input.to_vec(), + ..Self::default() + } + } + } + + unsafe extern "C" fn optional_decompress_read_fill( + opaque: *mut c_void, + requested: usize, + buffer: *mut *const u8, + loaded: *mut usize, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.fill_requests.push(requested); + let available = state.input.len() - state.input_pos; + let amount = if requested == 0 { + available + } else { + available.min(requested) + }; + unsafe { + *buffer = state.input.as_ptr().add(state.input_pos); + *loaded = amount; + } + } + + unsafe extern "C" fn optional_decompress_read_consume(opaque: *mut c_void, amount: usize) { + let state = unsafe { &mut *opaque.cast::() }; + assert!(amount <= state.input.len() - state.input_pos); + state.input_pos += amount; + state.consumed.push(amount); + } + + unsafe extern "C" fn optional_decompress_write_acquire( + opaque: *mut c_void, + job: *mut *mut c_void, + buffer: *mut *mut u8, + buffer_size: *mut usize, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.acquire_calls += 1; + unsafe { + *job = opaque; + *buffer = state.output.as_mut_ptr(); + *buffer_size = state.output.len(); + } + } + + unsafe extern "C" fn optional_decompress_write_enqueue( + opaque: *mut c_void, + job: *mut *mut c_void, + used: usize, + buffer: *mut *mut u8, + buffer_size: *mut usize, + ) { + let state = unsafe { &mut *opaque.cast::() }; + assert_eq!(unsafe { *job }, opaque); + assert!(used <= state.output.len()); + state.enqueue_sizes.push(used); + state.output_chunks.push(state.output[..used].to_vec()); + unsafe { + *buffer = state.output.as_mut_ptr(); + *buffer_size = state.output.len(); + } + } + + unsafe extern "C" fn optional_decompress_write_release(opaque: *mut c_void, job: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + assert_eq!(job, opaque); + state.release_calls += 1; + } + + unsafe extern "C" fn optional_decompress_sparse_end(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.sparse_end_calls += 1; + } + + fn optional_decompress_io( + state: &mut OptionalDecompressIoState, + read_buffer_size: usize, + ) -> FIO_rust_decompress_io_projection_t { + let opaque = (state as *mut OptionalDecompressIoState).cast::(); + FIO_rust_decompress_io_projection_t { + read_opaque: opaque, + write_opaque: opaque, + read_buffer_size, + read_fill: Some(optional_decompress_read_fill), + read_consume: Some(optional_decompress_read_consume), + write_acquire: Some(optional_decompress_write_acquire), + write_enqueue: Some(optional_decompress_write_enqueue), + write_release: Some(optional_decompress_write_release), + sparse_write_end: Some(optional_decompress_sparse_end), + } + } + + struct GzipDecompressState { + io: OptionalDecompressIoState, + init_result: c_int, + inflate_calls: usize, + stop_after_calls: usize, + finish_result: c_int, + end_result: c_int, + flushes: Vec, + end_calls: usize, + } + + impl GzipDecompressState { + fn new(input: &[u8]) -> Self { + Self { + io: OptionalDecompressIoState::new(input), + init_result: FIO_RUST_GZIP_DECOMPRESS_Z_OK, + inflate_calls: 0, + stop_after_calls: 0, + finish_result: FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END, + end_result: FIO_RUST_GZIP_DECOMPRESS_Z_OK, + flushes: Vec::new(), + end_calls: 0, + } + } + } + + unsafe extern "C" fn gzip_decompress_test_init(opaque: *mut c_void) -> c_int { + unsafe { (*opaque.cast::()).init_result } + } + + unsafe extern "C" fn gzip_decompress_test_inflate( + opaque: *mut c_void, + _input: *const u8, + input_size: usize, + output: *mut u8, + output_size: usize, + flush: c_int, + consumed: *mut usize, + produced: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.inflate_calls += 1; + state.flushes.push(flush); + unsafe { + *consumed = 0; + *produced = 0; + } + + if flush == FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH { + assert!(input_size != 0); + assert!(output_size != 0); + if state.stop_after_calls != 0 && state.inflate_calls > state.stop_after_calls { + unsafe { + *produced = 1; + *output = b'y'; + } + return FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END; + } + unsafe { + *consumed = input_size.min(2); + *produced = 1; + *output = b'x'; + } + return FIO_RUST_GZIP_DECOMPRESS_Z_OK; + } + + assert_eq!(flush, FIO_RUST_GZIP_DECOMPRESS_Z_FINISH); + assert_eq!(input_size, 0); + assert!(output_size != 0); + unsafe { + *produced = 1; + *output = b'f'; + } + state.finish_result + } + + unsafe extern "C" fn gzip_decompress_test_end(opaque: *mut c_void) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.end_calls += 1; + state.end_result + } + + fn gzip_decompress_test_projection( + state: &mut GzipDecompressState, + ) -> FIO_rust_gzip_decompress_projection_t { + let codec_opaque = (state as *mut GzipDecompressState).cast::(); + let io = optional_decompress_io(&mut state.io, 3); + FIO_rust_gzip_decompress_projection_t { + io, + zlib_opaque: codec_opaque, + zlib_init: Some(gzip_decompress_test_init), + zlib_inflate: Some(gzip_decompress_test_inflate), + zlib_end: Some(gzip_decompress_test_end), + } + } + + #[test] + fn gzip_decompress_projection_preserves_trailing_input_and_accounting() { + let mut state = GzipDecompressState::new(b"frame-tail"); + state.stop_after_calls = 1; + let projection = gzip_decompress_test_projection(&mut state); + let mut frame_size = 0; + let mut zlib_result = 0; + + assert_eq!( + unsafe { FIO_rust_decompressGzipFrame(&projection, &mut frame_size, &mut zlib_result) }, + FIO_RUST_GZIP_DECOMPRESS_OK + ); + assert_eq!(frame_size, 2); + assert_eq!(zlib_result, FIO_RUST_GZIP_DECOMPRESS_Z_OK); + assert_eq!(state.io.input_pos, 2); + assert_eq!(state.io.consumed, vec![2, 0]); + assert_eq!(state.io.enqueue_sizes, vec![1, 1]); + assert_eq!(state.io.output_chunks, vec![vec![b'x'], vec![b'y']]); + assert_eq!(state.flushes, vec![FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH; 2]); + assert_eq!(state.end_calls, 1); + assert_eq!(state.io.acquire_calls, 1); + assert_eq!(state.io.release_calls, 1); + assert_eq!(state.io.sparse_end_calls, 1); + } + + #[test] + fn gzip_decompress_projection_maps_eof_to_premature_error() { + let mut state = GzipDecompressState::new(b"abc"); + state.finish_result = FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR; + let projection = gzip_decompress_test_projection(&mut state); + let mut frame_size = 41; + let mut zlib_result = 0; + + assert_eq!( + unsafe { FIO_rust_decompressGzipFrame(&projection, &mut frame_size, &mut zlib_result) }, + FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR + ); + assert_eq!(frame_size, 0); + assert_eq!(zlib_result, FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR); + assert_eq!(state.io.input_pos, 3); + assert_eq!(state.io.consumed, vec![2, 1, 0]); + assert_eq!(state.end_calls, 1); + assert_eq!(state.io.release_calls, 1); + assert_eq!(state.io.sparse_end_calls, 1); + } + + struct LzmaDecompressState { + io: OptionalDecompressIoState, + init_result: c_int, + init_modes: Vec, + actions: Vec, + finish_result: c_int, + code_error: Option, + end_calls: usize, + } + + impl LzmaDecompressState { + fn new(input: &[u8]) -> Self { + Self { + io: OptionalDecompressIoState::new(input), + init_result: FIO_RUST_LZMA_DECOMPRESS_OK_CODE, + init_modes: Vec::new(), + actions: Vec::new(), + finish_result: FIO_RUST_LZMA_DECOMPRESS_STREAM_END, + code_error: None, + end_calls: 0, + } + } + } + + unsafe extern "C" fn lzma_decompress_test_init( + opaque: *mut c_void, + plain_lzma: c_int, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.init_modes.push(plain_lzma); + state.init_result + } + + unsafe extern "C" fn lzma_decompress_test_code( + opaque: *mut c_void, + _input: *const u8, + input_size: usize, + output: *mut u8, + output_size: usize, + action: c_int, + consumed: *mut usize, + produced: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.actions.push(action); + unsafe { + *consumed = 0; + *produced = 0; + } + if action == FIO_RUST_LZMA_DECOMPRESS_RUN { + assert!(input_size != 0); + assert!(output_size != 0); + unsafe { + *consumed = input_size.min(2); + *produced = 1; + *output = b'r'; + } + return state.code_error.unwrap_or(FIO_RUST_LZMA_DECOMPRESS_OK_CODE); + } + + assert_eq!(action, FIO_RUST_LZMA_DECOMPRESS_FINISH); + assert_eq!(input_size, 0); + assert!(output_size != 0); + unsafe { + *produced = 1; + *output = b'f'; + } + state.finish_result + } + + unsafe extern "C" fn lzma_decompress_test_end(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.end_calls += 1; + } + + fn lzma_decompress_test_projection( + state: &mut LzmaDecompressState, + ) -> FIO_rust_lzma_decompress_projection_t { + let codec_opaque = (state as *mut LzmaDecompressState).cast::(); + let io = optional_decompress_io(&mut state.io, 3); + FIO_rust_lzma_decompress_projection_t { + io, + lzma_opaque: codec_opaque, + lzma_init: Some(lzma_decompress_test_init), + lzma_code: Some(lzma_decompress_test_code), + lzma_end: Some(lzma_decompress_test_end), + } + } + + #[test] + fn lzma_decompress_projection_preserves_mode_and_short_read_accounting() { + let mut state = LzmaDecompressState::new(b"abcde"); + let projection = lzma_decompress_test_projection(&mut state); + let mut frame_size = 0; + let mut lzma_result = 0; + + assert_eq!( + unsafe { + FIO_rust_decompressLzmaFrame(&projection, 1, &mut frame_size, &mut lzma_result) + }, + FIO_RUST_LZMA_DECOMPRESS_OK + ); + assert_eq!(frame_size, 4); + assert_eq!(lzma_result, FIO_RUST_LZMA_DECOMPRESS_OK_CODE); + assert_eq!(state.init_modes, vec![1]); + assert_eq!( + state.actions, + vec![ + FIO_RUST_LZMA_DECOMPRESS_RUN, + FIO_RUST_LZMA_DECOMPRESS_RUN, + FIO_RUST_LZMA_DECOMPRESS_RUN, + FIO_RUST_LZMA_DECOMPRESS_FINISH, + ] + ); + assert_eq!(state.io.input_pos, 5); + assert_eq!(state.io.consumed, vec![2, 2, 1, 0]); + assert_eq!(state.io.enqueue_sizes, vec![1, 1, 1, 1]); + assert_eq!(state.end_calls, 1); + assert_eq!(state.io.release_calls, 1); + assert_eq!(state.io.sparse_end_calls, 1); + } + + #[test] + fn lzma_decompress_projection_propagates_code_error_after_consumption() { + let mut state = LzmaDecompressState::new(b"abc"); + state.code_error = Some(17); + let projection = lzma_decompress_test_projection(&mut state); + let mut frame_size = 41; + let mut lzma_result = 0; + + assert_eq!( + unsafe { + FIO_rust_decompressLzmaFrame(&projection, 0, &mut frame_size, &mut lzma_result) + }, + FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR + ); + assert_eq!(frame_size, 0); + assert_eq!(lzma_result, 17); + assert_eq!(state.io.consumed, vec![2]); + assert_eq!(state.end_calls, 1); + assert_eq!(state.io.release_calls, 1); + assert_eq!(state.io.sparse_end_calls, 1); + } + + struct Lz4DecompressState { + io: OptionalDecompressIoState, + create_status: c_int, + create_result: usize, + code_calls: usize, + code_error: Option, + next_after_code: usize, + unfinished: bool, + free_calls: usize, + progress: Vec, + } + + impl Lz4DecompressState { + fn new(input: &[u8]) -> Self { + Self { + io: OptionalDecompressIoState::new(input), + create_status: 0, + create_result: 0, + code_calls: 0, + code_error: None, + next_after_code: 0, + unfinished: false, + free_calls: 0, + progress: Vec::new(), + } + } + } + + unsafe extern "C" fn lz4_decompress_test_create( + opaque: *mut c_void, + _version: c_uint, + result: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + unsafe { *result = state.create_result }; + state.create_status + } + + unsafe extern "C" fn lz4_decompress_test_code( + opaque: *mut c_void, + output: *mut u8, + output_size: *mut usize, + _input: *const u8, + input_size: *mut usize, + next_to_load: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.code_calls += 1; + if let Some(error) = state.code_error { + unsafe { + *input_size = 0; + *next_to_load = error; + } + return error as c_int; + } + + let consumed = if state.code_calls == 1 { + (*input_size).min(2) + } else { + *input_size + }; + let produced = if state.unfinished && state.code_calls > 1 { + 1 + } else { + *output_size + }; + assert!(produced != 0); + unsafe { + *input_size = consumed; + std::ptr::write_bytes(output, b'u', produced); + *next_to_load = if state.code_calls == 1 || state.unfinished { + state.next_after_code + } else { + 0 + }; + } + 0 + } + + unsafe extern "C" fn lz4_decompress_test_free(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.free_calls += 1; + } + + unsafe extern "C" fn lz4_decompress_test_progress(opaque: *mut c_void, decoded: u64) { + let state = unsafe { &mut *opaque.cast::() }; + state.progress.push(decoded); + } + + fn lz4_decompress_test_projection( + state: &mut Lz4DecompressState, + ) -> FIO_rust_lz4_decompress_projection_t { + let io_opaque = (&mut state.io as *mut OptionalDecompressIoState).cast::(); + let codec_opaque = (state as *mut Lz4DecompressState).cast::(); + FIO_rust_lz4_decompress_projection_t { + io: FIO_rust_decompress_io_projection_t { + read_opaque: io_opaque, + write_opaque: io_opaque, + read_buffer_size: 0, + read_fill: Some(optional_decompress_read_fill), + read_consume: Some(optional_decompress_read_consume), + write_acquire: Some(optional_decompress_write_acquire), + write_enqueue: Some(optional_decompress_write_enqueue), + write_release: Some(optional_decompress_write_release), + sparse_write_end: Some(optional_decompress_sparse_end), + }, + codec_opaque, + progress_opaque: codec_opaque, + version: 100, + create: Some(lz4_decompress_test_create), + code: Some(lz4_decompress_test_code), + free_context: Some(lz4_decompress_test_free), + progress: Some(lz4_decompress_test_progress), + } + } + + #[test] + fn lz4_decompress_projection_retries_full_output_and_accounts_input() { + let mut state = Lz4DecompressState::new(b"abcdTAIL"); + state.next_after_code = 2; + let projection = lz4_decompress_test_projection(&mut state); + let mut frame_size = 0; + let mut lz4_result = 0; + + assert_eq!( + unsafe { FIO_rust_decompressLz4Frame(&projection, &mut frame_size, &mut lz4_result) }, + FIO_RUST_LZ4_DECOMPRESS_OK + ); + assert_eq!(frame_size, 8); + assert_eq!(lz4_result, 0); + assert_eq!(state.code_calls, 2); + assert_eq!(state.io.input_pos, 4); + assert_eq!(state.io.consumed, vec![4]); + assert_eq!(state.io.enqueue_sizes, vec![4, 4]); + assert_eq!(state.progress, vec![4, 8]); + assert_eq!(state.free_calls, 1); + assert_eq!(state.io.release_calls, 1); + assert_eq!(state.io.sparse_end_calls, 1); + } + + #[test] + fn lz4_decompress_projection_reports_errors_and_unfinished_input() { + let mut error_state = Lz4DecompressState::new(b"abcd"); + error_state.code_error = Some(23); + let error_projection = lz4_decompress_test_projection(&mut error_state); + let mut frame_size = 41; + let mut lz4_result = 0; + assert_eq!( + unsafe { + FIO_rust_decompressLz4Frame(&error_projection, &mut frame_size, &mut lz4_result) + }, + FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR + ); + assert_eq!(frame_size, 0); + assert_eq!(lz4_result, 23); + assert_eq!(error_state.io.input_pos, 0); + assert_eq!(error_state.free_calls, 1); + assert_eq!(error_state.io.release_calls, 1); + assert_eq!(error_state.io.sparse_end_calls, 1); + + let mut unfinished_state = Lz4DecompressState::new(b"abcd"); + unfinished_state.next_after_code = 4; + unfinished_state.unfinished = true; + let unfinished_projection = lz4_decompress_test_projection(&mut unfinished_state); + let mut frame_size = 41; + let mut lz4_result = 0; + assert_eq!( + unsafe { + FIO_rust_decompressLz4Frame( + &unfinished_projection, + &mut frame_size, + &mut lz4_result, + ) + }, + FIO_RUST_LZ4_DECOMPRESS_UNFINISHED + ); + assert_eq!(frame_size, 0); + assert_eq!(unfinished_state.io.input_pos, 4); + assert_eq!(unfinished_state.free_calls, 1); + assert_eq!(unfinished_state.io.release_calls, 1); + assert_eq!(unfinished_state.io.sparse_end_calls, 1); + } + #[test] fn classifies_all_cli_decompression_headers() { let is_mock_zstd = |buffer: &[u8]| buffer.starts_with(&[0x28, 0xB5, 0x2F, 0xFD]);