From b6611f9f902e52cd898c48d8a273a668885d4866 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 18 Jul 2026 01:43:25 +0200 Subject: [PATCH] feat(fileio): port async pools to Rust Replace programs/fileio_asyncio.c with a declaration-only shim and register Rust's async file-I/O implementation in the library archive. Preserve the C ABI for pool and job contexts, including DEBUGLEVEL pointer-based pthread wrappers, and select separate Rust build configurations for those layouts. Keep ordered-read queue state locked while waiting, drain queued work before shutdown frees job buffers, and use platform large-file seeks for sparse writes. Test Plan: - `cargo test --no-default-features --lib fileio_asyncio` (7 passed). - `cargo test --no-default-features --features debug-pthread --lib fileio_asyncio` (7 passed). - `cargo check --all-targets`, clippy library/benches/tests, and nightly fmt check (passed). - Threaded program build and README round-trip (passed). - `make -C tests poolTests` and `./tests/poolTests` (passed). - Current C shim object defines no `AIO_*` symbols; the Rust archive exports all 20 declarations. - Full standalone all-target Rust tests and the default CLI archive rebuild remain affected by unrelated C/Rust integration and another worker's in-progress benchmark changes. --- programs/Makefile | 26 +- programs/fileio_asyncio.c | 658 +------------- rust/Cargo.toml | 2 + rust/src/fileio_asyncio.rs | 1742 ++++++++++++++++++++++++++++++++++++ rust/src/lib.rs | 1 + 5 files changed, 1773 insertions(+), 656 deletions(-) create mode 100644 rust/src/fileio_asyncio.rs diff --git a/programs/Makefile b/programs/Makefile index d8696d207..020ea6837 100644 --- a/programs/Makefile +++ b/programs/Makefile @@ -85,7 +85,13 @@ ifeq ($(shell test $(ZSTD_LEGACY_SUPPORT) -lt 8; echo $$?), 0) RUST_LEGACY_FEATURES := $(addprefix legacy-v0,$(wordlist $(ZSTD_LEGACY_SUPPORT),7,1 2 3 4 5 6 7)) endif endif -RUST_BUILD_CONFIG := $(RUST_HUF_MODE)-legacy$(ZSTD_LEGACY_SUPPORT) +RUST_DEBUG_FEATURE := +RUST_DEBUG_MODE := release +ifneq ($(filter-out 0,$(DEBUGLEVEL)),) +RUST_DEBUG_FEATURE := debug-pthread +RUST_DEBUG_MODE := debug +endif +RUST_BUILD_CONFIG := $(RUST_HUF_MODE)-legacy$(ZSTD_LEGACY_SUPPORT)-$(RUST_DEBUG_MODE) RUST_TARGET_DIR := $(RUST_DIR)/target/$(RUST_BUILD_CONFIG) RUST_STATICLIB := $(RUST_TARGET_DIR)/release/libzstd_rs.a @@ -100,6 +106,9 @@ endif ifneq ($(RUST_LEGACY_FEATURES),) RUST_CARGO_FLAGS += --features $(subst $(space),$(comma),$(strip $(RUST_LEGACY_FEATURES))) endif +ifneq ($(RUST_DEBUG_FEATURE),) +RUST_CARGO_FLAGS += --features $(RUST_DEBUG_FEATURE) +endif $(RUST_STATICLIB): $(RUST_SOURCES) $(CARGO) build $(RUST_CARGO_FLAGS) @@ -121,7 +130,7 @@ $(RUST_CLI_STATICLIB): $(RUST_CLI_SOURCES) $(RUST_CLI_STATICLIB_32): $(RUST_CLI_SOURCES) $(CARGO) build $(RUST_CLI_CARGO_FLAGS) --target $(RUST_TARGET_32) -RUST_DECOMPRESS_BUILD_CONFIG := lib-c0-d1-b0-$(RUST_HUF_MODE) +RUST_DECOMPRESS_BUILD_CONFIG := lib-c0-d1-b0-$(RUST_HUF_MODE)-$(RUST_DEBUG_MODE) RUST_DECOMPRESS_TARGET_DIR := $(RUST_DIR)/target/$(RUST_DECOMPRESS_BUILD_CONFIG) RUST_DECOMPRESS_STATICLIB := $(RUST_DECOMPRESS_TARGET_DIR)/release/libzstd_rs.a RUST_DECOMPRESS_CARGO_FLAGS := --manifest-path $(RUST_MANIFEST) --release \ @@ -130,6 +139,9 @@ RUST_DECOMPRESS_CARGO_FLAGS := --manifest-path $(RUST_MANIFEST) --release \ ifneq ($(RUST_HUF_FEATURE),) RUST_DECOMPRESS_CARGO_FLAGS += --features $(RUST_HUF_FEATURE) endif +ifneq ($(RUST_DEBUG_FEATURE),) +RUST_DECOMPRESS_CARGO_FLAGS += --features $(RUST_DEBUG_FEATURE) +endif $(RUST_DECOMPRESS_STATICLIB): $(RUST_SOURCES) $(CARGO) build $(RUST_DECOMPRESS_CARGO_FLAGS) @@ -144,23 +156,29 @@ RUST_DECOMPRESS_CLI_CARGO_FLAGS := --manifest-path $(RUST_CLI_MANIFEST) --releas $(RUST_DECOMPRESS_CLI_STATICLIB): $(RUST_CLI_SOURCES) $(CARGO) build $(RUST_DECOMPRESS_CLI_CARGO_FLAGS) -RUST_COMPRESS_BUILD_CONFIG := lib-c1-d0-b0-$(RUST_HUF_MODE) +RUST_COMPRESS_BUILD_CONFIG := lib-c1-d0-b0-$(RUST_HUF_MODE)-$(RUST_DEBUG_MODE) RUST_COMPRESS_TARGET_DIR := $(RUST_DIR)/target/$(RUST_COMPRESS_BUILD_CONFIG) RUST_COMPRESS_STATICLIB := $(RUST_COMPRESS_TARGET_DIR)/release/libzstd_rs.a RUST_COMPRESS_CARGO_FLAGS := --manifest-path $(RUST_MANIFEST) --release \ --target-dir $(RUST_COMPRESS_TARGET_DIR) \ --no-default-features --features compression +ifneq ($(RUST_DEBUG_FEATURE),) +RUST_COMPRESS_CARGO_FLAGS += --features $(RUST_DEBUG_FEATURE) +endif $(RUST_COMPRESS_STATICLIB): $(RUST_SOURCES) $(CARGO) build $(RUST_COMPRESS_CARGO_FLAGS) -RUST_DICTBUILDER_BUILD_CONFIG := lib-c1-d0-b1-$(RUST_HUF_MODE) +RUST_DICTBUILDER_BUILD_CONFIG := lib-c1-d0-b1-$(RUST_HUF_MODE)-$(RUST_DEBUG_MODE) RUST_DICTBUILDER_TARGET_DIR := $(RUST_DIR)/target/$(RUST_DICTBUILDER_BUILD_CONFIG) RUST_DICTBUILDER_STATICLIB := $(RUST_DICTBUILDER_TARGET_DIR)/release/libzstd_rs.a RUST_DICTBUILDER_CARGO_FLAGS := --manifest-path $(RUST_MANIFEST) --release \ --target-dir $(RUST_DICTBUILDER_TARGET_DIR) \ --no-default-features \ --features compression,dict-builder +ifneq ($(RUST_DEBUG_FEATURE),) +RUST_DICTBUILDER_CARGO_FLAGS += --features $(RUST_DEBUG_FEATURE) +endif $(RUST_DICTBUILDER_STATICLIB): $(RUST_SOURCES) $(CARGO) build $(RUST_DICTBUILDER_CARGO_FLAGS) diff --git a/programs/fileio_asyncio.c b/programs/fileio_asyncio.c index 42a472016..0f9e89461 100644 --- a/programs/fileio_asyncio.c +++ b/programs/fileio_asyncio.c @@ -8,656 +8,10 @@ * You may select, at your option, one of the above-listed licenses. */ -#include "platform.h" -#include /* fprintf, open, fdopen, fread, _fileno, stdin, stdout */ -#include /* malloc, free */ -#include -#include /* errno */ - -#if defined (_MSC_VER) -# include -# include -#endif - +/* + * The async file-I/O implementation lives in rust/src/fileio_asyncio.rs. + * Keep this translation unit in the program source list so existing C build + * rules and fileio.c callers continue to include the public declarations, but + * do not provide a second definition of the AIO_* symbols. + */ #include "fileio_asyncio.h" -#include "fileio_common.h" - -/* ********************************************************************** - * Sparse write - ************************************************************************/ - -/** AIO_fwriteSparse() : -* @return : storedSkips, -* argument for next call to AIO_fwriteSparse() or AIO_fwriteSparseEnd() */ -static unsigned -AIO_fwriteSparse(FILE* file, - const void* buffer, size_t bufferSize, - const FIO_prefs_t* const prefs, - unsigned storedSkips) -{ - const size_t* const bufferT = (const size_t*)buffer; /* Buffer is supposed malloc'ed, hence aligned on size_t */ - size_t bufferSizeT = bufferSize / sizeof(size_t); - const size_t* const bufferTEnd = bufferT + bufferSizeT; - const size_t* ptrT = bufferT; - static const size_t segmentSizeT = (32 KB) / sizeof(size_t); /* check every 32 KB */ - - if (prefs->testMode) return 0; /* do not output anything in test mode */ - - if (!prefs->sparseFileSupport) { /* normal write */ - size_t const sizeCheck = fwrite(buffer, 1, bufferSize, file); - if (sizeCheck != bufferSize) - EXM_THROW(70, "Write error : cannot write block : %s", - strerror(errno)); - return 0; - } - - /* avoid int overflow */ - if (storedSkips > 1 GB) { - if (LONG_SEEK(file, 1 GB, SEEK_CUR) != 0) - EXM_THROW(91, "1 GB skip error (sparse file support)"); - storedSkips -= 1 GB; - } - - while (ptrT < bufferTEnd) { - size_t nb0T; - - /* adjust last segment if < 32 KB */ - size_t seg0SizeT = segmentSizeT; - if (seg0SizeT > bufferSizeT) seg0SizeT = bufferSizeT; - bufferSizeT -= seg0SizeT; - - /* count leading zeroes */ - for (nb0T=0; (nb0T < seg0SizeT) && (ptrT[nb0T] == 0); nb0T++) ; - storedSkips += (unsigned)(nb0T * sizeof(size_t)); - - if (nb0T != seg0SizeT) { /* not all 0s */ - size_t const nbNon0ST = seg0SizeT - nb0T; - /* skip leading zeros */ - if (LONG_SEEK(file, storedSkips, SEEK_CUR) != 0) - EXM_THROW(92, "Sparse skip error ; try --no-sparse"); - storedSkips = 0; - /* write the rest */ - if (fwrite(ptrT + nb0T, sizeof(size_t), nbNon0ST, file) != nbNon0ST) - EXM_THROW(93, "Write error : cannot write block : %s", - strerror(errno)); - } - ptrT += seg0SizeT; - } - - { static size_t const maskT = sizeof(size_t)-1; - if (bufferSize & maskT) { - /* size not multiple of sizeof(size_t) : implies end of block */ - const char* const restStart = (const char*)bufferTEnd; - const char* restPtr = restStart; - const char* const restEnd = (const char*)buffer + bufferSize; - assert(restEnd > restStart && restEnd < restStart + sizeof(size_t)); - for ( ; (restPtr < restEnd) && (*restPtr == 0); restPtr++) ; - storedSkips += (unsigned) (restPtr - restStart); - if (restPtr != restEnd) { - /* not all remaining bytes are 0 */ - size_t const restSize = (size_t)(restEnd - restPtr); - if (LONG_SEEK(file, storedSkips, SEEK_CUR) != 0) - EXM_THROW(92, "Sparse skip error ; try --no-sparse"); - if (fwrite(restPtr, 1, restSize, file) != restSize) - EXM_THROW(95, "Write error : cannot write end of decoded block : %s", - strerror(errno)); - storedSkips = 0; - } } } - - return storedSkips; -} - -static void -AIO_fwriteSparseEnd(const FIO_prefs_t* const prefs, FILE* file, unsigned storedSkips) -{ - if (prefs->testMode) assert(storedSkips == 0); - if (storedSkips>0) { - assert(prefs->sparseFileSupport > 0); /* storedSkips>0 implies sparse support is enabled */ - (void)prefs; /* assert can be disabled, in which case prefs becomes unused */ - if (LONG_SEEK(file, storedSkips-1, SEEK_CUR) != 0) - EXM_THROW(69, "Final skip error (sparse file support)"); - /* last zero must be explicitly written, - * so that skipped ones get implicitly translated as zero by FS */ - { const char lastZeroByte[1] = { 0 }; - if (fwrite(lastZeroByte, 1, 1, file) != 1) - EXM_THROW(69, "Write error : cannot write last zero : %s", strerror(errno)); - } } -} - - -/* ********************************************************************** - * AsyncIO functionality - ************************************************************************/ - -/* AIO_supported: - * Returns 1 if AsyncIO is supported on the system, 0 otherwise. */ -int AIO_supported(void) { -#ifdef ZSTD_MULTITHREAD - return 1; -#else - return 0; -#endif -} - -/* *********************************** - * Generic IoPool implementation - *************************************/ - -static IOJob_t *AIO_IOPool_createIoJob(IOPoolCtx_t *ctx, size_t bufferSize) { - IOJob_t* const job = (IOJob_t*) malloc(sizeof(IOJob_t)); - void* const buffer = malloc(bufferSize); - if(!job || !buffer) - EXM_THROW(101, "Allocation error : not enough memory"); - job->buffer = buffer; - job->bufferSize = bufferSize; - job->usedBufferSize = 0; - job->file = NULL; - job->ctx = ctx; - job->offset = 0; - return job; -} - - -/* AIO_IOPool_createThreadPool: - * Creates a thread pool and a mutex for threaded IO pool. - * Displays warning if asyncio is requested but MT isn't available. */ -static void AIO_IOPool_createThreadPool(IOPoolCtx_t* ctx, const FIO_prefs_t* prefs) { - ctx->threadPool = NULL; - ctx->threadPoolActive = 0; - if(prefs->asyncIO) { - if (ZSTD_pthread_mutex_init(&ctx->ioJobsMutex, NULL)) - EXM_THROW(102,"Failed creating ioJobsMutex mutex"); - /* We want MAX_IO_JOBS-2 queue items because we need to always have 1 free buffer to - * decompress into and 1 buffer that's actively written to disk and owned by the writing thread. */ - assert(MAX_IO_JOBS >= 2); - ctx->threadPool = POOL_create(1, MAX_IO_JOBS - 2); - ctx->threadPoolActive = 1; - if (!ctx->threadPool) - EXM_THROW(104, "Failed creating I/O thread pool"); - } -} - -/* AIO_IOPool_init: - * Allocates and sets and a new I/O thread pool including its included availableJobs. */ -static void AIO_IOPool_init(IOPoolCtx_t* ctx, const FIO_prefs_t* prefs, POOL_function poolFunction, size_t bufferSize) { - int i; - AIO_IOPool_createThreadPool(ctx, prefs); - ctx->prefs = prefs; - ctx->poolFunction = poolFunction; - ctx->totalIoJobs = ctx->threadPool ? MAX_IO_JOBS : 2; - ctx->availableJobsCount = ctx->totalIoJobs; - for(i=0; i < ctx->availableJobsCount; i++) { - ctx->availableJobs[i] = AIO_IOPool_createIoJob(ctx, bufferSize); - } - ctx->jobBufferSize = bufferSize; - ctx->file = NULL; -} - - -/* AIO_IOPool_threadPoolActive: - * Check if current operation uses thread pool. - * Note that in some cases we have a thread pool initialized but choose not to use it. */ -static int AIO_IOPool_threadPoolActive(IOPoolCtx_t* ctx) { - return ctx->threadPool && ctx->threadPoolActive; -} - - -/* AIO_IOPool_lockJobsMutex: - * Locks the IO jobs mutex if threading is active */ -static void AIO_IOPool_lockJobsMutex(IOPoolCtx_t* ctx) { - if(AIO_IOPool_threadPoolActive(ctx)) - ZSTD_pthread_mutex_lock(&ctx->ioJobsMutex); -} - -/* AIO_IOPool_unlockJobsMutex: - * Unlocks the IO jobs mutex if threading is active */ -static void AIO_IOPool_unlockJobsMutex(IOPoolCtx_t* ctx) { - if(AIO_IOPool_threadPoolActive(ctx)) - ZSTD_pthread_mutex_unlock(&ctx->ioJobsMutex); -} - -/* AIO_IOPool_releaseIoJob: - * Releases an acquired job back to the pool. Doesn't execute the job. */ -static void AIO_IOPool_releaseIoJob(IOJob_t* job) { - IOPoolCtx_t* const ctx = (IOPoolCtx_t *) job->ctx; - AIO_IOPool_lockJobsMutex(ctx); - assert(ctx->availableJobsCount < ctx->totalIoJobs); - ctx->availableJobs[ctx->availableJobsCount++] = job; - AIO_IOPool_unlockJobsMutex(ctx); -} - -/* AIO_IOPool_join: - * Waits for all tasks in the pool to finish executing. */ -static void AIO_IOPool_join(IOPoolCtx_t* ctx) { - if(AIO_IOPool_threadPoolActive(ctx)) - POOL_joinJobs(ctx->threadPool); -} - -/* AIO_IOPool_setThreaded: - * Allows (de)activating threaded mode, to be used when the expected overhead - * of threading costs more than the expected gains. */ -static void AIO_IOPool_setThreaded(IOPoolCtx_t* ctx, int threaded) { - assert(threaded == 0 || threaded == 1); - assert(ctx != NULL); - if(ctx->threadPoolActive != threaded) { - AIO_IOPool_join(ctx); - ctx->threadPoolActive = threaded; - } -} - -/* AIO_IOPool_free: - * Release a previously allocated IO thread pool. Makes sure all tasks are done and released. */ -static void AIO_IOPool_destroy(IOPoolCtx_t* ctx) { - int i; - if(ctx->threadPool) { - /* Make sure we finish all tasks and then free the resources */ - AIO_IOPool_join(ctx); - /* Make sure we are not leaking availableJobs */ - assert(ctx->availableJobsCount == ctx->totalIoJobs); - POOL_free(ctx->threadPool); - ZSTD_pthread_mutex_destroy(&ctx->ioJobsMutex); - } - assert(ctx->file == NULL); - for(i=0; iavailableJobsCount; i++) { - IOJob_t* job = (IOJob_t*) ctx->availableJobs[i]; - free(job->buffer); - free(job); - } -} - -/* AIO_IOPool_acquireJob: - * Returns an available io job to be used for a future io. */ -static IOJob_t* AIO_IOPool_acquireJob(IOPoolCtx_t* ctx) { - IOJob_t* job; - assert(ctx->file != NULL || ctx->prefs->testMode); - AIO_IOPool_lockJobsMutex(ctx); - assert(ctx->availableJobsCount > 0); - job = (IOJob_t*) ctx->availableJobs[--ctx->availableJobsCount]; - AIO_IOPool_unlockJobsMutex(ctx); - job->usedBufferSize = 0; - job->file = ctx->file; - job->offset = 0; - return job; -} - - -/* AIO_IOPool_setFile: - * Sets the destination file for future files in the pool. - * Requires completion of all queued jobs and release of all otherwise acquired jobs. */ -static void AIO_IOPool_setFile(IOPoolCtx_t* ctx, FILE* file) { - assert(ctx!=NULL); - AIO_IOPool_join(ctx); - assert(ctx->availableJobsCount == ctx->totalIoJobs); - ctx->file = file; -} - -static FILE* AIO_IOPool_getFile(const IOPoolCtx_t* ctx) { - return ctx->file; -} - -/* AIO_IOPool_enqueueJob: - * Enqueues an io job for execution. - * The queued job shouldn't be used directly after queueing it. */ -static void AIO_IOPool_enqueueJob(IOJob_t* job) { - IOPoolCtx_t* const ctx = (IOPoolCtx_t *)job->ctx; - if(AIO_IOPool_threadPoolActive(ctx)) - POOL_add(ctx->threadPool, ctx->poolFunction, job); - else - ctx->poolFunction(job); -} - -/* *********************************** - * WritePool implementation - *************************************/ - -/* AIO_WritePool_acquireJob: - * Returns an available write job to be used for a future write. */ -IOJob_t* AIO_WritePool_acquireJob(WritePoolCtx_t* ctx) { - return AIO_IOPool_acquireJob(&ctx->base); -} - -/* AIO_WritePool_enqueueAndReacquireWriteJob: - * Queues a write job for execution and acquires a new one. - * After execution `job`'s pointed value would change to the newly acquired job. - * Make sure to set `usedBufferSize` to the wanted length before call. - * The queued job shouldn't be used directly after queueing it. */ -void AIO_WritePool_enqueueAndReacquireWriteJob(IOJob_t **job) { - AIO_IOPool_enqueueJob(*job); - *job = AIO_IOPool_acquireJob((IOPoolCtx_t *)(*job)->ctx); -} - -/* AIO_WritePool_sparseWriteEnd: - * Ends sparse writes to the current file. - * Blocks on completion of all current write jobs before executing. */ -void AIO_WritePool_sparseWriteEnd(WritePoolCtx_t* ctx) { - assert(ctx != NULL); - AIO_IOPool_join(&ctx->base); - AIO_fwriteSparseEnd(ctx->base.prefs, ctx->base.file, ctx->storedSkips); - ctx->storedSkips = 0; -} - -/* AIO_WritePool_setFile: - * Sets the destination file for future writes in the pool. - * Requires completion of all queues write jobs and release of all otherwise acquired jobs. - * Also requires ending of sparse write if a previous file was used in sparse mode. */ -void AIO_WritePool_setFile(WritePoolCtx_t* ctx, FILE* file) { - AIO_IOPool_setFile(&ctx->base, file); - assert(ctx->storedSkips == 0); -} - -/* AIO_WritePool_getFile: - * Returns the file the writePool is currently set to write to. */ -FILE* AIO_WritePool_getFile(const WritePoolCtx_t* ctx) { - return AIO_IOPool_getFile(&ctx->base); -} - -/* AIO_WritePool_releaseIoJob: - * Releases an acquired job back to the pool. Doesn't execute the job. */ -void AIO_WritePool_releaseIoJob(IOJob_t* job) { - AIO_IOPool_releaseIoJob(job); -} - -/* AIO_WritePool_closeFile: - * Ends sparse write and closes the writePool's current file and sets the file to NULL. - * Requires completion of all queues write jobs and release of all otherwise acquired jobs. */ -int AIO_WritePool_closeFile(WritePoolCtx_t* ctx) { - FILE* const dstFile = ctx->base.file; - assert(dstFile!=NULL || ctx->base.prefs->testMode!=0); - AIO_WritePool_sparseWriteEnd(ctx); - AIO_IOPool_setFile(&ctx->base, NULL); - return fclose(dstFile); -} - -/* AIO_WritePool_executeWriteJob: - * Executes a write job synchronously. Can be used as a function for a thread pool. */ -static void AIO_WritePool_executeWriteJob(void* opaque){ - IOJob_t* const job = (IOJob_t*) opaque; - WritePoolCtx_t* const ctx = (WritePoolCtx_t*) job->ctx; - ctx->storedSkips = AIO_fwriteSparse(job->file, job->buffer, job->usedBufferSize, ctx->base.prefs, ctx->storedSkips); - AIO_IOPool_releaseIoJob(job); -} - -/* AIO_WritePool_create: - * Allocates and sets and a new write pool including its included jobs. */ -WritePoolCtx_t* AIO_WritePool_create(const FIO_prefs_t* prefs, size_t bufferSize) { - WritePoolCtx_t* const ctx = (WritePoolCtx_t*) malloc(sizeof(WritePoolCtx_t)); - if(!ctx) EXM_THROW(100, "Allocation error : not enough memory"); - AIO_IOPool_init(&ctx->base, prefs, AIO_WritePool_executeWriteJob, bufferSize); - ctx->storedSkips = 0; - return ctx; -} - -/* AIO_WritePool_free: - * Frees and releases a writePool and its resources. Closes destination file if needs to. */ -void AIO_WritePool_free(WritePoolCtx_t* ctx) { - /* Make sure we finish all tasks and then free the resources */ - if(AIO_WritePool_getFile(ctx)) - AIO_WritePool_closeFile(ctx); - AIO_IOPool_destroy(&ctx->base); - assert(ctx->storedSkips==0); - free(ctx); -} - -/* AIO_WritePool_setAsync: - * Allows (de)activating async mode, to be used when the expected overhead - * of asyncio costs more than the expected gains. */ -void AIO_WritePool_setAsync(WritePoolCtx_t* ctx, int async) { - AIO_IOPool_setThreaded(&ctx->base, async); -} - - -/* *********************************** - * ReadPool implementation - *************************************/ -static void AIO_ReadPool_releaseAllCompletedJobs(ReadPoolCtx_t* ctx) { - int i; - for(i=0; icompletedJobsCount; i++) { - IOJob_t* job = (IOJob_t*) ctx->completedJobs[i]; - AIO_IOPool_releaseIoJob(job); - } - ctx->completedJobsCount = 0; -} - -static void AIO_ReadPool_addJobToCompleted(IOJob_t* job) { - ReadPoolCtx_t* const ctx = (ReadPoolCtx_t *)job->ctx; - AIO_IOPool_lockJobsMutex(&ctx->base); - assert(ctx->completedJobsCount < MAX_IO_JOBS); - ctx->completedJobs[ctx->completedJobsCount++] = job; - if(AIO_IOPool_threadPoolActive(&ctx->base)) { - ZSTD_pthread_cond_signal(&ctx->jobCompletedCond); - } - AIO_IOPool_unlockJobsMutex(&ctx->base); -} - -/* AIO_ReadPool_findNextWaitingOffsetCompletedJob_locked: - * Looks through the completed jobs for a job matching the waitingOnOffset and returns it, - * if job wasn't found returns NULL. - * IMPORTANT: assumes ioJobsMutex is locked. */ -static IOJob_t* AIO_ReadPool_findNextWaitingOffsetCompletedJob_locked(ReadPoolCtx_t* ctx) { - IOJob_t *job = NULL; - int i; - /* This implementation goes through all completed jobs and looks for the one matching the next offset. - * While not strictly needed for a single threaded reader implementation (as in such a case we could expect - * reads to be completed in order) this implementation was chosen as it better fits other asyncio - * interfaces (such as io_uring) that do not provide promises regarding order of completion. */ - for (i=0; icompletedJobsCount; i++) { - job = (IOJob_t *) ctx->completedJobs[i]; - if (job->offset == ctx->waitingOnOffset) { - ctx->completedJobs[i] = ctx->completedJobs[--ctx->completedJobsCount]; - return job; - } - } - return NULL; -} - -/* AIO_ReadPool_numReadsInFlight: - * Returns the number of IO read jobs currently in flight. */ -static size_t AIO_ReadPool_numReadsInFlight(ReadPoolCtx_t* ctx) { - const int jobsHeld = (ctx->currentJobHeld==NULL ? 0 : 1); - return (size_t)(ctx->base.totalIoJobs - (ctx->base.availableJobsCount + ctx->completedJobsCount + jobsHeld)); -} - -/* AIO_ReadPool_getNextCompletedJob: - * Returns a completed IOJob_t for the next read in line based on waitingOnOffset and advances waitingOnOffset. - * Would block. */ -static IOJob_t* AIO_ReadPool_getNextCompletedJob(ReadPoolCtx_t* ctx) { - IOJob_t *job = NULL; - AIO_IOPool_lockJobsMutex(&ctx->base); - - job = AIO_ReadPool_findNextWaitingOffsetCompletedJob_locked(ctx); - - /* As long as we didn't find the job matching the next read, and we have some reads in flight continue waiting */ - while (!job && (AIO_ReadPool_numReadsInFlight(ctx) > 0)) { - assert(ctx->base.threadPool != NULL); /* we shouldn't be here if we work in sync mode */ - ZSTD_pthread_cond_wait(&ctx->jobCompletedCond, &ctx->base.ioJobsMutex); - job = AIO_ReadPool_findNextWaitingOffsetCompletedJob_locked(ctx); - } - - if(job) { - assert(job->offset == ctx->waitingOnOffset); - ctx->waitingOnOffset += job->usedBufferSize; - } - - AIO_IOPool_unlockJobsMutex(&ctx->base); - return job; -} - - -/* AIO_ReadPool_executeReadJob: - * Executes a read job synchronously. Can be used as a function for a thread pool. */ -static void AIO_ReadPool_executeReadJob(void* opaque){ - IOJob_t* const job = (IOJob_t*) opaque; - ReadPoolCtx_t* const ctx = (ReadPoolCtx_t *)job->ctx; - if(ctx->reachedEof) { - job->usedBufferSize = 0; - AIO_ReadPool_addJobToCompleted(job); - return; - } - job->usedBufferSize = fread(job->buffer, 1, job->bufferSize, job->file); - if(job->usedBufferSize < job->bufferSize) { - if(ferror(job->file)) { - EXM_THROW(37, "Read error"); - } else if(feof(job->file)) { - ctx->reachedEof = 1; - } else { - EXM_THROW(37, "Unexpected short read"); - } - } - AIO_ReadPool_addJobToCompleted(job); -} - -static void AIO_ReadPool_enqueueRead(ReadPoolCtx_t* ctx) { - IOJob_t* const job = AIO_IOPool_acquireJob(&ctx->base); - job->offset = ctx->nextReadOffset; - ctx->nextReadOffset += job->bufferSize; - AIO_IOPool_enqueueJob(job); -} - -static void AIO_ReadPool_startReading(ReadPoolCtx_t* ctx) { - while(ctx->base.availableJobsCount) { - AIO_ReadPool_enqueueRead(ctx); - } -} - -/* AIO_ReadPool_setFile: - * Sets the source file for future read in the pool. Initiates reading immediately if file is not NULL. - * Waits for all current enqueued tasks to complete if a previous file was set. */ -void AIO_ReadPool_setFile(ReadPoolCtx_t* ctx, FILE* file) { - assert(ctx!=NULL); - AIO_IOPool_join(&ctx->base); - AIO_ReadPool_releaseAllCompletedJobs(ctx); - if (ctx->currentJobHeld) { - AIO_IOPool_releaseIoJob((IOJob_t *)ctx->currentJobHeld); - ctx->currentJobHeld = NULL; - } - AIO_IOPool_setFile(&ctx->base, file); - ctx->nextReadOffset = 0; - ctx->waitingOnOffset = 0; - ctx->srcBuffer = ctx->coalesceBuffer; - ctx->srcBufferLoaded = 0; - ctx->reachedEof = 0; - if(file != NULL) - AIO_ReadPool_startReading(ctx); -} - -/* AIO_ReadPool_create: - * Allocates and sets and a new readPool including its included jobs. - * bufferSize should be set to the maximal buffer we want to read at a time, will also be used - * as our basic read size. */ -ReadPoolCtx_t* AIO_ReadPool_create(const FIO_prefs_t* prefs, size_t bufferSize) { - ReadPoolCtx_t* const ctx = (ReadPoolCtx_t*) malloc(sizeof(ReadPoolCtx_t)); - if(!ctx) EXM_THROW(100, "Allocation error : not enough memory"); - AIO_IOPool_init(&ctx->base, prefs, AIO_ReadPool_executeReadJob, bufferSize); - - ctx->coalesceBuffer = (U8*) malloc(bufferSize * 2); - if(!ctx->coalesceBuffer) EXM_THROW(100, "Allocation error : not enough memory"); - ctx->srcBuffer = ctx->coalesceBuffer; - ctx->srcBufferLoaded = 0; - ctx->completedJobsCount = 0; - ctx->currentJobHeld = NULL; - - if(ctx->base.threadPool) - if (ZSTD_pthread_cond_init(&ctx->jobCompletedCond, NULL)) - EXM_THROW(103,"Failed creating jobCompletedCond cond"); - - return ctx; -} - -/* AIO_ReadPool_free: - * Frees and releases a readPool and its resources. Closes source file. */ -void AIO_ReadPool_free(ReadPoolCtx_t* ctx) { - if(AIO_ReadPool_getFile(ctx)) - AIO_ReadPool_closeFile(ctx); - if(ctx->base.threadPool) - ZSTD_pthread_cond_destroy(&ctx->jobCompletedCond); - AIO_IOPool_destroy(&ctx->base); - free(ctx->coalesceBuffer); - free(ctx); -} - -/* AIO_ReadPool_consumeBytes: - * Consumes byes from srcBuffer's beginning and updates srcBufferLoaded accordingly. */ -void AIO_ReadPool_consumeBytes(ReadPoolCtx_t* ctx, size_t n) { - assert(n <= ctx->srcBufferLoaded); - ctx->srcBufferLoaded -= n; - ctx->srcBuffer += n; -} - -/* AIO_ReadPool_releaseCurrentlyHeldAndGetNext: - * Release the current held job and get the next one, returns NULL if no next job available. */ -static IOJob_t* AIO_ReadPool_releaseCurrentHeldAndGetNext(ReadPoolCtx_t* ctx) { - if (ctx->currentJobHeld) { - AIO_IOPool_releaseIoJob((IOJob_t *)ctx->currentJobHeld); - ctx->currentJobHeld = NULL; - AIO_ReadPool_enqueueRead(ctx); - } - ctx->currentJobHeld = AIO_ReadPool_getNextCompletedJob(ctx); - return (IOJob_t*) ctx->currentJobHeld; -} - -/* AIO_ReadPool_fillBuffer: - * Tries to fill the buffer with at least n or jobBufferSize bytes (whichever is smaller). - * Returns if srcBuffer has at least the expected number of bytes loaded or if we've reached the end of the file. - * Return value is the number of bytes added to the buffer. - * Note that srcBuffer might have up to 2 times jobBufferSize bytes. */ -size_t AIO_ReadPool_fillBuffer(ReadPoolCtx_t* ctx, size_t n) { - IOJob_t *job; - int useCoalesce = 0; - if(n > ctx->base.jobBufferSize) - n = ctx->base.jobBufferSize; - - /* We are good, don't read anything */ - if (ctx->srcBufferLoaded >= n) - return 0; - - /* We still have bytes loaded, but not enough to satisfy caller. We need to get the next job - * and coalesce the remaining bytes with the next job's buffer */ - if (ctx->srcBufferLoaded > 0) { - useCoalesce = 1; - memcpy(ctx->coalesceBuffer, ctx->srcBuffer, ctx->srcBufferLoaded); - ctx->srcBuffer = ctx->coalesceBuffer; - } - - /* Read the next chunk */ - job = AIO_ReadPool_releaseCurrentHeldAndGetNext(ctx); - if(!job) - return 0; - if(useCoalesce) { - assert(ctx->srcBufferLoaded + job->usedBufferSize <= 2*ctx->base.jobBufferSize); - memcpy(ctx->coalesceBuffer + ctx->srcBufferLoaded, job->buffer, job->usedBufferSize); - ctx->srcBufferLoaded += job->usedBufferSize; - } - else { - ctx->srcBuffer = (U8 *) job->buffer; - ctx->srcBufferLoaded = job->usedBufferSize; - } - return job->usedBufferSize; -} - -/* AIO_ReadPool_consumeAndRefill: - * Consumes the current buffer and refills it with bufferSize bytes. */ -size_t AIO_ReadPool_consumeAndRefill(ReadPoolCtx_t* ctx) { - AIO_ReadPool_consumeBytes(ctx, ctx->srcBufferLoaded); - return AIO_ReadPool_fillBuffer(ctx, ctx->base.jobBufferSize); -} - -/* AIO_ReadPool_getFile: - * Returns the current file set for the read pool. */ -FILE* AIO_ReadPool_getFile(const ReadPoolCtx_t* ctx) { - return AIO_IOPool_getFile(&ctx->base); -} - -/* AIO_ReadPool_closeFile: - * Closes the current set file. Waits for all current enqueued tasks to complete and resets state. */ -int AIO_ReadPool_closeFile(ReadPoolCtx_t* ctx) { - FILE* const file = AIO_ReadPool_getFile(ctx); - AIO_ReadPool_setFile(ctx, NULL); - return fclose(file); -} - -/* AIO_ReadPool_setAsync: - * Allows (de)activating async mode, to be used when the expected overhead - * of asyncio costs more than the expected gains. */ -void AIO_ReadPool_setAsync(ReadPoolCtx_t* ctx, int async) { - AIO_IOPool_setThreaded(&ctx->base, async); -} diff --git a/rust/Cargo.toml b/rust/Cargo.toml index ec33fc8cf..4f1c7fc7b 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -13,6 +13,8 @@ decompression = [] dict-builder = [] huf-force-decompress-x1 = [] huf-force-decompress-x2 = [] +# POSIX DEBUGLEVEL >= 1 replaces pthread objects in the C ABI with pointers. +debug-pthread = [] # Legacy-format decoders (zstd v0.1 .. v0.7). Never default features: the # build systems map ZSTD_LEGACY_SUPPORT=N to the features for versions >= N, # exactly mirroring which lib/legacy/zstd_v0N.c files the C build compiles. diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs new file mode 100644 index 000000000..e642b1ad4 --- /dev/null +++ b/rust/src/fileio_asyncio.rs @@ -0,0 +1,1742 @@ +#![allow(non_camel_case_types)] +#![allow(non_snake_case)] +#![allow(clippy::missing_safety_doc)] + +//! Rust implementation of the exported asynchronous file-I/O pools. +//! +//! The C header exposes the read-pool buffer fields directly to `fileio.c`, +//! while the rest of each context is private implementation detail. The +//! public context types below are therefore deliberately opaque. Contexts +//! are allocated with the C allocator and begin with the exact C layout; the +//! layout is calculated at runtime because `threading.h` changes the mutex +//! and condition-variable fields when multithreading is disabled. +//! +//! The C translation unit remains in the program source list as a declaration +//! shim. The exported symbols below are the single implementation linked into +//! program, library, and C-test archives. + +use std::collections::VecDeque; +use std::ffi::c_void; +use std::mem::{align_of, size_of}; +#[cfg(any(windows, not(any(unix, windows))))] +use std::os::raw::c_long; +use std::os::raw::{c_int, c_uint}; +use std::ptr; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{self, JoinHandle}; + +const MAX_IO_JOBS: usize = 10; +const IO_QUEUE_SIZE: usize = MAX_IO_JOBS - 2; +const SPARSE_SEGMENT_SIZE: usize = 32 * 1024; +const SPARSE_SKIP_CHUNK: u64 = 1 << 30; + +/// C's `FIO_prefs_t` from `programs/fileio_types.h`. +/// +/// `fileio_prefs.rs` contains the same C layout for the preferences API. It +/// is repeated here so this module remains independently compilable; both +/// types are ABI-compatible and are only passed by pointer across the C ABI. +#[repr(C)] +pub struct FIO_prefs_t { + pub compressionType: c_int, + pub sparseFileSupport: c_int, + pub dictIDFlag: c_int, + pub checksumFlag: c_int, + pub blockSize: c_int, + pub overlapLog: c_int, + pub adaptiveMode: c_int, + pub useRowMatchFinder: c_int, + pub rsyncable: c_int, + pub minAdaptLevel: c_int, + pub maxAdaptLevel: c_int, + pub ldmFlag: c_int, + pub ldmHashLog: c_int, + pub ldmMinMatch: c_int, + pub ldmBucketSizeLog: c_int, + pub ldmHashRateLog: c_int, + pub streamSrcSize: usize, + pub targetCBlockSize: usize, + pub srcSizeHint: c_int, + pub testMode: c_int, + pub literalCompressionMode: c_int, + pub removeSrcFile: c_int, + pub overwrite: c_int, + pub asyncIO: c_int, + pub memLimit: c_uint, + pub nbWorkers: c_int, + pub excludeCompressedFiles: c_int, + pub patchFromMode: c_int, + pub contentSize: c_int, + pub allowBlockDevices: c_int, + pub passThrough: c_int, + pub mmapDict: c_int, +} + +/// Opaque C context handles. The actual allocations start with the C +/// `IOPoolCtx_t`, `ReadPoolCtx_t`, and `WritePoolCtx_t` layouts described in +/// `programs/fileio_asyncio.h`; Rust accesses them through `AbiLayout` below. +#[repr(C)] +pub struct IOPoolCtx_t { + _opaque: [u8; 0], +} + +#[repr(C)] +pub struct ReadPoolCtx_t { + _opaque: [u8; 0], +} + +#[repr(C)] +pub struct WritePoolCtx_t { + _opaque: [u8; 0], +} + +/// Public job layout from `programs/fileio_asyncio.h`. +#[repr(C)] +pub struct IOJob_t { + pub ctx: *mut c_void, + pub file: *mut libc::FILE, + pub buffer: *mut c_void, + pub bufferSize: usize, + pub usedBufferSize: usize, + pub offset: u64, +} + +type PoolFunction = unsafe extern "C" fn(*mut c_void); + +#[cfg(not(test))] +unsafe extern "C" { + /// Exported by `lib/common/pool.c`; this is the C preprocessor bridge for + /// `ZSTD_MULTITHREAD` used by the Rust pool implementation as well. + fn ZSTD_rust_pool_is_multithreaded() -> c_int; +} + +#[cfg(windows)] +unsafe extern "C" { + fn _fseeki64(file: *mut libc::FILE, offset: i64, whence: c_int) -> c_int; +} + +#[inline] +fn multithreading_enabled() -> bool { + #[cfg(test)] + { + true + } + #[cfg(not(test))] + { + unsafe { ZSTD_rust_pool_is_multithreaded() != 0 } + } +} + +/// A Windows `CRITICAL_SECTION` layout used only to reserve the C-visible +/// field. Rust owns synchronization, so no value of this type is initialized. +#[cfg(windows)] +#[repr(C)] +struct WindowsCriticalSection { + debug_info: *mut c_void, + lock_count: c_long, + recursion_count: c_long, + owning_thread: *mut c_void, + lock_semaphore: *mut c_void, + spin_count: usize, +} + +#[cfg(windows)] +#[repr(C)] +struct WindowsConditionVariable { + ptr: *mut c_void, +} + +#[allow(dead_code)] +#[derive(Clone, Copy, Debug)] +struct AbiLayout { + pointer_size: usize, + read_size: usize, + write_size: usize, + thread_pool: usize, + thread_pool_active: usize, + total_io_jobs: usize, + prefs: usize, + pool_function: usize, + file: usize, + io_jobs_mutex: usize, + available_jobs: usize, + available_jobs_count: usize, + job_buffer_size: usize, + write_stored_skips: usize, + read_reached_eof: usize, + read_next_offset: usize, + read_waiting_offset: usize, + read_current_job: usize, + read_coalesce_buffer: usize, + read_src_buffer: usize, + read_src_buffer_loaded: usize, + read_completed_jobs: usize, + read_completed_jobs_count: usize, + read_job_completed_cond: usize, +} + +#[inline] +const fn align_up(value: usize, alignment: usize) -> usize { + (value + alignment - 1) & !(alignment - 1) +} + +fn abi_layout(threaded: bool) -> AbiLayout { + let pointer_size = size_of::<*mut c_void>(); + let pointer_align = align_of::<*mut c_void>(); + let int_size = size_of::(); + let int_align = align_of::(); + let word_size = size_of::(); + let word_align = align_of::(); + let function_size = size_of::(); + let function_align = align_of::(); + + let (mutex_size, mutex_align, condition_size, condition_align) = if !threaded { + (int_size, int_align, int_size, int_align) + } else if cfg!(all(feature = "debug-pthread", unix)) { + // DEBUGLEVEL >= 1 uses pthread_mutex_t*/pthread_cond_t* in + // threading.h so that forgotten init/destroy calls remain visible. + (pointer_size, pointer_align, pointer_size, pointer_align) + } else { + #[cfg(unix)] + { + ( + size_of::(), + align_of::(), + size_of::(), + align_of::(), + ) + } + #[cfg(windows)] + { + ( + size_of::(), + align_of::(), + size_of::(), + align_of::(), + ) + } + #[cfg(not(any(unix, windows)))] + { + // `threading.h` assumes POSIX for other multithreaded targets. + // Keep a pointer-sized opaque reservation for such targets until + // their native synchronization representation is defined. + (pointer_size, pointer_align, pointer_size, pointer_align) + } + }; + + let mut offset = 0; + let thread_pool = offset; + offset += pointer_size; + let thread_pool_active = offset; + offset += int_size; + let total_io_jobs = offset; + offset += int_size; + offset = align_up(offset, pointer_align); + let prefs = offset; + offset += pointer_size; + offset = align_up(offset, function_align); + let pool_function = offset; + offset += function_size; + offset = align_up(offset, pointer_align); + let file = offset; + offset += pointer_size; + offset = align_up(offset, mutex_align); + let io_jobs_mutex = offset; + offset += mutex_size; + offset = align_up(offset, pointer_align); + let available_jobs = offset; + offset += MAX_IO_JOBS * pointer_size; + let available_jobs_count = offset; + offset += int_size; + offset = align_up(offset, word_align); + let job_buffer_size = offset; + offset += word_size; + let base_align = pointer_align + .max(word_align) + .max(mutex_align) + .max(int_align) + .max(function_align); + let base_size = align_up(offset, base_align); + + let write_stored_skips = base_size; + let write_align = base_align.max(int_align); + let write_size = align_up(write_stored_skips + size_of::(), write_align); + + let read_reached_eof = base_size; + offset = read_reached_eof + int_size; + offset = align_up(offset, align_of::()); + let read_next_offset = offset; + offset += size_of::(); + let read_waiting_offset = offset; + offset += size_of::(); + let read_current_job = offset; + offset += pointer_size; + let read_coalesce_buffer = offset; + offset += pointer_size; + let read_src_buffer = offset; + offset += pointer_size; + offset = align_up(offset, word_align); + let read_src_buffer_loaded = offset; + offset += word_size; + let read_completed_jobs = offset; + offset += MAX_IO_JOBS * pointer_size; + let read_completed_jobs_count = offset; + offset += int_size; + offset = align_up(offset, condition_align); + let read_job_completed_cond = offset; + offset += condition_size; + let read_align = base_align.max(condition_align).max(align_of::()); + let read_size = align_up(offset, read_align); + + AbiLayout { + pointer_size, + read_size, + write_size, + thread_pool, + thread_pool_active, + total_io_jobs, + prefs, + pool_function, + file, + io_jobs_mutex, + available_jobs, + available_jobs_count, + job_buffer_size, + write_stored_skips, + read_reached_eof, + read_next_offset, + read_waiting_offset, + read_current_job, + read_coalesce_buffer, + read_src_buffer, + read_src_buffer_loaded, + read_completed_jobs, + read_completed_jobs_count, + read_job_completed_cond, + } +} + +#[inline] +unsafe fn read_at(base: *const u8, offset: usize) -> T { + unsafe { base.add(offset).cast::().read() } +} + +#[inline] +unsafe fn write_at(base: *mut u8, offset: usize, value: T) { + unsafe { base.add(offset).cast::().write(value) }; +} + +#[inline] +unsafe fn base_file(base: *const u8, layout: AbiLayout) -> *mut libc::FILE { + unsafe { read_at(base, layout.file) } +} + +#[inline] +unsafe fn base_inner(base: *mut u8) -> &'static PoolInner { + let inner = unsafe { read_at::<*mut PoolInner>(base, 0) }; + assert!(!inner.is_null()); + unsafe { &*inner } +} + +#[inline] +fn fatal(code: c_int, message: &str) -> ! { + eprintln!("zstd: error {code} : {message}"); + std::process::exit(code); +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PoolKind { + Read, + Write, +} + +struct QueueState { + queued: VecDeque<*mut IOJob_t>, + running: bool, + stopping: bool, +} + +// A job is allocated by the C allocator and remains owned by the context +// until its callback has returned. The C API supplies that lifetime proof. +unsafe impl Send for QueueState {} + +struct AsyncQueue { + state: Mutex, + work_available: Condvar, + queue_space: Condvar, + idle: Condvar, + worker: Mutex>>, + threaded: bool, + kind: PoolKind, +} + +unsafe impl Send for AsyncQueue {} +unsafe impl Sync for AsyncQueue {} + +impl AsyncQueue { + fn new(threaded: bool, kind: PoolKind) -> Arc { + let queue = Arc::new(Self { + state: Mutex::new(QueueState { + queued: VecDeque::with_capacity(IO_QUEUE_SIZE), + running: false, + stopping: false, + }), + work_available: Condvar::new(), + queue_space: Condvar::new(), + idle: Condvar::new(), + worker: Mutex::new(None), + threaded, + kind, + }); + + if threaded { + let worker_queue = Arc::clone(&queue); + let worker = thread::Builder::new() + .spawn(move || worker_queue.worker_loop()) + .unwrap_or_else(|_| fatal(104, "Failed creating I/O thread pool")); + *queue + .worker + .lock() + .unwrap_or_else(|error| error.into_inner()) = Some(worker); + } + queue + } + + fn worker_loop(&self) { + loop { + let job = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + loop { + if let Some(job) = state.queued.pop_front() { + state.running = true; + self.queue_space.notify_one(); + break job; + } + if state.stopping { + return; + } + state = self + .work_available + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + }; + + unsafe { execute_job(self.kind, job) }; + + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.running = false; + self.idle.notify_all(); + if state.stopping && state.queued.is_empty() { + return; + } + } + } + + fn enqueue(&self, job: *mut IOJob_t) { + if !self.threaded { + unsafe { execute_job(self.kind, job) }; + return; + } + + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + while state.queued.len() >= IO_QUEUE_SIZE && !state.stopping { + state = self + .queue_space + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + if state.stopping { + return; + } + state.queued.push_back(job); + self.work_available.notify_one(); + } + + fn join(&self) { + if !self.threaded { + return; + } + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + while !state.queued.is_empty() || state.running { + state = self + .idle + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + } + + fn shutdown(&self) { + if !self.threaded { + return; + } + { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.stopping = true; + self.work_available.notify_all(); + } + let worker = self + .worker + .lock() + .unwrap_or_else(|error| error.into_inner()) + .take(); + if let Some(worker) = worker { + let _ = worker.join(); + } + } +} + +impl Drop for AsyncQueue { + fn drop(&mut self) { + self.shutdown(); + } +} + +struct JobState { + available: Vec<*mut IOJob_t>, + completed: Vec<*mut IOJob_t>, +} + +unsafe impl Send for JobState {} + +struct PoolInner { + context: *mut u8, + layout: AbiLayout, + prefs: *const FIO_prefs_t, + kind: PoolKind, + total_jobs: usize, + jobs: Mutex, + jobs_changed: Condvar, + queue: Option>, +} + +unsafe impl Send for PoolInner {} +unsafe impl Sync for PoolInner {} + +impl PoolInner { + fn new( + context: *mut u8, + layout: AbiLayout, + prefs: *const FIO_prefs_t, + kind: PoolKind, + pool_exists: bool, + worker_thread: bool, + ) -> Self { + let queue = pool_exists.then(|| AsyncQueue::new(worker_thread, kind)); + let total_jobs = if pool_exists { MAX_IO_JOBS } else { 2 }; + Self { + context, + layout, + prefs, + kind, + total_jobs, + jobs: Mutex::new(JobState { + available: Vec::with_capacity(total_jobs), + completed: Vec::with_capacity(MAX_IO_JOBS), + }), + jobs_changed: Condvar::new(), + queue, + } + } + + unsafe fn init_jobs(&self, buffer_size: usize) { + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + for _ in 0..self.total_jobs { + let job = libc::malloc(size_of::()).cast::(); + if job.is_null() { + fatal(101, "Allocation error: not enough memory"); + } + let allocation_size = buffer_size.max(1); + let buffer = libc::malloc(allocation_size); + if buffer.is_null() { + libc::free(job.cast()); + fatal(101, "Allocation error: not enough memory"); + } + unsafe { + job.write(IOJob_t { + ctx: self.context.cast(), + file: ptr::null_mut(), + buffer, + bufferSize: buffer_size, + usedBufferSize: 0, + offset: 0, + }); + } + state.available.push(job); + } + unsafe { self.sync_available_locked(&state) }; + } + + unsafe fn sync_available_locked(&self, state: &JobState) { + let base = self.context; + for index in 0..MAX_IO_JOBS { + let job = state + .available + .get(index) + .copied() + .unwrap_or(ptr::null_mut()); + unsafe { + write_at( + base, + self.layout.available_jobs + index * self.layout.pointer_size, + job.cast::(), + ); + } + } + unsafe { + write_at( + base, + self.layout.available_jobs_count, + state.available.len() as c_int, + ); + } + } + + unsafe fn sync_completed_locked(&self, state: &JobState) { + let base = self.context; + for index in 0..MAX_IO_JOBS { + let job = state + .completed + .get(index) + .copied() + .unwrap_or(ptr::null_mut()); + unsafe { + write_at( + base, + self.layout.read_completed_jobs + index * self.layout.pointer_size, + job.cast::(), + ); + } + } + unsafe { + write_at( + base, + self.layout.read_completed_jobs_count, + state.completed.len() as c_int, + ); + } + } + + unsafe fn acquire_job(&self) -> *mut IOJob_t { + let file = unsafe { base_file(self.context, self.layout) }; + let test_mode = unsafe { !self.prefs.is_null() && (*self.prefs).testMode != 0 }; + assert!(!file.is_null() || test_mode); + + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + while state.available.is_empty() { + state = self + .jobs_changed + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + let job = state.available.pop().unwrap(); + unsafe { self.sync_available_locked(&state) }; + drop(state); + + unsafe { + (*job).usedBufferSize = 0; + (*job).file = file; + (*job).offset = 0; + } + job + } + + unsafe fn release_job(&self, job: *mut IOJob_t) { + assert!(!job.is_null()); + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + debug_assert!(state.available.len() < self.total_jobs); + state.available.push(job); + unsafe { self.sync_available_locked(&state) }; + self.jobs_changed.notify_all(); + } + + unsafe fn add_completed(&self, job: *mut IOJob_t) { + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + debug_assert!(state.completed.len() < MAX_IO_JOBS); + state.completed.push(job); + unsafe { self.sync_completed_locked(&state) }; + self.jobs_changed.notify_all(); + } + + unsafe fn release_all_completed(&self) { + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + let completed = std::mem::take(&mut state.completed); + state.available.extend(completed); + unsafe { + self.sync_available_locked(&state); + self.sync_completed_locked(&state); + } + self.jobs_changed.notify_all(); + } + + unsafe fn get_next_completed(&self) -> *mut IOJob_t { + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + loop { + let waiting = unsafe { read_at::(self.context, self.layout.read_waiting_offset) }; + if let Some(index) = state + .completed + .iter() + .position(|job| unsafe { (**job).offset == waiting }) + { + let job = state.completed.swap_remove(index); + unsafe { + write_at( + self.context, + self.layout.read_waiting_offset, + waiting.wrapping_add((*job).usedBufferSize as u64), + ); + self.sync_completed_locked(&state); + } + return job; + } + + let held = usize::from( + !unsafe { read_at::<*mut c_void>(self.context, self.layout.read_current_job) } + .is_null(), + ); + let reads_in_flight = self + .total_jobs + .saturating_sub(state.available.len() + state.completed.len() + held); + if reads_in_flight == 0 { + return ptr::null_mut(); + } + state = self + .jobs_changed + .wait(state) + .unwrap_or_else(|error| error.into_inner()); + } + } + + unsafe fn enqueue_job(&self, job: *mut IOJob_t) { + if self.is_active() { + self.queue.as_ref().unwrap().enqueue(job); + } else { + unsafe { execute_job(self.kind, job) }; + } + } + + unsafe fn is_active(&self) -> bool { + let active = unsafe { read_at::(self.context, self.layout.thread_pool_active) }; + active != 0 && self.queue.is_some() + } + + unsafe fn join(&self) { + if let Some(queue) = &self.queue { + queue.join(); + } + } + + unsafe fn set_async(&self, async_mode: c_int) { + assert!(async_mode == 0 || async_mode == 1); + let active = unsafe { read_at::(self.context, self.layout.thread_pool_active) }; + if active != async_mode { + unsafe { self.join() }; + unsafe { + write_at(self.context, self.layout.thread_pool_active, async_mode); + } + } + } + + unsafe fn destroy_jobs(&self) { + unsafe { self.join() }; + let mut state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + debug_assert!(state.completed.is_empty()); + let available = std::mem::take(&mut state.available); + drop(state); + for job in available { + unsafe { + libc::free((*job).buffer); + libc::free(job.cast()); + } + } + } +} + +#[inline] +unsafe fn execute_job(kind: PoolKind, job: *mut IOJob_t) { + match kind { + PoolKind::Read => unsafe { execute_read_job(job) }, + PoolKind::Write => unsafe { execute_write_job(job) }, + } +} + +unsafe extern "C" fn read_pool_callback(opaque: *mut c_void) { + unsafe { execute_read_job(opaque.cast()) }; +} + +unsafe extern "C" fn write_pool_callback(opaque: *mut c_void) { + unsafe { execute_write_job(opaque.cast()) }; +} + +unsafe fn create_context(prefs: *const FIO_prefs_t, buffer_size: usize, kind: PoolKind) -> *mut u8 { + assert!(!prefs.is_null()); + let configured_threading = multithreading_enabled(); + let layout = abi_layout(configured_threading); + let context_size = match kind { + PoolKind::Read => layout.read_size, + PoolKind::Write => layout.write_size, + }; + let context = libc::malloc(context_size).cast::(); + if context.is_null() { + fatal(100, "Allocation error: not enough memory"); + } + unsafe { ptr::write_bytes(context, 0, context_size) }; + + let pool_exists = unsafe { (*prefs).asyncIO != 0 }; + let inner = Box::new(PoolInner::new( + context, + layout, + prefs, + kind, + pool_exists, + pool_exists && configured_threading, + )); + let inner = Box::into_raw(inner); + unsafe { + write_at(context, layout.thread_pool, inner); + write_at(context, layout.thread_pool_active, c_int::from(pool_exists)); + write_at( + context, + layout.total_io_jobs, + if pool_exists { MAX_IO_JOBS as c_int } else { 2 }, + ); + write_at(context, layout.prefs, prefs); + write_at( + context, + layout.pool_function, + match kind { + PoolKind::Read => read_pool_callback, + PoolKind::Write => write_pool_callback, + } as PoolFunction, + ); + write_at(context, layout.file, ptr::null_mut::()); + write_at(context, layout.available_jobs_count, 0 as c_int); + write_at(context, layout.job_buffer_size, buffer_size); + } + + unsafe { (&*inner).init_jobs(buffer_size) }; + + if kind == PoolKind::Write { + unsafe { write_at(context, layout.write_stored_skips, 0 as c_uint) }; + } else { + let coalesce_size = buffer_size + .checked_mul(2) + .unwrap_or_else(|| fatal(100, "Allocation error: not enough memory")); + let coalesce = libc::malloc(coalesce_size.max(1)).cast::(); + if coalesce.is_null() { + fatal(100, "Allocation error: not enough memory"); + } + unsafe { + write_at(context, layout.read_reached_eof, 0 as c_int); + write_at(context, layout.read_next_offset, 0_u64); + write_at(context, layout.read_waiting_offset, 0_u64); + write_at(context, layout.read_current_job, ptr::null_mut::()); + write_at(context, layout.read_coalesce_buffer, coalesce); + write_at(context, layout.read_src_buffer, coalesce); + write_at(context, layout.read_src_buffer_loaded, 0_usize); + write_at(context, layout.read_completed_jobs_count, 0 as c_int); + } + } + + context +} + +#[inline] +unsafe fn inner_for_job(job: *mut IOJob_t) -> &'static PoolInner { + assert!(!job.is_null()); + unsafe { base_inner((*job).ctx.cast()) } +} + +#[inline] +unsafe fn prefs_for(inner: &PoolInner) -> &FIO_prefs_t { + assert!(!inner.prefs.is_null()); + unsafe { &*inner.prefs } +} + +#[inline] +unsafe fn seek_relative(file: *mut libc::FILE, mut amount: u64) -> bool { + while amount != 0 { + let step = amount.min(SPARSE_SKIP_CHUNK); + #[cfg(unix)] + let result = unsafe { libc::fseeko(file, step as libc::off_t, libc::SEEK_CUR) }; + #[cfg(windows)] + let result = unsafe { _fseeki64(file, step as i64, libc::SEEK_CUR) }; + #[cfg(not(any(unix, windows)))] + let result = unsafe { libc::fseek(file, step as c_long, libc::SEEK_CUR) }; + if result != 0 { + return false; + } + amount -= step; + } + true +} + +unsafe fn write_exact(file: *mut libc::FILE, buffer: *const c_void, size: usize, code: c_int) { + if size != 0 && unsafe { libc::fwrite(buffer, 1, size, file) } != size { + fatal(code, "Write error"); + } +} + +unsafe fn sparse_write( + file: *mut libc::FILE, + buffer: *const c_void, + buffer_size: usize, + prefs: &FIO_prefs_t, + mut stored_skips: c_uint, +) -> c_uint { + if prefs.testMode != 0 { + return 0; + } + assert!(!file.is_null()); + + if prefs.sparseFileSupport == 0 { + unsafe { write_exact(file, buffer, buffer_size, 70) }; + return 0; + } + + if u64::from(stored_skips) > SPARSE_SKIP_CHUNK { + if !unsafe { seek_relative(file, SPARSE_SKIP_CHUNK) } { + fatal(91, "1 GB skip error (sparse file support)"); + } + stored_skips = stored_skips.wrapping_sub(SPARSE_SKIP_CHUNK as c_uint); + } + + let word_size = size_of::(); + let segment_words = SPARSE_SEGMENT_SIZE / word_size; + let word_count = buffer_size / word_size; + let words = buffer.cast::(); + let mut processed_words = 0; + let mut remaining_words = word_count; + + while remaining_words != 0 { + let segment = remaining_words.min(segment_words); + let mut leading = 0; + while leading < segment && unsafe { words.add(processed_words + leading).read() } == 0 { + leading += 1; + } + stored_skips = stored_skips.wrapping_add((leading * word_size) as c_uint); + + if leading != segment { + let nonzero_words = segment - leading; + if !unsafe { seek_relative(file, u64::from(stored_skips)) } { + fatal(92, "Sparse skip error; try --no-sparse"); + } + stored_skips = 0; + let write_ptr = unsafe { words.add(processed_words + leading).cast::() }; + if unsafe { libc::fwrite(write_ptr, word_size, nonzero_words, file) } != nonzero_words { + fatal(93, "Write error: cannot write block"); + } + } + + processed_words += segment; + remaining_words -= segment; + } + + let remainder = buffer_size & (word_size - 1); + if remainder != 0 { + let rest_start = unsafe { buffer.cast::().add(word_count * word_size) }; + let mut leading = 0; + while leading < remainder && unsafe { rest_start.add(leading).read() } == 0 { + leading += 1; + } + stored_skips = stored_skips.wrapping_add(leading as c_uint); + if leading != remainder { + if !unsafe { seek_relative(file, u64::from(stored_skips)) } { + fatal(92, "Sparse skip error; try --no-sparse"); + } + let rest = unsafe { rest_start.add(leading) }; + unsafe { write_exact(file, rest.cast(), remainder - leading, 95) }; + stored_skips = 0; + } + } + + stored_skips +} + +unsafe fn sparse_write_end(file: *mut libc::FILE, prefs: &FIO_prefs_t, stored_skips: c_uint) { + if prefs.testMode != 0 { + debug_assert_eq!(stored_skips, 0); + return; + } + if stored_skips == 0 { + return; + } + assert!(!file.is_null()); + if !unsafe { seek_relative(file, u64::from(stored_skips - 1)) } { + fatal(69, "Final skip error (sparse file support)"); + } + let zero = [0_u8; 1]; + unsafe { write_exact(file, zero.as_ptr().cast(), 1, 69) }; +} + +unsafe fn execute_write_job(job: *mut IOJob_t) { + let inner = unsafe { inner_for_job(job) }; + let stored = unsafe { read_at::(inner.context, inner.layout.write_stored_skips) }; + let prefs = unsafe { prefs_for(inner) }; + let new_stored = unsafe { + sparse_write( + (*job).file, + (*job).buffer, + (*job).usedBufferSize, + prefs, + stored, + ) + }; + unsafe { + write_at(inner.context, inner.layout.write_stored_skips, new_stored); + inner.release_job(job); + } +} + +unsafe fn execute_read_job(job: *mut IOJob_t) { + let inner = unsafe { inner_for_job(job) }; + let reached_eof = unsafe { read_at::(inner.context, inner.layout.read_reached_eof) }; + if reached_eof != 0 { + unsafe { + (*job).usedBufferSize = 0; + inner.add_completed(job); + } + return; + } + + let file = unsafe { (*job).file }; + let size = unsafe { (*job).bufferSize }; + let read = if file.is_null() || size == 0 { + 0 + } else { + unsafe { libc::fread((*job).buffer, 1, size, file) } + }; + unsafe { (*job).usedBufferSize = read }; + if read < size { + if !file.is_null() && unsafe { libc::ferror(file) } != 0 { + fatal(37, "Read error"); + } else if !file.is_null() && unsafe { libc::feof(file) } != 0 { + unsafe { write_at(inner.context, inner.layout.read_reached_eof, 1 as c_int) }; + } else if !file.is_null() && size != 0 { + fatal(37, "Unexpected short read"); + } else { + unsafe { write_at(inner.context, inner.layout.read_reached_eof, 1 as c_int) }; + } + } + unsafe { inner.add_completed(job) }; +} + +unsafe fn read_enqueue(inner: &PoolInner) { + let job = unsafe { inner.acquire_job() }; + let next = unsafe { read_at::(inner.context, inner.layout.read_next_offset) }; + unsafe { + (*job).offset = next; + write_at( + inner.context, + inner.layout.read_next_offset, + next.wrapping_add((*job).bufferSize as u64), + ); + inner.enqueue_job(job); + } +} + +unsafe fn read_start(inner: &PoolInner) { + while unsafe { read_at::(inner.context, inner.layout.available_jobs_count) } > 0 { + unsafe { read_enqueue(inner) }; + } +} + +unsafe fn read_release_current_and_get_next(inner: &PoolInner) -> *mut IOJob_t { + let current = unsafe { read_at::<*mut IOJob_t>(inner.context, inner.layout.read_current_job) }; + if !current.is_null() { + unsafe { + inner.release_job(current); + write_at( + inner.context, + inner.layout.read_current_job, + ptr::null_mut::(), + ); + read_enqueue(inner); + } + } + let next = unsafe { inner.get_next_completed() }; + unsafe { + write_at(inner.context, inner.layout.read_current_job, next); + } + next +} + +#[no_mangle] +pub extern "C" fn AIO_supported() -> c_int { + c_int::from(multithreading_enabled()) +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_releaseIoJob(job: *mut IOJob_t) { + assert!(!job.is_null()); + let inner = unsafe { inner_for_job(job) }; + unsafe { inner.release_job(job) }; +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_acquireJob(ctx: *mut WritePoolCtx_t) -> *mut IOJob_t { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + unsafe { inner.acquire_job() } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_enqueueAndReacquireWriteJob(job: *mut *mut IOJob_t) { + assert!(!job.is_null()); + let queued = unsafe { *job }; + assert!(!queued.is_null()); + let inner = unsafe { inner_for_job(queued) }; + unsafe { inner.enqueue_job(queued) }; + unsafe { *job = inner.acquire_job() }; +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_sparseWriteEnd(ctx: *mut WritePoolCtx_t) { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + unsafe { inner.join() }; + let stored = unsafe { read_at::(inner.context, inner.layout.write_stored_skips) }; + let prefs = unsafe { prefs_for(inner) }; + let file = unsafe { base_file(inner.context, inner.layout) }; + unsafe { sparse_write_end(file, prefs, stored) }; + unsafe { + write_at(inner.context, inner.layout.write_stored_skips, 0 as c_uint); + } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_setFile(ctx: *mut WritePoolCtx_t, file: *mut libc::FILE) { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + unsafe { inner.join() }; + debug_assert!(unsafe { inner.all_jobs_available() }); + debug_assert_eq!( + unsafe { read_at::(inner.context, inner.layout.write_stored_skips) }, + 0 + ); + unsafe { write_at(inner.context, inner.layout.file, file) }; +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_getFile(ctx: *const WritePoolCtx_t) -> *mut libc::FILE { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast_mut().cast()) }; + unsafe { base_file(inner.context, inner.layout) } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_closeFile(ctx: *mut WritePoolCtx_t) -> c_int { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + let file = unsafe { base_file(inner.context, inner.layout) }; + unsafe { AIO_WritePool_sparseWriteEnd(ctx) }; + unsafe { + write_at( + inner.context, + inner.layout.file, + ptr::null_mut::(), + ) + }; + if file.is_null() { + return -1; + } + unsafe { libc::fclose(file) } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_create( + prefs: *const FIO_prefs_t, + buffer_size: usize, +) -> *mut WritePoolCtx_t { + unsafe { create_context(prefs, buffer_size, PoolKind::Write).cast() } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_free(ctx: *mut WritePoolCtx_t) { + if ctx.is_null() { + return; + } + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let file = unsafe { base_file(context, inner.layout) }; + if !file.is_null() { + unsafe { AIO_WritePool_closeFile(ctx) }; + } + let inner_ptr = unsafe { read_at::<*mut PoolInner>(context, inner.layout.thread_pool) }; + unsafe { (&*inner_ptr).destroy_jobs() }; + unsafe { + drop(Box::from_raw(inner_ptr)); + libc::free(context.cast()); + } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_WritePool_setAsync(ctx: *mut WritePoolCtx_t, async_mode: c_int) { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + unsafe { inner.set_async(async_mode) }; +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_create( + prefs: *const FIO_prefs_t, + buffer_size: usize, +) -> *mut ReadPoolCtx_t { + unsafe { create_context(prefs, buffer_size, PoolKind::Read).cast() } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_free(ctx: *mut ReadPoolCtx_t) { + if ctx.is_null() { + return; + } + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let file = unsafe { base_file(context, inner.layout) }; + if !file.is_null() { + unsafe { AIO_ReadPool_closeFile(ctx) }; + } else { + unsafe { inner.join() }; + unsafe { inner.release_all_completed() }; + let current = unsafe { read_at::<*mut IOJob_t>(context, inner.layout.read_current_job) }; + if !current.is_null() { + unsafe { + inner.release_job(current); + write_at( + context, + inner.layout.read_current_job, + ptr::null_mut::(), + ); + } + } + } + let coalesce = unsafe { read_at::<*mut u8>(context, inner.layout.read_coalesce_buffer) }; + let inner_ptr = unsafe { read_at::<*mut PoolInner>(context, inner.layout.thread_pool) }; + unsafe { (&*inner_ptr).destroy_jobs() }; + unsafe { + libc::free(coalesce.cast()); + drop(Box::from_raw(inner_ptr)); + libc::free(context.cast()); + } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_setAsync(ctx: *mut ReadPoolCtx_t, async_mode: c_int) { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + unsafe { inner.set_async(async_mode) }; +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_consumeBytes(ctx: *mut ReadPoolCtx_t, n: usize) { + assert!(!ctx.is_null()); + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let loaded = unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }; + assert!(n <= loaded); + unsafe { + write_at(context, inner.layout.read_src_buffer_loaded, loaded - n); + if n != 0 { + let src = read_at::<*mut u8>(context, inner.layout.read_src_buffer); + write_at(context, inner.layout.read_src_buffer, src.add(n)); + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_fillBuffer(ctx: *mut ReadPoolCtx_t, mut n: usize) -> usize { + assert!(!ctx.is_null()); + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let buffer_size = unsafe { read_at::(context, inner.layout.job_buffer_size) }; + n = n.min(buffer_size); + + let loaded = unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }; + if loaded >= n { + return 0; + } + + let use_coalesce = loaded != 0; + if use_coalesce { + let src = unsafe { read_at::<*mut u8>(context, inner.layout.read_src_buffer) }; + let coalesce = unsafe { read_at::<*mut u8>(context, inner.layout.read_coalesce_buffer) }; + unsafe { ptr::copy(src, coalesce, loaded) }; + unsafe { write_at(context, inner.layout.read_src_buffer, coalesce) }; + } + + let job = unsafe { read_release_current_and_get_next(inner) }; + if job.is_null() { + return 0; + } + let used = unsafe { (*job).usedBufferSize }; + if use_coalesce { + let coalesce = unsafe { read_at::<*mut u8>(context, inner.layout.read_coalesce_buffer) }; + unsafe { + ptr::copy_nonoverlapping((*job).buffer.cast::(), coalesce.add(loaded), used); + write_at(context, inner.layout.read_src_buffer_loaded, loaded + used); + write_at(context, inner.layout.read_src_buffer, coalesce); + } + } else { + unsafe { + write_at( + context, + inner.layout.read_src_buffer, + (*job).buffer.cast::(), + ); + write_at(context, inner.layout.read_src_buffer_loaded, used); + } + } + used +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_consumeAndRefill(ctx: *mut ReadPoolCtx_t) -> usize { + assert!(!ctx.is_null()); + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let loaded = unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }; + unsafe { AIO_ReadPool_consumeBytes(ctx, loaded) }; + unsafe { AIO_ReadPool_fillBuffer(ctx, read_at(context, inner.layout.job_buffer_size)) } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_setFile(ctx: *mut ReadPoolCtx_t, file: *mut libc::FILE) { + assert!(!ctx.is_null()); + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + unsafe { inner.join() }; + unsafe { inner.release_all_completed() }; + let current = unsafe { read_at::<*mut IOJob_t>(context, inner.layout.read_current_job) }; + if !current.is_null() { + unsafe { + inner.release_job(current); + write_at( + context, + inner.layout.read_current_job, + ptr::null_mut::(), + ); + } + } + debug_assert!(unsafe { inner.all_jobs_available() }); + unsafe { + write_at(context, inner.layout.file, file); + write_at(context, inner.layout.read_next_offset, 0_u64); + write_at(context, inner.layout.read_waiting_offset, 0_u64); + write_at(context, inner.layout.read_reached_eof, 0 as c_int); + let coalesce = read_at::<*mut u8>(context, inner.layout.read_coalesce_buffer); + write_at(context, inner.layout.read_src_buffer, coalesce); + write_at(context, inner.layout.read_src_buffer_loaded, 0_usize); + } + if !file.is_null() { + unsafe { read_start(inner) }; + } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_getFile(ctx: *const ReadPoolCtx_t) -> *mut libc::FILE { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast_mut().cast()) }; + unsafe { base_file(inner.context, inner.layout) } +} + +#[no_mangle] +pub unsafe extern "C" fn AIO_ReadPool_closeFile(ctx: *mut ReadPoolCtx_t) -> c_int { + assert!(!ctx.is_null()); + let inner = unsafe { base_inner(ctx.cast()) }; + let file = unsafe { base_file(inner.context, inner.layout) }; + unsafe { AIO_ReadPool_setFile(ctx, ptr::null_mut()) }; + if file.is_null() { + return -1; + } + unsafe { libc::fclose(file) } +} + +impl PoolInner { + unsafe fn all_jobs_available(&self) -> bool { + let state = self.jobs.lock().unwrap_or_else(|error| error.into_inner()); + state.available.len() == self.total_jobs && state.completed.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_prefs(async_io: c_int) -> FIO_prefs_t { + let mut prefs: FIO_prefs_t = unsafe { std::mem::zeroed() }; + prefs.asyncIO = async_io; + prefs.testMode = 1; + prefs.sparseFileSupport = 1; + prefs + } + + #[repr(C)] + struct CBase { + thread_pool: *mut c_void, + thread_pool_active: c_int, + total_io_jobs: c_int, + prefs: *const FIO_prefs_t, + pool_function: PoolFunction, + file: *mut libc::FILE, + io_jobs_mutex: M, + available_jobs: [*mut c_void; MAX_IO_JOBS], + available_jobs_count: c_int, + job_buffer_size: usize, + } + + #[repr(C)] + struct CRead { + base: CBase, + reached_eof: c_int, + next_read_offset: u64, + waiting_on_offset: u64, + current_job_held: *mut c_void, + coalesce_buffer: *mut u8, + src_buffer: *mut u8, + src_buffer_loaded: usize, + completed_jobs: [*mut c_void; MAX_IO_JOBS], + completed_jobs_count: c_int, + job_completed_cond: C, + } + + #[repr(C)] + struct CWrite { + base: CBase, + stored_skips: c_uint, + } + + #[repr(C)] + struct CJob { + ctx: *mut c_void, + file: *mut libc::FILE, + buffer: *mut c_void, + buffer_size: usize, + used_buffer_size: usize, + offset: u64, + } + + fn assert_base_offsets(layout: AbiLayout) { + assert_eq!( + std::mem::offset_of!(CBase, thread_pool), + layout.thread_pool + ); + assert_eq!( + std::mem::offset_of!(CBase, thread_pool_active), + layout.thread_pool_active + ); + assert_eq!( + std::mem::offset_of!(CBase, total_io_jobs), + layout.total_io_jobs + ); + assert_eq!(std::mem::offset_of!(CBase, prefs), layout.prefs); + assert_eq!( + std::mem::offset_of!(CBase, pool_function), + layout.pool_function + ); + assert_eq!(std::mem::offset_of!(CBase, file), layout.file); + assert_eq!( + std::mem::offset_of!(CBase, io_jobs_mutex), + layout.io_jobs_mutex + ); + assert_eq!( + std::mem::offset_of!(CBase, available_jobs), + layout.available_jobs + ); + assert_eq!( + std::mem::offset_of!(CBase, available_jobs_count), + layout.available_jobs_count + ); + assert_eq!( + std::mem::offset_of!(CBase, job_buffer_size), + layout.job_buffer_size + ); + } + + fn assert_read_offsets(layout: AbiLayout) { + assert_eq!( + std::mem::offset_of!(CRead, reached_eof), + layout.read_reached_eof + ); + assert_eq!( + std::mem::offset_of!(CRead, next_read_offset), + layout.read_next_offset + ); + assert_eq!( + std::mem::offset_of!(CRead, waiting_on_offset), + layout.read_waiting_offset + ); + assert_eq!( + std::mem::offset_of!(CRead, current_job_held), + layout.read_current_job + ); + assert_eq!( + std::mem::offset_of!(CRead, coalesce_buffer), + layout.read_coalesce_buffer + ); + assert_eq!( + std::mem::offset_of!(CRead, src_buffer), + layout.read_src_buffer + ); + assert_eq!( + std::mem::offset_of!(CRead, src_buffer_loaded), + layout.read_src_buffer_loaded + ); + assert_eq!( + std::mem::offset_of!(CRead, completed_jobs), + layout.read_completed_jobs + ); + assert_eq!( + std::mem::offset_of!(CRead, completed_jobs_count), + layout.read_completed_jobs_count + ); + assert_eq!( + std::mem::offset_of!(CRead, job_completed_cond), + layout.read_job_completed_cond + ); + } + + #[test] + fn calculated_context_layout_matches_the_c_structs() { + let non_threaded = abi_layout(false); + assert_base_offsets::(non_threaded); + assert_read_offsets::(non_threaded); + assert_eq!(non_threaded.read_reached_eof, size_of::>()); + assert_eq!(non_threaded.read_size, size_of::>()); + assert_eq!(non_threaded.write_size, size_of::>()); + + #[cfg(unix)] + { + let threaded = abi_layout(true); + #[cfg(feature = "debug-pthread")] + { + type CThreadedBase = CBase<*mut libc::pthread_mutex_t>; + type CThreadedRead = CRead<*mut libc::pthread_mutex_t, *mut libc::pthread_cond_t>; + type CThreadedWrite = CWrite<*mut libc::pthread_mutex_t>; + assert_base_offsets::<*mut libc::pthread_mutex_t>(threaded); + assert_read_offsets::<*mut libc::pthread_mutex_t, *mut libc::pthread_cond_t>( + threaded, + ); + assert_eq!(threaded.read_reached_eof, size_of::()); + assert_eq!(threaded.read_size, size_of::()); + assert_eq!(threaded.write_size, size_of::()); + } + #[cfg(not(feature = "debug-pthread"))] + { + type CThreadedBase = CBase; + type CThreadedRead = CRead; + type CThreadedWrite = CWrite; + assert_base_offsets::(threaded); + assert_read_offsets::(threaded); + assert_eq!(threaded.read_reached_eof, size_of::()); + assert_eq!(threaded.read_size, size_of::()); + assert_eq!(threaded.write_size, size_of::()); + } + } + + assert_eq!(size_of::(), size_of::()); + assert_eq!( + std::mem::offset_of!(IOJob_t, ctx), + std::mem::offset_of!(CJob, ctx) + ); + assert_eq!( + std::mem::offset_of!(IOJob_t, file), + std::mem::offset_of!(CJob, file) + ); + assert_eq!( + std::mem::offset_of!(IOJob_t, buffer), + std::mem::offset_of!(CJob, buffer) + ); + assert_eq!( + std::mem::offset_of!(IOJob_t, bufferSize), + std::mem::offset_of!(CJob, buffer_size) + ); + assert_eq!( + std::mem::offset_of!(IOJob_t, usedBufferSize), + std::mem::offset_of!(CJob, used_buffer_size) + ); + assert_eq!( + std::mem::offset_of!(IOJob_t, offset), + std::mem::offset_of!(CJob, offset) + ); + } + + #[test] + fn non_threaded_pool_starts_with_two_available_jobs() { + let prefs = test_prefs(0); + let ctx = unsafe { AIO_WritePool_create(&prefs, 32) }; + let base = ctx.cast::(); + let inner = unsafe { base_inner(base) }; + assert!(inner.queue.is_none()); + assert_eq!( + unsafe { read_at::(base, inner.layout.total_io_jobs) }, + 2 + ); + assert_eq!( + unsafe { read_at::(base, inner.layout.available_jobs_count) }, + 2 + ); + + let job = unsafe { AIO_WritePool_acquireJob(ctx) }; + assert_eq!( + unsafe { read_at::(base, inner.layout.available_jobs_count) }, + 1 + ); + unsafe { AIO_WritePool_releaseIoJob(job) }; + assert_eq!( + unsafe { read_at::(base, inner.layout.available_jobs_count) }, + 2 + ); + unsafe { AIO_WritePool_free(ctx) }; + } + + #[test] + fn async_pool_toggle_keeps_jobs_owned_by_the_pool() { + let prefs = test_prefs(1); + let ctx = unsafe { AIO_WritePool_create(&prefs, 32) }; + let base = ctx.cast::(); + let inner = unsafe { base_inner(base) }; + assert!(inner.queue.is_some()); + assert_eq!( + unsafe { read_at::(base, inner.layout.available_jobs_count) }, + MAX_IO_JOBS as c_int + ); + + let mut job = unsafe { AIO_WritePool_acquireJob(ctx) }; + unsafe { (*job).usedBufferSize = 0 }; + unsafe { AIO_WritePool_enqueueAndReacquireWriteJob(&mut job) }; + assert!(!job.is_null()); + unsafe { AIO_WritePool_releaseIoJob(job) }; + unsafe { AIO_WritePool_setAsync(ctx, 0) }; + assert_eq!( + unsafe { read_at::(base, inner.layout.thread_pool_active) }, + 0 + ); + unsafe { AIO_WritePool_setAsync(ctx, 1) }; + assert_eq!( + unsafe { read_at::(base, inner.layout.thread_pool_active) }, + 1 + ); + unsafe { AIO_WritePool_free(ctx) }; + } + + #[test] + fn async_pool_shutdown_drains_queued_jobs_before_freeing_buffers() { + let prefs = test_prefs(1); + let ctx = unsafe { AIO_WritePool_create(&prefs, 32) }; + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + + for _ in 0..IO_QUEUE_SIZE { + let job = unsafe { AIO_WritePool_acquireJob(ctx) }; + unsafe { + (*job).usedBufferSize = 0; + inner.enqueue_job(job); + } + } + + // Freeing the context must join the worker before releasing any job + // buffers. The queue is intentionally still populated here. + unsafe { AIO_WritePool_free(ctx) }; + } + + #[cfg(unix)] + #[test] + fn read_pool_preserves_file_order_in_threaded_and_non_threaded_modes() { + let input = b"async file I/O keeps this order"; + for async_io in [0, 1] { + let file = unsafe { libc::tmpfile() }; + assert!(!file.is_null()); + assert_eq!( + unsafe { libc::fwrite(input.as_ptr().cast(), 1, input.len(), file) }, + input.len() + ); + assert_eq!(unsafe { libc::fflush(file) }, 0); + assert_eq!(unsafe { libc::fseek(file, 0, libc::SEEK_SET) }, 0); + + let mut prefs = test_prefs(async_io); + prefs.testMode = 0; + let ctx = unsafe { AIO_ReadPool_create(&prefs, 4) }; + unsafe { AIO_ReadPool_setFile(ctx, file) }; + + let mut output = Vec::new(); + loop { + unsafe { AIO_ReadPool_fillBuffer(ctx, 4) }; + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let loaded = + unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }; + if loaded == 0 { + break; + } + let source = unsafe { read_at::<*const u8>(context, inner.layout.read_src_buffer) }; + output.extend_from_slice(unsafe { std::slice::from_raw_parts(source, loaded) }); + unsafe { AIO_ReadPool_consumeBytes(ctx, loaded) }; + } + + assert_eq!(output, input); + assert_eq!(unsafe { AIO_ReadPool_closeFile(ctx) }, 0); + unsafe { AIO_ReadPool_free(ctx) }; + } + } + + #[test] + fn read_buffer_consumption_preserves_unread_bytes_without_a_file() { + let prefs = test_prefs(0); + let ctx = unsafe { AIO_ReadPool_create(&prefs, 32) }; + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + let coalesce = unsafe { read_at::<*mut u8>(context, inner.layout.read_coalesce_buffer) }; + unsafe { ptr::copy_nonoverlapping(b"abc".as_ptr(), coalesce, 3) }; + unsafe { + write_at(context, inner.layout.read_src_buffer, coalesce); + write_at(context, inner.layout.read_src_buffer_loaded, 3_usize); + } + + unsafe { AIO_ReadPool_consumeBytes(ctx, 1) }; + assert_eq!( + unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }, + 2 + ); + assert_eq!( + unsafe { *read_at::<*mut u8>(context, inner.layout.read_src_buffer) }, + b'b' + ); + assert_eq!(unsafe { AIO_ReadPool_fillBuffer(ctx, 3) }, 0); + assert_eq!( + unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }, + 2 + ); + unsafe { AIO_ReadPool_free(ctx) }; + } + + #[test] + fn read_set_file_none_resets_visible_stream_state() { + let prefs = test_prefs(0); + let ctx = unsafe { AIO_ReadPool_create(&prefs, 32) }; + let context = ctx.cast::(); + let inner = unsafe { base_inner(context) }; + unsafe { + write_at(context, inner.layout.read_reached_eof, 1 as c_int); + write_at(context, inner.layout.read_next_offset, 123_u64); + write_at(context, inner.layout.read_waiting_offset, 77_u64); + write_at(context, inner.layout.read_src_buffer_loaded, 11_usize); + } + unsafe { AIO_ReadPool_setFile(ctx, ptr::null_mut()) }; + assert_eq!( + unsafe { read_at::(context, inner.layout.read_reached_eof) }, + 0 + ); + assert_eq!( + unsafe { read_at::(context, inner.layout.read_next_offset) }, + 0 + ); + assert_eq!( + unsafe { read_at::(context, inner.layout.read_waiting_offset) }, + 0 + ); + assert_eq!( + unsafe { read_at::(context, inner.layout.read_src_buffer_loaded) }, + 0 + ); + unsafe { AIO_ReadPool_free(ctx) }; + } +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 5acbad404..28894462b 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -15,6 +15,7 @@ pub mod dict_builder_zdict; pub mod divsufsort; pub mod entropy_common; pub mod errors; +pub mod fileio_asyncio; #[cfg(feature = "compression")] pub mod fse_compress; pub mod fse_decompress;