diff --git a/programs/fileio.c b/programs/fileio.c index b365a3d6d..df8768252 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1063,6 +1063,126 @@ int FIO_rust_compressGzipFrame( const char* srcFileName, U64 srcFileSize, int compressionLevel, U64* readsize, U64* compressedSize, int* zlibResult); +enum { + FIO_RUST_LZMA_OK = 0, + FIO_RUST_LZMA_INIT_PRESET_ERROR = 1, + FIO_RUST_LZMA_INIT_ALONE_ERROR = 2, + FIO_RUST_LZMA_INIT_XZ_ERROR = 3, + FIO_RUST_LZMA_CODE_ERROR = 4, + FIO_RUST_LZMA_INVALID_PROJECTION = 5 +}; + +typedef void (*FIO_rust_lzma_read_fill_fn)( + void* opaque, size_t requested, const unsigned char** buffer, size_t* loaded); +typedef void (*FIO_rust_lzma_read_consume_fn)(void* opaque, size_t consumed); +typedef void (*FIO_rust_lzma_write_acquire_fn)( + void* opaque, void** job, unsigned char** buffer, size_t* bufferSize); +typedef void (*FIO_rust_lzma_write_enqueue_fn)( + void* opaque, void** job, size_t usedBufferSize, + unsigned char** buffer, size_t* bufferSize); +typedef void (*FIO_rust_lzma_write_release_fn)(void* opaque, void* job); +typedef void (*FIO_rust_lzma_sparse_write_end_fn)(void* opaque); +typedef int (*FIO_rust_lzma_init_fn)( + void* opaque, int compressionLevel, int plainLzma, int* lzmaResult); +typedef int (*FIO_rust_lzma_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_end_fn)(void* opaque); +typedef void (*FIO_rust_lzma_progress_fn)( + void* opaque, U64 srcFileSize, U64 inFileSize, U64 outFileSize); + +typedef struct { + void* readOpaque; + void* writeOpaque; + void* lzmaOpaque; + void* progressOpaque; + size_t readBufferSize; + FIO_rust_lzma_read_fill_fn readFill; + FIO_rust_lzma_read_consume_fn readConsume; + FIO_rust_lzma_write_acquire_fn writeAcquire; + FIO_rust_lzma_write_enqueue_fn writeEnqueue; + FIO_rust_lzma_write_release_fn writeRelease; + FIO_rust_lzma_sparse_write_end_fn sparseWriteEnd; + FIO_rust_lzma_init_fn lzmaInit; + FIO_rust_lzma_code_fn lzmaCode; + FIO_rust_lzma_end_fn lzmaEnd; + FIO_rust_lzma_progress_fn progress; +} FIO_rust_lzma_compress_projection_t; + +int FIO_rust_compressLzmaFrame( + const FIO_rust_lzma_compress_projection_t* projection, + const char* srcFileName, U64 srcFileSize, int compressionLevel, + int plainLzma, U64* readsize, U64* compressedSize, int* lzmaResult); + +#ifdef ZSTD_LZ4COMPRESS +enum { + FIO_RUST_LZ4_OK = 0, + FIO_RUST_LZ4_CREATE_ERROR = 1, + FIO_RUST_LZ4_HEADER_ERROR = 2, + FIO_RUST_LZ4_UPDATE_ERROR = 3, + FIO_RUST_LZ4_END_ERROR = 4, + FIO_RUST_LZ4_INVALID_PROJECTION = 5 +}; + +typedef int (*FIO_rust_lz4_create_fn)(void* opaque, unsigned version, + size_t* result); +typedef void (*FIO_rust_lz4_prepare_fn)(void* opaque, U64 srcFileSize, + int compressionLevel, int checksumFlag, + size_t blockSize, size_t bufferSize); +typedef int (*FIO_rust_lz4_begin_fn)(void* opaque, unsigned char* output, + size_t outputSize, size_t* result); +typedef int (*FIO_rust_lz4_update_fn)(void* opaque, unsigned char* output, + size_t outputSize, const unsigned char* input, + size_t inputSize, size_t* result); +typedef int (*FIO_rust_lz4_end_fn)(void* opaque, unsigned char* output, + size_t outputSize, size_t* result); +typedef void (*FIO_rust_lz4_free_fn)(void* opaque); +typedef size_t (*FIO_rust_lz4_read_fill_fn)(void* opaque, size_t requested, + const unsigned char** buffer, + size_t* loaded); +typedef void (*FIO_rust_lz4_read_consume_fn)(void* opaque, size_t consumed); +typedef void (*FIO_rust_lz4_write_acquire_fn)(void* opaque, void** job, + unsigned char** buffer, + size_t* bufferSize); +typedef void (*FIO_rust_lz4_write_enqueue_fn)(void* opaque, void** job, + size_t usedBufferSize, + unsigned char** buffer, + size_t* bufferSize); +typedef void (*FIO_rust_lz4_write_release_fn)(void* opaque, void* job); +typedef void (*FIO_rust_lz4_sparse_write_end_fn)(void* opaque); +typedef void (*FIO_rust_lz4_progress_fn)(void* opaque, U64 srcFileSize, + U64 inFileSize, U64 outFileSize); + +typedef struct { + void* readOpaque; + void* writeOpaque; + void* codecOpaque; + void* progressOpaque; + unsigned version; + size_t blockSize; + FIO_rust_lz4_create_fn create; + FIO_rust_lz4_prepare_fn prepare; + FIO_rust_lz4_begin_fn begin; + FIO_rust_lz4_update_fn update; + FIO_rust_lz4_end_fn end; + FIO_rust_lz4_free_fn freeContext; + FIO_rust_lz4_read_fill_fn readFill; + FIO_rust_lz4_read_consume_fn readConsume; + FIO_rust_lz4_write_acquire_fn writeAcquire; + FIO_rust_lz4_write_enqueue_fn writeEnqueue; + FIO_rust_lz4_write_release_fn writeRelease; + FIO_rust_lz4_sparse_write_end_fn sparseWriteEnd; + FIO_rust_lz4_progress_fn progress; +} FIO_rust_lz4_compress_projection_t; + +int FIO_rust_compressLz4Frame( + const FIO_rust_lz4_compress_projection_t* projection, + const char* srcFileName, U64 srcFileSize, int compressionLevel, + int checksumFlag, U64* readsize, U64* compressedSize, + size_t* lz4Result); +#endif + int FIO_rust_compressFilenameInternal( void* fCtx, FIO_prefs_t* prefs, void* ress, const char* dstFileName, const char* srcFileName, @@ -1314,83 +1434,165 @@ static void FIO_rust_gzip_progress(void* opaque, U64 srcFileSize, #ifdef ZSTD_LZMACOMPRESS +static void FIO_rust_lzma_readFill(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 = readCtx->srcBuffer; + *loaded = readCtx->srcBufferLoaded; +} + +static void FIO_rust_lzma_readConsume(void* opaque, size_t consumed) +{ + AIO_ReadPool_consumeBytes((ReadPoolCtx_t*)opaque, consumed); +} + +static void FIO_rust_lzma_writeAcquire(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_lzma_writeEnqueue(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_lzma_writeRelease(void* opaque, void* job) +{ + AIO_WritePool_releaseIoJob((IOJob_t*)job); + (void)opaque; +} + +static void FIO_rust_lzma_sparseWriteEnd(void* opaque) +{ + AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque); +} + +static int FIO_rust_lzma_init(void* opaque, int compressionLevel, + int plainLzma, int* lzmaResult) +{ + lzma_stream* const strm = (lzma_stream*)opaque; + lzma_ret ret; + + *lzmaResult = (int)LZMA_OK; + if (plainLzma) { + lzma_options_lzma opt_lzma; + if (lzma_lzma_preset(&opt_lzma, (uint32_t)compressionLevel)) + return FIO_RUST_LZMA_INIT_PRESET_ERROR; + ret = lzma_alone_encoder(strm, &opt_lzma); /* LZMA */ + if (ret != LZMA_OK) { + *lzmaResult = (int)ret; + return FIO_RUST_LZMA_INIT_ALONE_ERROR; + } + } else { + ret = lzma_easy_encoder(strm, (uint32_t)compressionLevel, + LZMA_CHECK_CRC64); /* XZ */ + if (ret != LZMA_OK) { + *lzmaResult = (int)ret; + return FIO_RUST_LZMA_INIT_XZ_ERROR; + } + } + return FIO_RUST_LZMA_OK; +} + +static int FIO_rust_lzma_code(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 availInBefore = inputSize; + size_t const availOutBefore = 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 = availInBefore - strm->avail_in; + *produced = availOutBefore - strm->avail_out; + return (int)ret; +} + +static void FIO_rust_lzma_end(void* opaque) +{ + lzma_end((lzma_stream*)opaque); +} + +static void FIO_rust_lzma_progress(void* opaque, U64 srcFileSize, + U64 inFileSize, U64 outFileSize) +{ + (void)opaque; + if (srcFileSize == UTIL_FILESIZE_UNKNOWN) + DISPLAYUPDATE_PROGRESS("\rRead : %u MB ==> %.2f%%", + (unsigned)(inFileSize>>20), + (double)outFileSize/(double)inFileSize*100) + else + DISPLAYUPDATE_PROGRESS("\rRead : %u / %u MB ==> %.2f%%", + (unsigned)(inFileSize>>20), (unsigned)(srcFileSize>>20), + (double)outFileSize/(double)inFileSize*100); +} + static unsigned long long FIO_compressLzmaFrame(cRess_t* ress, const char* srcFileName, U64 const srcFileSize, int compressionLevel, U64* readsize, int plain_lzma) { - unsigned long long inFileSize = 0, outFileSize = 0; lzma_stream strm = LZMA_STREAM_INIT; - lzma_action action = LZMA_RUN; - lzma_ret ret; - IOJob_t *writeJob = NULL; + FIO_rust_lzma_compress_projection_t projection; + U64 compressedSize = 0; + int lzmaResult = (int)LZMA_OK; + int status; - if (compressionLevel < 0) compressionLevel = 0; - if (compressionLevel > 9) compressionLevel = 9; + memset(&projection, 0, sizeof(projection)); + projection.readOpaque = (void*)ress->readCtx; + projection.writeOpaque = (void*)ress->writeCtx; + projection.lzmaOpaque = &strm; + projection.progressOpaque = NULL; + projection.readBufferSize = ZSTD_CStreamInSize(); + projection.readFill = FIO_rust_lzma_readFill; + projection.readConsume = FIO_rust_lzma_readConsume; + projection.writeAcquire = FIO_rust_lzma_writeAcquire; + projection.writeEnqueue = FIO_rust_lzma_writeEnqueue; + projection.writeRelease = FIO_rust_lzma_writeRelease; + projection.sparseWriteEnd = FIO_rust_lzma_sparseWriteEnd; + projection.lzmaInit = FIO_rust_lzma_init; + projection.lzmaCode = FIO_rust_lzma_code; + projection.lzmaEnd = FIO_rust_lzma_end; + projection.progress = FIO_rust_lzma_progress; - if (plain_lzma) { - lzma_options_lzma opt_lzma; - if (lzma_lzma_preset(&opt_lzma, compressionLevel)) + status = FIO_rust_compressLzmaFrame( + &projection, srcFileName, srcFileSize, compressionLevel, plain_lzma, + readsize, &compressedSize, &lzmaResult); + switch (status) { + case FIO_RUST_LZMA_OK: + return compressedSize; + case FIO_RUST_LZMA_INIT_PRESET_ERROR: EXM_THROW(81, "zstd: %s: lzma_lzma_preset error", srcFileName); - ret = lzma_alone_encoder(&strm, &opt_lzma); /* LZMA */ - if (ret != LZMA_OK) - EXM_THROW(82, "zstd: %s: lzma_alone_encoder error %d", srcFileName, ret); - } else { - ret = lzma_easy_encoder(&strm, compressionLevel, LZMA_CHECK_CRC64); /* XZ */ - if (ret != LZMA_OK) - EXM_THROW(83, "zstd: %s: lzma_easy_encoder error %d", srcFileName, ret); + case FIO_RUST_LZMA_INIT_ALONE_ERROR: + EXM_THROW(82, "zstd: %s: lzma_alone_encoder error %d", srcFileName, lzmaResult); + case FIO_RUST_LZMA_INIT_XZ_ERROR: + EXM_THROW(83, "zstd: %s: lzma_easy_encoder error %d", srcFileName, lzmaResult); + case FIO_RUST_LZMA_CODE_ERROR: + EXM_THROW(84, "zstd: %s: lzma_code encoding error %d", srcFileName, lzmaResult); + default: + assert(status == FIO_RUST_LZMA_INVALID_PROJECTION); + EXM_THROW(84, "zstd: %s: lzma_code encoding error %d", srcFileName, lzmaResult); } - - writeJob =AIO_WritePool_acquireJob(ress->writeCtx); - strm.next_out = (BYTE*)writeJob->buffer; - strm.avail_out = writeJob->bufferSize; - strm.next_in = 0; - strm.avail_in = 0; - - while (1) { - if (strm.avail_in == 0) { - size_t const inSize = AIO_ReadPool_fillBuffer(ress->readCtx, ZSTD_CStreamInSize()); - if (ress->readCtx->srcBufferLoaded == 0) action = LZMA_FINISH; - inFileSize += inSize; - strm.next_in = (BYTE const*)ress->readCtx->srcBuffer; - strm.avail_in = ress->readCtx->srcBufferLoaded; - } - - { - size_t const availBefore = strm.avail_in; - ret = lzma_code(&strm, action); - AIO_ReadPool_consumeBytes(ress->readCtx, availBefore - strm.avail_in); - } - - - if (ret != LZMA_OK && ret != LZMA_STREAM_END) - EXM_THROW(84, "zstd: %s: lzma_code encoding error %d", srcFileName, ret); - { size_t const compBytes = writeJob->bufferSize - strm.avail_out; - if (compBytes) { - writeJob->usedBufferSize = compBytes; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - outFileSize += compBytes; - strm.next_out = (BYTE*)writeJob->buffer; - strm.avail_out = writeJob->bufferSize; - } } - if (srcFileSize == UTIL_FILESIZE_UNKNOWN) - DISPLAYUPDATE_PROGRESS("\rRead : %u MB ==> %.2f%%", - (unsigned)(inFileSize>>20), - (double)outFileSize/(double)inFileSize*100) - else - DISPLAYUPDATE_PROGRESS("\rRead : %u / %u MB ==> %.2f%%", - (unsigned)(inFileSize>>20), (unsigned)(srcFileSize>>20), - (double)outFileSize/(double)inFileSize*100); - if (ret == LZMA_STREAM_END) break; - } - - lzma_end(&strm); - *readsize = inFileSize; - - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - - return outFileSize; } #endif @@ -1406,96 +1608,195 @@ static int FIO_LZ4_GetBlockSize_FromBlockId (int id) return FIO_rust_LZ4_GetBlockSize_FromBlockId(id); } +typedef struct { + LZ4F_compressionContext_t ctx; + LZ4F_preferences_t prefs; +} FIO_rust_lz4_context_t; + +static int FIO_rust_lz4_create(void* opaque, unsigned version, size_t* result) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + LZ4F_errorCode_t const errorCode = LZ4F_createCompressionContext(&context->ctx, version); + *result = (size_t)errorCode; + return LZ4F_isError(errorCode); +} + +static void FIO_rust_lz4_prepare(void* opaque, U64 srcFileSize, + int compressionLevel, int checksumFlag, + size_t blockSize, size_t bufferSize) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + memset(&context->prefs, 0, sizeof(context->prefs)); + + /* autoflush off to mitigate a bug in lz4<=1.9.3 for compression level 12 */ + context->prefs.autoFlush = 0; + context->prefs.compressionLevel = compressionLevel; + context->prefs.frameInfo.blockMode = LZ4F_blockLinked; + context->prefs.frameInfo.blockSizeID = LZ4F_max64KB; + context->prefs.frameInfo.contentChecksumFlag = (contentChecksum_t)checksumFlag; +#if LZ4_VERSION_NUMBER >= 10600 + context->prefs.frameInfo.contentSize = (srcFileSize==UTIL_FILESIZE_UNKNOWN) ? 0 : srcFileSize; +#else + (void)srcFileSize; +#endif + assert(LZ4F_compressBound(blockSize, &context->prefs) <= bufferSize); +} + +static int FIO_rust_lz4_begin(void* opaque, unsigned char* output, + size_t outputSize, size_t* result) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + size_t const headerSize = LZ4F_compressBegin(context->ctx, output, outputSize, + &context->prefs); + *result = headerSize; + return LZ4F_isError(headerSize); +} + +static int FIO_rust_lz4_update(void* opaque, unsigned char* output, + size_t outputSize, const unsigned char* input, + size_t inputSize, size_t* result) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + size_t const outSize = LZ4F_compressUpdate(context->ctx, output, outputSize, + input, inputSize, NULL); + *result = outSize; + return LZ4F_isError(outSize); +} + +static int FIO_rust_lz4_end(void* opaque, unsigned char* output, + size_t outputSize, size_t* result) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + size_t const endSize = LZ4F_compressEnd(context->ctx, output, outputSize, NULL); + *result = endSize; + return LZ4F_isError(endSize); +} + +static void FIO_rust_lz4_free(void* opaque) +{ + FIO_rust_lz4_context_t* const context = (FIO_rust_lz4_context_t*)opaque; + LZ4F_freeCompressionContext(context->ctx); + context->ctx = NULL; +} + +static size_t FIO_rust_lz4_readFill(void* opaque, size_t requested, + const unsigned char** buffer, size_t* loaded) +{ + ReadPoolCtx_t* const readCtx = (ReadPoolCtx_t*)opaque; + size_t const added = AIO_ReadPool_fillBuffer(readCtx, requested); + *buffer = readCtx->srcBuffer; + *loaded = readCtx->srcBufferLoaded; + return added; +} + +static void FIO_rust_lz4_readConsume(void* opaque, size_t consumed) +{ + AIO_ReadPool_consumeBytes((ReadPoolCtx_t*)opaque, consumed); +} + +static void FIO_rust_lz4_writeAcquire(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_lz4_writeEnqueue(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_lz4_writeRelease(void* opaque, void* job) +{ + AIO_WritePool_releaseIoJob((IOJob_t*)job); + (void)opaque; +} + +static void FIO_rust_lz4_sparseWriteEnd(void* opaque) +{ + AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque); +} + +static void FIO_rust_lz4_progress(void* opaque, U64 srcFileSize, + U64 inFileSize, U64 outFileSize) +{ + (void)opaque; + if (srcFileSize == UTIL_FILESIZE_UNKNOWN) { + DISPLAYUPDATE_PROGRESS("\rRead : %u MB ==> %.2f%%", + (unsigned)(inFileSize>>20), + (double)outFileSize/(double)inFileSize*100) + } else { + DISPLAYUPDATE_PROGRESS("\rRead : %u / %u MB ==> %.2f%%", + (unsigned)(inFileSize>>20), (unsigned)(srcFileSize>>20), + (double)outFileSize/(double)inFileSize*100); + } +} + static unsigned long long FIO_compressLz4Frame(cRess_t* ress, const char* srcFileName, U64 const srcFileSize, int compressionLevel, int checksumFlag, U64* readsize) { - const size_t blockSize = FIO_LZ4_GetBlockSize_FromBlockId(LZ4F_max64KB); - unsigned long long inFileSize = 0, outFileSize = 0; + FIO_rust_lz4_context_t codec; + FIO_rust_lz4_compress_projection_t projection; + U64 compressedSize = 0; + size_t lz4Result = 0; + int status; - LZ4F_preferences_t prefs; - LZ4F_compressionContext_t ctx; + memset(&codec, 0, sizeof(codec)); + memset(&projection, 0, sizeof(projection)); + projection.readOpaque = ress->readCtx; + projection.writeOpaque = ress->writeCtx; + projection.codecOpaque = &codec; + projection.progressOpaque = NULL; + projection.version = LZ4F_VERSION; + projection.blockSize = (size_t)FIO_LZ4_GetBlockSize_FromBlockId(LZ4F_max64KB); + projection.create = FIO_rust_lz4_create; + projection.prepare = FIO_rust_lz4_prepare; + projection.begin = FIO_rust_lz4_begin; + projection.update = FIO_rust_lz4_update; + projection.end = FIO_rust_lz4_end; + projection.freeContext = FIO_rust_lz4_free; + projection.readFill = FIO_rust_lz4_readFill; + projection.readConsume = FIO_rust_lz4_readConsume; + projection.writeAcquire = FIO_rust_lz4_writeAcquire; + projection.writeEnqueue = FIO_rust_lz4_writeEnqueue; + projection.writeRelease = FIO_rust_lz4_writeRelease; + projection.sparseWriteEnd = FIO_rust_lz4_sparseWriteEnd; + projection.progress = FIO_rust_lz4_progress; - IOJob_t* writeJob = AIO_WritePool_acquireJob(ress->writeCtx); - - LZ4F_errorCode_t const errorCode = LZ4F_createCompressionContext(&ctx, LZ4F_VERSION); - if (LZ4F_isError(errorCode)) - EXM_THROW(31, "zstd: failed to create lz4 compression context"); - - memset(&prefs, 0, sizeof(prefs)); - - assert(blockSize <= ress->readCtx->base.jobBufferSize); - - /* autoflush off to mitigate a bug in lz4<=1.9.3 for compression level 12 */ - prefs.autoFlush = 0; - prefs.compressionLevel = compressionLevel; - prefs.frameInfo.blockMode = LZ4F_blockLinked; - prefs.frameInfo.blockSizeID = LZ4F_max64KB; - prefs.frameInfo.contentChecksumFlag = (contentChecksum_t)checksumFlag; -#if LZ4_VERSION_NUMBER >= 10600 - prefs.frameInfo.contentSize = (srcFileSize==UTIL_FILESIZE_UNKNOWN) ? 0 : srcFileSize; -#endif - assert(LZ4F_compressBound(blockSize, &prefs) <= writeJob->bufferSize); - - { - size_t headerSize = LZ4F_compressBegin(ctx, writeJob->buffer, writeJob->bufferSize, &prefs); - if (LZ4F_isError(headerSize)) + status = FIO_rust_compressLz4Frame( + &projection, srcFileName, srcFileSize, compressionLevel, checksumFlag, + readsize, &compressedSize, &lz4Result); + switch (status) { + case FIO_RUST_LZ4_OK: + return compressedSize; + case FIO_RUST_LZ4_CREATE_ERROR: + EXM_THROW(31, "zstd: failed to create lz4 compression context"); + case FIO_RUST_LZ4_HEADER_ERROR: EXM_THROW(33, "File header generation failed : %s", - LZ4F_getErrorName(headerSize)); - writeJob->usedBufferSize = headerSize; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - outFileSize += headerSize; - - /* Read first block */ - inFileSize += AIO_ReadPool_fillBuffer(ress->readCtx, blockSize); - - /* Main Loop */ - while (ress->readCtx->srcBufferLoaded) { - size_t inSize = MIN(blockSize, ress->readCtx->srcBufferLoaded); - size_t const outSize = LZ4F_compressUpdate(ctx, writeJob->buffer, writeJob->bufferSize, - ress->readCtx->srcBuffer, inSize, NULL); - if (LZ4F_isError(outSize)) - EXM_THROW(35, "zstd: %s: lz4 compression failed : %s", - srcFileName, LZ4F_getErrorName(outSize)); - outFileSize += outSize; - if (srcFileSize == UTIL_FILESIZE_UNKNOWN) { - DISPLAYUPDATE_PROGRESS("\rRead : %u MB ==> %.2f%%", - (unsigned)(inFileSize>>20), - (double)outFileSize/(double)inFileSize*100) - } else { - DISPLAYUPDATE_PROGRESS("\rRead : %u / %u MB ==> %.2f%%", - (unsigned)(inFileSize>>20), (unsigned)(srcFileSize>>20), - (double)outFileSize/(double)inFileSize*100); - } - - /* Write Block */ - writeJob->usedBufferSize = outSize; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - - /* Read next block */ - AIO_ReadPool_consumeBytes(ress->readCtx, inSize); - inFileSize += AIO_ReadPool_fillBuffer(ress->readCtx, blockSize); - } - - /* End of Stream mark */ - headerSize = LZ4F_compressEnd(ctx, writeJob->buffer, writeJob->bufferSize, NULL); - if (LZ4F_isError(headerSize)) + LZ4F_getErrorName(lz4Result)); + case FIO_RUST_LZ4_UPDATE_ERROR: + EXM_THROW(35, "zstd: %s: lz4 compression failed : %s", + srcFileName, LZ4F_getErrorName(lz4Result)); + case FIO_RUST_LZ4_END_ERROR: EXM_THROW(38, "zstd: %s: lz4 end of file generation failed : %s", - srcFileName, LZ4F_getErrorName(headerSize)); - - writeJob->usedBufferSize = headerSize; - AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob); - outFileSize += headerSize; + srcFileName, LZ4F_getErrorName(lz4Result)); + default: + assert(status == FIO_RUST_LZ4_INVALID_PROJECTION); + EXM_THROW(31, "zstd: failed to create lz4 compression context"); } - - *readsize = inFileSize; - LZ4F_freeCompressionContext(ctx); - AIO_WritePool_releaseIoJob(writeJob); - AIO_WritePool_sparseWriteEnd(ress->writeCtx); - - return outFileSize; } #endif diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 7c0bcb208..8c4b5298e 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -180,6 +180,118 @@ pub struct FIO_rust_gzip_compress_projection_t { pub progress: Option, } +pub const FIO_RUST_LZMA_OK: c_int = 0; +pub const FIO_RUST_LZMA_INIT_PRESET_ERROR: c_int = 1; +pub const FIO_RUST_LZMA_INIT_ALONE_ERROR: c_int = 2; +pub const FIO_RUST_LZMA_INIT_XZ_ERROR: c_int = 3; +pub const FIO_RUST_LZMA_CODE_ERROR: c_int = 4; +pub const FIO_RUST_LZMA_INVALID_PROJECTION: c_int = 5; + +const FIO_RUST_LZMA_OK_CODE: c_int = 0; +const FIO_RUST_LZMA_STREAM_END: c_int = 1; +const FIO_RUST_LZMA_RUN: c_int = 0; +const FIO_RUST_LZMA_FINISH: c_int = 3; + +pub type FIO_rust_lzma_read_fill_fn = + unsafe extern "C" fn(*mut c_void, usize, *mut *const u8, *mut usize); +pub type FIO_rust_lzma_read_consume_fn = unsafe extern "C" fn(*mut c_void, usize); +pub type FIO_rust_lzma_write_acquire_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, *mut *mut u8, *mut usize); +pub type FIO_rust_lzma_write_enqueue_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, usize, *mut *mut u8, *mut usize); +pub type FIO_rust_lzma_write_release_fn = unsafe extern "C" fn(*mut c_void, *mut c_void); +pub type FIO_rust_lzma_sparse_write_end_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_lzma_init_fn = + unsafe extern "C" fn(*mut c_void, c_int, c_int, *mut c_int) -> c_int; +pub type FIO_rust_lzma_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_end_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_lzma_progress_fn = unsafe extern "C" fn(*mut c_void, u64, u64, u64); + +/// Rust owns the LZMA/xz read, code, finish, and accounting loop. C keeps +/// the liblzma stream, asynchronous pools, sparse-output behavior, and +/// diagnostics behind opaque callbacks. +#[repr(C)] +pub struct FIO_rust_lzma_compress_projection_t { + pub read_opaque: *mut c_void, + pub write_opaque: *mut c_void, + pub lzma_opaque: *mut c_void, + pub progress_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 lzma_init: Option, + pub lzma_code: Option, + pub lzma_end: Option, + pub progress: Option, +} + +pub const FIO_RUST_LZ4_OK: c_int = 0; +pub const FIO_RUST_LZ4_CREATE_ERROR: c_int = 1; +pub const FIO_RUST_LZ4_HEADER_ERROR: c_int = 2; +pub const FIO_RUST_LZ4_UPDATE_ERROR: c_int = 3; +pub const FIO_RUST_LZ4_END_ERROR: c_int = 4; +pub const FIO_RUST_LZ4_INVALID_PROJECTION: c_int = 5; + +pub type FIO_rust_lz4_create_fn = unsafe extern "C" fn(*mut c_void, c_uint, *mut usize) -> c_int; +pub type FIO_rust_lz4_prepare_fn = + unsafe extern "C" fn(*mut c_void, u64, c_int, c_int, usize, usize); +pub type FIO_rust_lz4_begin_fn = + unsafe extern "C" fn(*mut c_void, *mut u8, usize, *mut usize) -> c_int; +pub type FIO_rust_lz4_update_fn = + unsafe extern "C" fn(*mut c_void, *mut u8, usize, *const u8, usize, *mut usize) -> c_int; +pub type FIO_rust_lz4_end_fn = + unsafe extern "C" fn(*mut c_void, *mut u8, usize, *mut usize) -> c_int; +pub type FIO_rust_lz4_free_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_lz4_read_fill_fn = + unsafe extern "C" fn(*mut c_void, usize, *mut *const u8, *mut usize) -> usize; +pub type FIO_rust_lz4_read_consume_fn = unsafe extern "C" fn(*mut c_void, usize); +pub type FIO_rust_lz4_write_acquire_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, *mut *mut u8, *mut usize); +pub type FIO_rust_lz4_write_enqueue_fn = + unsafe extern "C" fn(*mut c_void, *mut *mut c_void, usize, *mut *mut u8, *mut usize); +pub type FIO_rust_lz4_write_release_fn = unsafe extern "C" fn(*mut c_void, *mut c_void); +pub type FIO_rust_lz4_sparse_write_end_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_lz4_progress_fn = unsafe extern "C" fn(*mut c_void, u64, u64, u64); + +/// Rust owns the LZ4 frame loop. C keeps the LZ4F context/preferences and +/// asynchronous pool objects private, exposing only the operations required +/// to preserve the original header, block, progress, and cleanup ordering. +#[repr(C)] +pub struct FIO_rust_lz4_compress_projection_t { + pub read_opaque: *mut c_void, + pub write_opaque: *mut c_void, + pub codec_opaque: *mut c_void, + pub progress_opaque: *mut c_void, + pub version: c_uint, + pub block_size: usize, + pub create: Option, + pub prepare: Option, + pub begin: Option, + pub update: Option, + pub end: Option, + pub free_context: Option, + 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 progress: Option, +} + /// C's `FIO_prefs_t` from `programs/fileio_types.h`. /// /// `fileio_prefs.rs` contains the same C layout for the preferences API. It @@ -1932,6 +2044,390 @@ pub unsafe extern "C" fn FIO_rust_compressGzipFrame( FIO_RUST_GZIP_OK } +/// Compresses one LZ4 frame through C-owned LZ4F and asynchronous resource +/// callbacks. The callback projection deliberately reports bytes added by a +/// read-pool fill separately from the currently loaded byte count: the former +/// is the accounting used by the original C leaf, while the latter drives the +/// current block update. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_compressLz4Frame( + projection: *const FIO_rust_lz4_compress_projection_t, + _src_file_name: *const c_char, + src_file_size: u64, + compression_level: c_int, + checksum_flag: c_int, + read_size: *mut u64, + compressed_size: *mut u64, + lz4_result: *mut usize, +) -> c_int { + assert!(!projection.is_null()); + assert!(!read_size.is_null()); + assert!(!compressed_size.is_null()); + assert!(!lz4_result.is_null()); + + unsafe { + *read_size = 0; + *compressed_size = 0; + *lz4_result = 0; + } + + let projection = unsafe { &*projection }; + if projection.block_size == 0 { + return FIO_RUST_LZ4_INVALID_PROJECTION; + } + let Some(create) = projection.create else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(prepare) = projection.prepare else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(begin) = projection.begin else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(update) = projection.update else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(end) = projection.end else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(free_context) = projection.free_context else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(read_fill) = projection.read_fill else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(read_consume) = projection.read_consume else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(write_acquire) = projection.write_acquire else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(write_enqueue) = projection.write_enqueue else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(write_release) = projection.write_release else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(sparse_write_end) = projection.sparse_write_end else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + let Some(progress) = projection.progress else { + return FIO_RUST_LZ4_INVALID_PROJECTION; + }; + + let mut job = ptr::null_mut::(); + let mut output = ptr::null_mut::(); + let mut output_size = 0_usize; + unsafe { + write_acquire( + projection.write_opaque, + &mut job, + &mut output, + &mut output_size, + ); + } + if job.is_null() || (output.is_null() && output_size != 0) { + return FIO_RUST_LZ4_INVALID_PROJECTION; + } + + let mut result = 0_usize; + let create_status = unsafe { create(projection.codec_opaque, projection.version, &mut result) }; + if create_status != 0 { + unsafe { *lz4_result = result }; + return FIO_RUST_LZ4_CREATE_ERROR; + } + + unsafe { + prepare( + projection.codec_opaque, + src_file_size, + compression_level, + checksum_flag, + projection.block_size, + output_size, + ); + } + + result = 0; + let begin_status = unsafe { begin(projection.codec_opaque, output, output_size, &mut result) }; + if begin_status != 0 { + unsafe { *lz4_result = result }; + return FIO_RUST_LZ4_HEADER_ERROR; + } + let header_size = result; + unsafe { + write_enqueue( + projection.write_opaque, + &mut job, + header_size, + &mut output, + &mut output_size, + ); + } + let mut in_file_size = 0_u64; + let mut out_file_size = header_size as u64; + + let mut input = ptr::null::(); + let mut loaded = 0_usize; + in_file_size = in_file_size.wrapping_add(unsafe { + read_fill( + projection.read_opaque, + projection.block_size, + &mut input, + &mut loaded, + ) as u64 + }); + + while loaded != 0 { + if input.is_null() { + return FIO_RUST_LZ4_INVALID_PROJECTION; + } + let input_size = projection.block_size.min(loaded); + result = 0; + let update_status = unsafe { + update( + projection.codec_opaque, + output, + output_size, + input, + input_size, + &mut result, + ) + }; + if update_status != 0 { + unsafe { *lz4_result = result }; + return FIO_RUST_LZ4_UPDATE_ERROR; + } + out_file_size = out_file_size.wrapping_add(result as u64); + unsafe { + progress( + projection.progress_opaque, + src_file_size, + in_file_size, + out_file_size, + ); + write_enqueue( + projection.write_opaque, + &mut job, + result, + &mut output, + &mut output_size, + ); + read_consume(projection.read_opaque, input_size); + } + in_file_size = in_file_size.wrapping_add(unsafe { + read_fill( + projection.read_opaque, + projection.block_size, + &mut input, + &mut loaded, + ) as u64 + }); + } + + result = 0; + let end_status = unsafe { end(projection.codec_opaque, output, output_size, &mut result) }; + if end_status != 0 { + unsafe { *lz4_result = result }; + return FIO_RUST_LZ4_END_ERROR; + } + unsafe { + write_enqueue( + projection.write_opaque, + &mut job, + result, + &mut output, + &mut output_size, + ); + } + out_file_size = out_file_size.wrapping_add(result as u64); + + unsafe { + *read_size = in_file_size; + *compressed_size = out_file_size; + free_context(projection.codec_opaque); + write_release(projection.write_opaque, job); + sparse_write_end(projection.write_opaque); + } + FIO_RUST_LZ4_OK +} + +/// Compresses one LZMA or xz member through the C-owned liblzma stream and +/// asynchronous resource callbacks. The action transition and accounting +/// intentionally mirror the original C loop: input is counted when a pool +/// buffer is filled, consumed bytes are returned immediately after each +/// `lzma_code()` call, non-empty output is enqueued before progress is shown, +/// and `LZMA_FINISH` is retried until `LZMA_STREAM_END`. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_compressLzmaFrame( + projection: *const FIO_rust_lzma_compress_projection_t, + _src_file_name: *const c_char, + src_file_size: u64, + compression_level: c_int, + plain_lzma: c_int, + read_size: *mut u64, + compressed_size: *mut u64, + lzma_result: *mut c_int, +) -> c_int { + assert!(!projection.is_null()); + assert!(!read_size.is_null()); + assert!(!compressed_size.is_null()); + assert!(!lzma_result.is_null()); + + unsafe { + *read_size = 0; + *compressed_size = 0; + *lzma_result = FIO_RUST_LZMA_OK_CODE; + } + + let projection = unsafe { &*projection }; + let Some(read_fill) = projection.read_fill else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(read_consume) = projection.read_consume else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(write_acquire) = projection.write_acquire else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(write_enqueue) = projection.write_enqueue else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(write_release) = projection.write_release else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(sparse_write_end) = projection.sparse_write_end else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(lzma_init) = projection.lzma_init else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(lzma_code) = projection.lzma_code else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(lzma_end) = projection.lzma_end else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + let Some(progress) = projection.progress else { + return FIO_RUST_LZMA_INVALID_PROJECTION; + }; + + let compression_level = compression_level.clamp(0, 9); + let init_result = unsafe { + lzma_init( + projection.lzma_opaque, + compression_level, + plain_lzma, + lzma_result, + ) + }; + if init_result != FIO_RUST_LZMA_OK { + return init_result; + } + + let mut job = ptr::null_mut::(); + let mut output = ptr::null_mut::(); + let mut output_size = 0_usize; + unsafe { + write_acquire( + projection.write_opaque, + &mut job, + &mut output, + &mut output_size, + ); + } + assert!(!job.is_null()); + assert!(!output.is_null() || output_size == 0); + + let mut action = FIO_RUST_LZMA_RUN; + let mut input = ptr::null(); + let mut input_size = 0_usize; + let mut in_file_size = 0_u64; + let mut out_file_size = 0_u64; + + loop { + if input_size == 0 { + let mut loaded = 0_usize; + unsafe { + read_fill( + projection.read_opaque, + projection.read_buffer_size, + &mut input, + &mut loaded, + ); + } + if loaded == 0 { + action = FIO_RUST_LZMA_FINISH; + } else { + assert!(!input.is_null()); + input_size = loaded; + in_file_size = in_file_size.wrapping_add(loaded as u64); + } + } + + 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.read_opaque, consumed) }; + if consumed != 0 { + input = unsafe { input.add(consumed) }; + } + input_size -= consumed; + + if result != FIO_RUST_LZMA_OK_CODE && result != FIO_RUST_LZMA_STREAM_END { + unsafe { *lzma_result = result }; + return FIO_RUST_LZMA_CODE_ERROR; + } + + if produced != 0 { + unsafe { + write_enqueue( + projection.write_opaque, + &mut job, + produced, + &mut output, + &mut output_size, + ); + } + out_file_size = out_file_size.wrapping_add(produced as u64); + } + unsafe { + progress( + projection.progress_opaque, + src_file_size, + in_file_size, + out_file_size, + ) + }; + + if result == FIO_RUST_LZMA_STREAM_END { + break; + } + } + + unsafe { lzma_end(projection.lzma_opaque) }; + unsafe { + *read_size = in_file_size; + *compressed_size = out_file_size; + write_release(projection.write_opaque, job); + sparse_write_end(projection.write_opaque); + } + FIO_RUST_LZMA_OK +} + /// Decompresses exactly one zstd frame using the existing asynchronous pools. /// /// C retains the frame dispatcher and all user-facing diagnostics. This ABI @@ -2927,6 +3423,719 @@ mod tests { assert_eq!(state.sparse_end_calls, 0); } + struct LzmaProjectionState { + input: [u8; 5], + input_pos: usize, + output_buffer: [u8; 4], + read_fill_calls: usize, + consumed: Vec, + enqueue_sizes: Vec, + output_chunks: Vec>, + acquire_calls: usize, + release_calls: usize, + sparse_end_calls: usize, + init_args: Vec<(c_int, c_int)>, + init_status: c_int, + init_result: c_int, + actions: Vec, + finish_calls: usize, + code_result: Option, + end_calls: usize, + progress: Vec<(u64, u64, u64)>, + } + + impl Default for LzmaProjectionState { + fn default() -> Self { + Self { + input: *b"abcde", + input_pos: 0, + output_buffer: [0; 4], + read_fill_calls: 0, + consumed: Vec::new(), + enqueue_sizes: Vec::new(), + output_chunks: Vec::new(), + acquire_calls: 0, + release_calls: 0, + sparse_end_calls: 0, + init_args: Vec::new(), + init_status: FIO_RUST_LZMA_OK, + init_result: FIO_RUST_LZMA_OK_CODE, + actions: Vec::new(), + finish_calls: 0, + code_result: None, + end_calls: 0, + progress: Vec::new(), + } + } + } + + unsafe extern "C" fn lzma_test_read_fill( + opaque: *mut c_void, + requested: usize, + buffer: *mut *const u8, + loaded: *mut usize, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.read_fill_calls += 1; + let available = state.input.len() - state.input_pos; + let amount = available.min(requested); + unsafe { + *buffer = state.input.as_ptr().add(state.input_pos); + *loaded = amount; + } + } + + unsafe extern "C" fn lzma_test_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 lzma_test_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_buffer.as_mut_ptr(); + *buffer_size = state.output_buffer.len(); + } + } + + unsafe extern "C" fn lzma_test_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_buffer.len()); + state.enqueue_sizes.push(used); + state + .output_chunks + .push(state.output_buffer[..used].to_vec()); + unsafe { + *buffer = state.output_buffer.as_mut_ptr(); + *buffer_size = state.output_buffer.len(); + } + } + + unsafe extern "C" fn lzma_test_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 lzma_test_sparse_end(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.sparse_end_calls += 1; + } + + unsafe extern "C" fn lzma_test_init( + opaque: *mut c_void, + level: c_int, + plain_lzma: c_int, + result: *mut c_int, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.init_args.push((level, plain_lzma)); + unsafe { *result = state.init_result }; + state.init_status + } + + unsafe extern "C" fn lzma_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_RUN { + assert!(input_size != 0); + assert!(output_size != 0); + unsafe { + *consumed = input_size.min(2); + *produced = 1; + *output = b'r'; + } + return state.code_result.unwrap_or(FIO_RUST_LZMA_OK_CODE); + } + + assert_eq!(action, FIO_RUST_LZMA_FINISH); + assert_eq!(input_size, 0); + assert!(output_size != 0); + state.finish_calls += 1; + if state.finish_calls == 1 { + unsafe { + *produced = 1; + *output = b'f'; + } + return FIO_RUST_LZMA_OK_CODE; + } + assert!(output_size >= 2); + unsafe { + *produced = 2; + *output.add(0) = b'g'; + *output.add(1) = b'h'; + } + FIO_RUST_LZMA_STREAM_END + } + + unsafe extern "C" fn lzma_test_end(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.end_calls += 1; + } + + unsafe extern "C" fn lzma_test_progress( + opaque: *mut c_void, + src_file_size: u64, + in_file_size: u64, + out_file_size: u64, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state + .progress + .push((src_file_size, in_file_size, out_file_size)); + } + + fn lzma_test_projection( + state: &mut LzmaProjectionState, + ) -> FIO_rust_lzma_compress_projection_t { + let opaque = (state as *mut LzmaProjectionState).cast::(); + FIO_rust_lzma_compress_projection_t { + read_opaque: opaque, + write_opaque: opaque, + lzma_opaque: opaque, + progress_opaque: opaque, + read_buffer_size: 3, + read_fill: Some(lzma_test_read_fill), + read_consume: Some(lzma_test_read_consume), + write_acquire: Some(lzma_test_write_acquire), + write_enqueue: Some(lzma_test_write_enqueue), + write_release: Some(lzma_test_write_release), + sparse_write_end: Some(lzma_test_sparse_end), + lzma_init: Some(lzma_test_init), + lzma_code: Some(lzma_test_code), + lzma_end: Some(lzma_test_end), + progress: Some(lzma_test_progress), + } + } + + #[test] + fn lzma_projection_preserves_run_finish_order_and_accounting() { + let mut state = LzmaProjectionState::default(); + let projection = lzma_test_projection(&mut state); + let mut read_size = 0; + let mut compressed_size = 0; + let mut lzma_result = 0; + + assert_eq!( + unsafe { + FIO_rust_compressLzmaFrame( + &projection, + c"lzma-test".as_ptr(), + 5, + 99, + 1, + &mut read_size, + &mut compressed_size, + &mut lzma_result, + ) + }, + FIO_RUST_LZMA_OK + ); + assert_eq!(read_size, 5); + assert_eq!(compressed_size, 6); + assert_eq!(lzma_result, FIO_RUST_LZMA_OK_CODE); + assert_eq!(state.init_args, vec![(9, 1)]); + assert_eq!(state.read_fill_calls, 4); + assert_eq!(state.consumed, vec![2, 1, 2, 0, 0]); + assert_eq!( + state.actions, + vec![ + FIO_RUST_LZMA_RUN, + FIO_RUST_LZMA_RUN, + FIO_RUST_LZMA_RUN, + FIO_RUST_LZMA_FINISH, + FIO_RUST_LZMA_FINISH, + ] + ); + assert_eq!(state.enqueue_sizes, vec![1, 1, 1, 1, 2]); + assert_eq!( + state.output_chunks, + vec![ + vec![b'r'], + vec![b'r'], + vec![b'r'], + vec![b'f'], + vec![b'g', b'h'] + ] + ); + assert_eq!(state.acquire_calls, 1); + assert_eq!(state.release_calls, 1); + assert_eq!(state.sparse_end_calls, 1); + assert_eq!(state.finish_calls, 2); + assert_eq!(state.end_calls, 1); + assert_eq!( + state.progress, + vec![(5, 3, 1), (5, 3, 2), (5, 5, 3), (5, 5, 4), (5, 5, 6)] + ); + } + + #[test] + fn lzma_projection_clamps_level_and_stops_before_pool_on_init_error() { + let mut state = LzmaProjectionState { + init_status: FIO_RUST_LZMA_INIT_XZ_ERROR, + init_result: 11, + ..LzmaProjectionState::default() + }; + let projection = lzma_test_projection(&mut state); + let mut read_size = 41; + let mut compressed_size = 43; + let mut lzma_result = 0; + + assert_eq!( + unsafe { + FIO_rust_compressLzmaFrame( + &projection, + c"lzma-test".as_ptr(), + 5, + -7, + 0, + &mut read_size, + &mut compressed_size, + &mut lzma_result, + ) + }, + FIO_RUST_LZMA_INIT_XZ_ERROR + ); + assert_eq!(lzma_result, 11); + assert_eq!(read_size, 0); + assert_eq!(compressed_size, 0); + assert_eq!(state.init_args, vec![(0, 0)]); + assert_eq!(state.acquire_calls, 0); + assert_eq!(state.end_calls, 0); + assert_eq!(state.sparse_end_calls, 0); + } + + #[test] + fn lzma_projection_propagates_code_error_after_consuming_input() { + let mut state = LzmaProjectionState { + code_result: Some(17), + ..LzmaProjectionState::default() + }; + let projection = lzma_test_projection(&mut state); + let mut read_size = 0; + let mut compressed_size = 0; + let mut lzma_result = 0; + + assert_eq!( + unsafe { + FIO_rust_compressLzmaFrame( + &projection, + c"lzma-test".as_ptr(), + 5, + 3, + 1, + &mut read_size, + &mut compressed_size, + &mut lzma_result, + ) + }, + FIO_RUST_LZMA_CODE_ERROR + ); + assert_eq!(lzma_result, 17); + assert_eq!(read_size, 0); + assert_eq!(compressed_size, 0); + assert_eq!(state.consumed, vec![2]); + assert_eq!(state.enqueue_sizes, Vec::::new()); + assert_eq!(state.release_calls, 0); + assert_eq!(state.end_calls, 0); + assert_eq!(state.sparse_end_calls, 0); + } + + struct Lz4ProjectionState { + input: [u8; 7], + input_len: usize, + input_pos: usize, + output_buffer: [u8; 8], + events: Vec<&'static str>, + consumed: Vec, + enqueue_sizes: Vec, + progress: Vec<(u64, u64, u64)>, + prepare_args: Vec<(u64, c_int, c_int, usize, usize)>, + acquire_calls: usize, + release_calls: usize, + sparse_end_calls: usize, + create_status: c_int, + create_result: usize, + begin_status: c_int, + begin_result: usize, + update_status: c_int, + update_result: usize, + end_status: c_int, + end_result: usize, + } + + impl Default for Lz4ProjectionState { + fn default() -> Self { + Self { + input: *b"abcdefg", + input_len: 7, + input_pos: 0, + output_buffer: [0; 8], + events: Vec::new(), + consumed: Vec::new(), + enqueue_sizes: Vec::new(), + progress: Vec::new(), + prepare_args: Vec::new(), + acquire_calls: 0, + release_calls: 0, + sparse_end_calls: 0, + create_status: 0, + create_result: 0, + begin_status: 0, + begin_result: 1, + update_status: 0, + update_result: 2, + end_status: 0, + end_result: 3, + } + } + } + + unsafe extern "C" fn lz4_test_create( + opaque: *mut c_void, + _version: c_uint, + result: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("create"); + unsafe { *result = state.create_result }; + state.create_status + } + + unsafe extern "C" fn lz4_test_prepare( + opaque: *mut c_void, + src_file_size: u64, + compression_level: c_int, + checksum_flag: c_int, + block_size: usize, + buffer_size: usize, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("prepare"); + state.prepare_args.push(( + src_file_size, + compression_level, + checksum_flag, + block_size, + buffer_size, + )); + } + + unsafe extern "C" fn lz4_test_begin( + opaque: *mut c_void, + _output: *mut u8, + _output_size: usize, + result: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("begin"); + unsafe { *result = state.begin_result }; + state.begin_status + } + + unsafe extern "C" fn lz4_test_update( + opaque: *mut c_void, + output: *mut u8, + output_size: usize, + _input: *const u8, + input_size: usize, + result: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("update"); + assert!(input_size != 0); + if state.update_status == 0 { + assert!(output_size >= state.update_result); + unsafe { + *output = b'u'; + *result = state.update_result; + } + } else { + unsafe { *result = state.update_result }; + } + state.update_status + } + + unsafe extern "C" fn lz4_test_end( + opaque: *mut c_void, + _output: *mut u8, + _output_size: usize, + result: *mut usize, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("end"); + unsafe { *result = state.end_result }; + state.end_status + } + + unsafe extern "C" fn lz4_test_free(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("free"); + } + + unsafe extern "C" fn lz4_test_read_fill( + opaque: *mut c_void, + requested: usize, + buffer: *mut *const u8, + loaded: *mut usize, + ) -> usize { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("fill"); + let available = state.input_len - state.input_pos; + let amount = available.min(requested); + unsafe { + *buffer = state.input.as_ptr().add(state.input_pos); + *loaded = amount; + } + amount + } + + unsafe extern "C" fn lz4_test_read_consume(opaque: *mut c_void, amount: usize) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("consume"); + assert!(amount <= state.input_len - state.input_pos); + state.input_pos += amount; + state.consumed.push(amount); + } + + unsafe extern "C" fn lz4_test_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.events.push("acquire"); + state.acquire_calls += 1; + unsafe { + *job = opaque; + *buffer = state.output_buffer.as_mut_ptr(); + *buffer_size = state.output_buffer.len(); + } + } + + unsafe extern "C" fn lz4_test_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::() }; + state.events.push("enqueue"); + assert_eq!(unsafe { *job }, opaque); + assert!(used <= state.output_buffer.len()); + state.enqueue_sizes.push(used); + unsafe { + *buffer = state.output_buffer.as_mut_ptr(); + *buffer_size = state.output_buffer.len(); + } + } + + unsafe extern "C" fn lz4_test_write_release(opaque: *mut c_void, job: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("release"); + assert_eq!(job, opaque); + state.release_calls += 1; + } + + unsafe extern "C" fn lz4_test_sparse_end(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("sparse"); + state.sparse_end_calls += 1; + } + + unsafe extern "C" fn lz4_test_progress( + opaque: *mut c_void, + src_file_size: u64, + in_file_size: u64, + out_file_size: u64, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push("progress"); + state + .progress + .push((src_file_size, in_file_size, out_file_size)); + } + + fn lz4_test_projection(state: &mut Lz4ProjectionState) -> FIO_rust_lz4_compress_projection_t { + let opaque = (state as *mut Lz4ProjectionState).cast::(); + FIO_rust_lz4_compress_projection_t { + read_opaque: opaque, + write_opaque: opaque, + codec_opaque: opaque, + progress_opaque: opaque, + version: 100, + block_size: 3, + create: Some(lz4_test_create), + prepare: Some(lz4_test_prepare), + begin: Some(lz4_test_begin), + update: Some(lz4_test_update), + end: Some(lz4_test_end), + free_context: Some(lz4_test_free), + read_fill: Some(lz4_test_read_fill), + read_consume: Some(lz4_test_read_consume), + write_acquire: Some(lz4_test_write_acquire), + write_enqueue: Some(lz4_test_write_enqueue), + write_release: Some(lz4_test_write_release), + sparse_write_end: Some(lz4_test_sparse_end), + progress: Some(lz4_test_progress), + } + } + + #[test] + fn lz4_projection_preserves_header_update_end_order_and_accounting() { + let mut state = Lz4ProjectionState::default(); + let projection = lz4_test_projection(&mut state); + let mut read_size = 0; + let mut compressed_size = 0; + let mut lz4_result = 0; + + assert_eq!( + unsafe { + FIO_rust_compressLz4Frame( + &projection, + c"lz4-test".as_ptr(), + 7, + 12, + 1, + &mut read_size, + &mut compressed_size, + &mut lz4_result, + ) + }, + FIO_RUST_LZ4_OK + ); + assert_eq!(read_size, 7); + assert_eq!(compressed_size, 10); + assert_eq!(lz4_result, 0); + assert_eq!(state.prepare_args, vec![(7, 12, 1, 3, 8)]); + assert_eq!(state.consumed, vec![3, 3, 1]); + assert_eq!(state.enqueue_sizes, vec![1, 2, 2, 2, 3]); + assert_eq!(state.progress, vec![(7, 3, 3), (7, 6, 5), (7, 7, 7)]); + assert_eq!(state.acquire_calls, 1); + assert_eq!(state.release_calls, 1); + assert_eq!(state.sparse_end_calls, 1); + assert_eq!( + state.events, + vec![ + "acquire", "create", "prepare", "begin", "enqueue", "fill", "update", "progress", + "enqueue", "consume", "fill", "update", "progress", "enqueue", "consume", "fill", + "update", "progress", "enqueue", "consume", "fill", "end", "enqueue", "free", + "release", "sparse", + ] + ); + } + + #[test] + fn lz4_projection_reports_creation_header_update_and_end_errors() { + let cases = [ + ( + Lz4ProjectionState { + create_status: 1, + create_result: 17, + ..Lz4ProjectionState::default() + }, + FIO_RUST_LZ4_CREATE_ERROR, + 17, + vec!["acquire", "create"], + ), + ( + Lz4ProjectionState { + begin_status: 1, + begin_result: 23, + ..Lz4ProjectionState::default() + }, + FIO_RUST_LZ4_HEADER_ERROR, + 23, + vec!["acquire", "create", "prepare", "begin"], + ), + ( + Lz4ProjectionState { + update_status: 1, + update_result: 29, + ..Lz4ProjectionState::default() + }, + FIO_RUST_LZ4_UPDATE_ERROR, + 29, + vec![ + "acquire", "create", "prepare", "begin", "enqueue", "fill", "update", + ], + ), + ( + Lz4ProjectionState { + input_len: 0, + end_status: 1, + end_result: 31, + ..Lz4ProjectionState::default() + }, + FIO_RUST_LZ4_END_ERROR, + 31, + vec![ + "acquire", "create", "prepare", "begin", "enqueue", "fill", "end", + ], + ), + ]; + + for (mut state, expected_status, expected_result, expected_events) in cases { + let projection = lz4_test_projection(&mut state); + let mut read_size = 41; + let mut compressed_size = 43; + let mut lz4_result = 0; + assert_eq!( + unsafe { + FIO_rust_compressLz4Frame( + &projection, + c"lz4-test".as_ptr(), + 7, + 1, + 0, + &mut read_size, + &mut compressed_size, + &mut lz4_result, + ) + }, + expected_status + ); + assert_eq!(lz4_result, expected_result); + assert_eq!(read_size, 0); + assert_eq!(compressed_size, 0); + assert_eq!(state.events, expected_events); + assert_eq!(state.release_calls, 0); + assert_eq!(state.sparse_end_calls, 0); + } + } + #[test] fn classifies_all_cli_decompression_headers() { let is_mock_zstd = |buffer: &[u8]| buffer.starts_with(&[0x28, 0xB5, 0x2F, 0xFD]);