feat(cli): move xz/lzma and lz4 codec loops into Rust
Move the xz/lzma and LZ4 file-compression orchestration into Rust projection functions. C retains liblzma and LZ4F contexts, asynchronous pool ownership, format-specific callbacks, sparse-output handling, and user-facing diagnostics. The Rust leaves preserve input accounting, output enqueue order, progress updates, finish/end handling, cleanup, and the original error-code mapping. Focused projection tests cover success ordering and codec initialization, header, update, finish, and end failures. Test Plan: - cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1 - cargo test --manifest-path rust/cli/Cargo.toml --all-targets -- --test-threads=1 - cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings - make -B -C programs -j2 zstd - make -B -C tests -j2 test-cli-tests test-zstd - FUZZERTEST=-T5s make -B -C tests -j2 test-fuzzer
This commit is contained in:
+447
-146
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user