Introduce zstd into cosmopolitan (#843)

third_party/zstd/programs/fileio_asyncio.h

@@ -0,0 +1,204 @@
+// clang-format off
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under both the BSD-style license (found in the
+ * LICENSE file in the root directory of this source tree) and the GPLv2 (found
+ * in the COPYING file in the root directory of this source tree).
+ * You may select, at your option, one of the above-listed licenses.
+ */
+
+ /*
+  * FileIO AsyncIO exposes read/write IO pools that allow doing IO asynchronously.
+  * Current implementation relies on having one thread that reads and one that
+  * writes.
+  * Each IO pool supports up to `MAX_IO_JOBS` that can be enqueued for work, but
+  * are performed serially by the appropriate worker thread.
+  * Most systems exposes better primitives to perform asynchronous IO, such as
+  * io_uring on newer linux systems. The API is built in such a way that in the
+  * future we could replace the threads with better solutions when available.
+  */
+
+#ifndef ZSTD_FILEIO_ASYNCIO_H
+#define ZSTD_FILEIO_ASYNCIO_H
+
+#if defined (__cplusplus)
+extern "C" {
+#endif
+
+#include "third_party/zstd/lib/common/mem.h"     /* U32, U64 */
+#include "third_party/zstd/programs/fileio_types.h"
+#include "third_party/zstd/programs/platform.h"
+#include "third_party/zstd/programs/util.h"
+#include "third_party/zstd/lib/common/pool.h"
+#include "third_party/zstd/lib/common/threading.h"
+
+#define MAX_IO_JOBS          (10)
+
+typedef struct {
+    /* These struct fields should be set only on creation and not changed afterwards */
+    POOL_ctx* threadPool;
+    int threadPoolActive;
+    int totalIoJobs;
+    const FIO_prefs_t* prefs;
+    POOL_function poolFunction;
+
+    /* Controls the file we currently write to, make changes only by using provided utility functions */
+    FILE* file;
+
+    /* The jobs and availableJobsCount fields are accessed by both the main and worker threads and should
+     * only be mutated after locking the mutex */
+    ZSTD_pthread_mutex_t ioJobsMutex;
+    void* availableJobs[MAX_IO_JOBS];
+    int availableJobsCount;
+    size_t jobBufferSize;
+} IOPoolCtx_t;
+
+typedef struct {
+    IOPoolCtx_t base;
+
+    /* State regarding the currently read file */
+    int reachedEof;
+    U64 nextReadOffset;
+    U64 waitingOnOffset;
+
+    /* We may hold an IOJob object as needed if we actively expose its buffer. */
+    void *currentJobHeld;
+
+    /* Coalesce buffer is used to join two buffers in case where we need to read more bytes than left in
+     * the first of them. Shouldn't be accessed from outside ot utility functions. */
+    U8 *coalesceBuffer;
+
+    /* Read buffer can be used by consumer code, take care when copying this pointer aside as it might
+     * change when consuming / refilling buffer. */
+    U8 *srcBuffer;
+    size_t srcBufferLoaded;
+
+    /* We need to know what tasks completed so we can use their buffers when their time comes.
+     * Should only be accessed after locking base.ioJobsMutex . */
+    void* completedJobs[MAX_IO_JOBS];
+    int completedJobsCount;
+    ZSTD_pthread_cond_t jobCompletedCond;
+} ReadPoolCtx_t;
+
+typedef struct {
+    IOPoolCtx_t base;
+    unsigned storedSkips;
+} WritePoolCtx_t;
+
+typedef struct {
+    /* These fields are automatically set and shouldn't be changed by non WritePool code. */
+    void *ctx;
+    FILE* file;
+    void *buffer;
+    size_t bufferSize;
+
+    /* This field should be changed before a job is queued for execution and should contain the number
+     * of bytes to write from the buffer. */
+    size_t usedBufferSize;
+    U64 offset;
+} IOJob_t;
+
+/* AIO_supported:
+ * Returns 1 if AsyncIO is supported on the system, 0 otherwise. */
+int AIO_supported(void);
+
+
+/* AIO_WritePool_releaseIoJob:
+ * Releases an acquired job back to the pool. Doesn't execute the job. */
+void AIO_WritePool_releaseIoJob(IOJob_t *job);
+
+/* AIO_WritePool_acquireJob:
+ * Returns an available write job to be used for a future write. */
+IOJob_t* AIO_WritePool_acquireJob(WritePoolCtx_t *ctx);
+
+/* AIO_WritePool_enqueueAndReacquireWriteJob:
+ * Enqueues 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_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);
+
+/* 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_WritePool_getFile:
+ * Returns the file the writePool is currently set to write to. */
+FILE* AIO_WritePool_getFile(const WritePoolCtx_t* ctx);
+
+/* 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);
+
+/* AIO_WritePool_create:
+ * Allocates and sets and a new write pool including its included jobs.
+ * bufferSize should be set to the maximal buffer we want to write to at a time. */
+WritePoolCtx_t* AIO_WritePool_create(const FIO_prefs_t* prefs, size_t bufferSize);
+
+/* AIO_WritePool_free:
+ * Frees and releases a writePool and its resources. Closes destination file. */
+void AIO_WritePool_free(WritePoolCtx_t* 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_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);
+
+/* AIO_ReadPool_free:
+ * Frees and releases a readPool and its resources. Closes source file. */
+void AIO_ReadPool_free(ReadPoolCtx_t* ctx);
+
+/* 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_ReadPool_consumeBytes:
+ * Consumes byes from srcBuffer's beginning and updates srcBufferLoaded accordingly. */
+void AIO_ReadPool_consumeBytes(ReadPoolCtx_t *ctx, size_t n);
+
+/* AIO_ReadPool_fillBuffer:
+ * Makes sure buffer has at least n bytes loaded (as long as n is not bigger than the initialized bufferSize).
+ * Returns if srcBuffer has at least n 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 bufferSize bytes. */
+size_t AIO_ReadPool_fillBuffer(ReadPoolCtx_t *ctx, size_t n);
+
+/* AIO_ReadPool_consumeAndRefill:
+ * Consumes the current buffer and refills it with bufferSize bytes. */
+size_t AIO_ReadPool_consumeAndRefill(ReadPoolCtx_t *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);
+
+/* AIO_ReadPool_getFile:
+ * Returns the current file set for the read pool. */
+FILE* AIO_ReadPool_getFile(const ReadPoolCtx_t *ctx);
+
+/* 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);
+
+#if defined (__cplusplus)
+}
+#endif
+
+#endif /* ZSTD_FILEIO_ASYNCIO_H */