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

@@ -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)

programs/fileio_asyncio.c

@@ -8,656 +8,10 @@
  * You may select, at your option, one of the above-listed licenses.
  */
 
-#include "platform.h"
-#include <stdio.h>      /* fprintf, open, fdopen, fread, _fileno, stdin, stdout */
-#include <stdlib.h>     /* malloc, free */
-#include <assert.h>
-#include <errno.h>      /* errno */
-
-#if defined (_MSC_VER)
-#  include <sys/stat.h>
-#  include <io.h>
-#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; i<ctx->availableJobsCount; 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; i<ctx->completedJobsCount; 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; i<ctx->completedJobsCount; 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);
-}

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.

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::<c_int>();
+    let int_align = align_of::<c_int>();
+    let word_size = size_of::<usize>();
+    let word_align = align_of::<usize>();
+    let function_size = size_of::<PoolFunction>();
+    let function_align = align_of::<PoolFunction>();
+
+    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::<libc::pthread_mutex_t>(),
+                align_of::<libc::pthread_mutex_t>(),
+                size_of::<libc::pthread_cond_t>(),
+                align_of::<libc::pthread_cond_t>(),
+            )
+        }
+        #[cfg(windows)]
+        {
+            (
+                size_of::<WindowsCriticalSection>(),
+                align_of::<WindowsCriticalSection>(),
+                size_of::<WindowsConditionVariable>(),
+                align_of::<WindowsConditionVariable>(),
+            )
+        }
+        #[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::<c_uint>(), write_align);
+
+    let read_reached_eof = base_size;
+    offset = read_reached_eof + int_size;
+    offset = align_up(offset, align_of::<u64>());
+    let read_next_offset = offset;
+    offset += size_of::<u64>();
+    let read_waiting_offset = offset;
+    offset += size_of::<u64>();
+    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::<u64>());
+    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<T: Copy>(base: *const u8, offset: usize) -> T {
+    unsafe { base.add(offset).cast::<T>().read() }
+}
+
+#[inline]
+unsafe fn write_at<T: Copy>(base: *mut u8, offset: usize, value: T) {
+    unsafe { base.add(offset).cast::<T>().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<QueueState>,
+    work_available: Condvar,
+    queue_space: Condvar,
+    idle: Condvar,
+    worker: Mutex<Option<JoinHandle<()>>>,
+    threaded: bool,
+    kind: PoolKind,
+}
+
+unsafe impl Send for AsyncQueue {}
+unsafe impl Sync for AsyncQueue {}
+
+impl AsyncQueue {
+    fn new(threaded: bool, kind: PoolKind) -> Arc<Self> {
+        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<JobState>,
+    jobs_changed: Condvar,
+    queue: Option<Arc<AsyncQueue>>,
+}
+
+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::<IOJob_t>()).cast::<IOJob_t>();
+            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::<c_void>(),
+                );
+            }
+        }
+        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::<c_void>(),
+                );
+            }
+        }
+        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::<u64>(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::<c_int>(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::<c_int>(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::<u8>();
+    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::<libc::FILE>());
+        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::<u8>();
+        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::<c_void>());
+            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::<usize>();
+    let segment_words = SPARSE_SEGMENT_SIZE / word_size;
+    let word_count = buffer_size / word_size;
+    let words = buffer.cast::<usize>();
+    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::<c_void>() };
+            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::<u8>().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::<c_uint>(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::<c_int>(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::<u64>(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::<c_int>(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::<c_void>(),
+            );
+            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::<c_uint>(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::<c_uint>(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::<libc::FILE>(),
+        )
+    };
+    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::<u8>();
+    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::<u8>();
+    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::<c_void>(),
+                );
+            }
+        }
+    }
+    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::<u8>();
+    let inner = unsafe { base_inner(context) };
+    let loaded = unsafe { read_at::<usize>(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::<u8>();
+    let inner = unsafe { base_inner(context) };
+    let buffer_size = unsafe { read_at::<usize>(context, inner.layout.job_buffer_size) };
+    n = n.min(buffer_size);
+
+    let loaded = unsafe { read_at::<usize>(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::<u8>(), 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::<u8>(),
+            );
+            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::<u8>();
+    let inner = unsafe { base_inner(context) };
+    let loaded = unsafe { read_at::<usize>(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::<u8>();
+    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::<c_void>(),
+            );
+        }
+    }
+    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<M> {
+        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<M, C> {
+        base: CBase<M>,
+        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<M> {
+        base: CBase<M>,
+        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<M>(layout: AbiLayout) {
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, thread_pool),
+            layout.thread_pool
+        );
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, thread_pool_active),
+            layout.thread_pool_active
+        );
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, total_io_jobs),
+            layout.total_io_jobs
+        );
+        assert_eq!(std::mem::offset_of!(CBase<M>, prefs), layout.prefs);
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, pool_function),
+            layout.pool_function
+        );
+        assert_eq!(std::mem::offset_of!(CBase<M>, file), layout.file);
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, io_jobs_mutex),
+            layout.io_jobs_mutex
+        );
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, available_jobs),
+            layout.available_jobs
+        );
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, available_jobs_count),
+            layout.available_jobs_count
+        );
+        assert_eq!(
+            std::mem::offset_of!(CBase<M>, job_buffer_size),
+            layout.job_buffer_size
+        );
+    }
+
+    fn assert_read_offsets<M, C>(layout: AbiLayout) {
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, reached_eof),
+            layout.read_reached_eof
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, next_read_offset),
+            layout.read_next_offset
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, waiting_on_offset),
+            layout.read_waiting_offset
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, current_job_held),
+            layout.read_current_job
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, coalesce_buffer),
+            layout.read_coalesce_buffer
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, src_buffer),
+            layout.read_src_buffer
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, src_buffer_loaded),
+            layout.read_src_buffer_loaded
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, completed_jobs),
+            layout.read_completed_jobs
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, completed_jobs_count),
+            layout.read_completed_jobs_count
+        );
+        assert_eq!(
+            std::mem::offset_of!(CRead<M, C>, 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::<c_int>(non_threaded);
+        assert_read_offsets::<c_int, c_int>(non_threaded);
+        assert_eq!(non_threaded.read_reached_eof, size_of::<CBase<c_int>>());
+        assert_eq!(non_threaded.read_size, size_of::<CRead<c_int, c_int>>());
+        assert_eq!(non_threaded.write_size, size_of::<CWrite<c_int>>());
+
+        #[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::<CThreadedBase>());
+                assert_eq!(threaded.read_size, size_of::<CThreadedRead>());
+                assert_eq!(threaded.write_size, size_of::<CThreadedWrite>());
+            }
+            #[cfg(not(feature = "debug-pthread"))]
+            {
+                type CThreadedBase = CBase<libc::pthread_mutex_t>;
+                type CThreadedRead = CRead<libc::pthread_mutex_t, libc::pthread_cond_t>;
+                type CThreadedWrite = CWrite<libc::pthread_mutex_t>;
+                assert_base_offsets::<libc::pthread_mutex_t>(threaded);
+                assert_read_offsets::<libc::pthread_mutex_t, libc::pthread_cond_t>(threaded);
+                assert_eq!(threaded.read_reached_eof, size_of::<CThreadedBase>());
+                assert_eq!(threaded.read_size, size_of::<CThreadedRead>());
+                assert_eq!(threaded.write_size, size_of::<CThreadedWrite>());
+            }
+        }
+
+        assert_eq!(size_of::<IOJob_t>(), size_of::<CJob>());
+        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::<u8>();
+        let inner = unsafe { base_inner(base) };
+        assert!(inner.queue.is_none());
+        assert_eq!(
+            unsafe { read_at::<c_int>(base, inner.layout.total_io_jobs) },
+            2
+        );
+        assert_eq!(
+            unsafe { read_at::<c_int>(base, inner.layout.available_jobs_count) },
+            2
+        );
+
+        let job = unsafe { AIO_WritePool_acquireJob(ctx) };
+        assert_eq!(
+            unsafe { read_at::<c_int>(base, inner.layout.available_jobs_count) },
+            1
+        );
+        unsafe { AIO_WritePool_releaseIoJob(job) };
+        assert_eq!(
+            unsafe { read_at::<c_int>(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::<u8>();
+        let inner = unsafe { base_inner(base) };
+        assert!(inner.queue.is_some());
+        assert_eq!(
+            unsafe { read_at::<c_int>(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::<c_int>(base, inner.layout.thread_pool_active) },
+            0
+        );
+        unsafe { AIO_WritePool_setAsync(ctx, 1) };
+        assert_eq!(
+            unsafe { read_at::<c_int>(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::<u8>();
+        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::<u8>();
+                let inner = unsafe { base_inner(context) };
+                let loaded =
+                    unsafe { read_at::<usize>(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::<u8>();
+        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::<usize>(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::<usize>(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::<u8>();
+        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::<c_int>(context, inner.layout.read_reached_eof) },
+            0
+        );
+        assert_eq!(
+            unsafe { read_at::<u64>(context, inner.layout.read_next_offset) },
+            0
+        );
+        assert_eq!(
+            unsafe { read_at::<u64>(context, inner.layout.read_waiting_offset) },
+            0
+        );
+        assert_eq!(
+            unsafe { read_at::<usize>(context, inner.layout.read_src_buffer_loaded) },
+            0
+        );
+        unsafe { AIO_ReadPool_free(ctx) };
+    }
+}

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;