add io_uring

10 files changed, 249 insertions, 16 deletions

diff --git a/include/kpl.h b/include/kpl.h
index 1d53bef..7c5e209 100644
--- a/include/kpl.h
+++ b/include/kpl.h
@@ -7,6 +7,7 @@
 #define KPL_VERSION "0.0.0"
 
 // System includes
+#define _GNU_SOURCE
 #include <stdlib.h>
 #include <stdio.h>
 #include <stdint.h>
@@ -14,11 +15,13 @@
 #include <stddef.h>
 #include <string.h>
 #include <ctype.h>
+#include <errno.h>
 #include <limits.h>
 #include <pthread.h>
 #include <semaphore.h>
 #include <fcntl.h>
 #include <unistd.h>
+#include <sys/syscall.h>
 #include <sys/stat.h>
 #include <sys/ioctl.h>
 #include <sys/mman.h>
@@ -32,7 +35,7 @@
 #endif
 
 // KPL forward declarations
-typedef struct _kpl_task kpl_task;
+typedef struct _kpl_object_task kpl_object_task;
 
 typedef struct _kpl_thread kpl_thread;
 
@@ -44,4 +47,5 @@ typedef struct _kpl_env kpl_env;
 #include "kpl/object.h"
 #include "kpl/tree.h"
 #include "kpl/memory.h"
+#include "kpl/io.h"
 #include "kpl/env.h"
diff --git a/include/kpl/class.h b/include/kpl/class.h
index b1b7054..f7f9678 100644
--- a/include/kpl/class.h
+++ b/include/kpl/class.h
@@ -2,6 +2,8 @@
  * The kpl class a 64bit value with an table of functions
  */
 
+#pragma once
+
 typedef uint8_t kpl_utf8[4];
 
 typedef union {
diff --git a/include/kpl/env.h b/include/kpl/env.h
index f3ad721..7806969 100644
--- a/include/kpl/env.h
+++ b/include/kpl/env.h
@@ -12,7 +12,7 @@ typedef enum : int32_t {
 typedef struct _kpl_thread {
     int32_t thread_id; // self id
     kpl_memory_allocator memory_allocator;
-    // TODO io_uring
+    kpl_io_uring io_uring;
     // TODO jit allocator
     _Atomic ssize_t priority; // running + the total number of tasks in threads queue, lower is higher priority
     kpl_object_queue task_queue;
@@ -39,12 +39,13 @@ inline void *kpl_thread_object_allocate(kpl_thread *thread, int32_t size) {
  * kpl_env_free &env
  */
 typedef struct _kpl_env {
+    int32_t io_uring_queue_size_pow2; // the power of two to calculate the entries in the io_uring queue
     int32_t thread_count; // -1 updates this value to max possible, 0 invalid, >= 1 for specific number
     _Atomic ssize_t total_priority; // the sum of the threads priority fields, application stops on zero
     kpl_thread *threads; // 0 is main thread, >= 1 are spawned pthreads
     kpl_tree *namespace_tree, *module_tree; // *namespace_tree lookups are thread safe, *module_tree needs mutex
     pthread_mutex_t module_tree_mutex;
-} kpl_env; // kpl state, no globals
+} kpl_env; // kpl state, no globals, must be zero initialized
 
 /*
  * All objects are returned by this
@@ -76,14 +77,14 @@ kpl_exit_code kpl_env_config(kpl_env *env, int32_t argc, char *argv[]);
 kpl_exit_code kpl_env_init(kpl_env *env);
 
 /*
- * Can be run multiple times as long as new tasks are added after each run
+ * Run once, not re-enetrant
+ * TODO RUN TASK AND IO ORDERING
  */
 kpl_exit_code kpl_env_run(kpl_env *env);
 
 /*
  * Stops KPL, can be called on any task in KPL
  * Does not return just signals to all threads to stop, the return status will be from kpl_env_run
- * Start can be called again
  */
 void kpl_env_stop(kpl_env *env);
 
diff --git a/include/kpl/io.h b/include/kpl/io.h
new file mode 100644
index 0000000..f353da2
--- /dev/null
+++ b/include/kpl/io.h
@@ -0,0 +1,95 @@
+/*
+ * IO uring
+ */
+
+#pragma once
+
+typedef enum : int32_t {
+    KPL_IO_URING_QUEUE_SIZE_POW2_MIN        = 12,
+    KPL_IO_URING_QUEUE_SIZE_POW2_MAX        = 20,
+    KPL_IO_URING_QUEUE_SIZE_POW2_DEFAULT    = 15
+} kpl_io_uring_enum;
+
+/*
+ * Per thread io_uring container, do not use across threads
+ * Operations are added to the submit queue by a functions specific to that operation
+ * Once one or more operations are added, call kpl_io_uring_submit to send the operations to the kernel
+ * Only one completed operation can be read at a time by kpl_io_uring_get_completed
+ * Uses defaults that require kernel version of 6.6 or greater
+ */
+typedef struct {
+    int32_t ring_fd, ring_queue_entries; // fd for submit, number of queue slots
+    uint32_t added_to_queue, submitted_to_kernel; //added to submission queue, added with io uring enter
+    uint32_t sq_cq_size, sqes_size; // using single mmap for kernel submission and completion
+    uint32_t *sring_head, *sring_tail, *sring_mask;
+    uint32_t *cring_head, *cring_tail, *cring_mask;
+    void *sring_ptr, *cring_ptr;
+    struct io_uring_sqe *sqes;
+    struct io_uring_cqe *cqes;
+    struct io_uring_params params;
+} kpl_io_uring; // must be zero initialized
+
+/*
+ * Called by owner thread
+ */
+int32_t kpl_io_uring_init(kpl_io_uring *io_uring, int32_t queue_size_pow2);
+
+/*
+ * Be sure no operations are in queue before calling
+ * Called by owner thread
+ */
+inline void kpl_io_uring_free(kpl_io_uring *io_uring) {
+    munmap(io_uring->sring_ptr, io_uring->sq_cq_size);
+    munmap(io_uring->sqes, io_uring->sqes_size);
+}
+
+/*
+ * Reads one from cqes, will call io uring enter if none are complete and submitted_to_kernel is not zero
+ * This will wait if io uring enter is called
+ * NULL for none completed
+ * If an error occurred, the error details will be on the kpl_task_object
+ */
+kpl_object_task *kpl_io_uring_get_completed(kpl_io_uring *io_uring);
+
+/*
+ * Submits all sqes (added_to_queue has total) in one io uring eneter
+ * This does not wait
+ */
+int32_t kpl_io_uring_submit(kpl_io_uring *io_uring);
+
+/*
+ * Only call this though one of the functions below to add an operation to the queue
+ */
+struct io_uring_sqe *kpl_io_uring_op(kpl_io_uring *io_uring, kpl_object_task *task, int32_t op, int32_t fd,
+    const void *addr, uint32_t len, uint64_t offset);
+
+inline bool kpl_io_uring_openat(kpl_io_uring *io_uring, kpl_object_task *task, int32_t dir_fd,
+    const char *path, int32_t flags, mode_t mode) {
+    struct io_uring_sqe *sqe = kpl_io_uring_op(io_uring, task, IORING_OP_OPENAT, dir_fd, path, mode, 0);
+    if (sqe)
+        sqe->open_flags = (uint32_t) flags;
+    return sqe;
+}
+
+inline bool kpl_io_uring_close(kpl_io_uring *io_uring, kpl_object_task *task, int32_t fd) {
+    return kpl_io_uring_op(io_uring, task, IORING_OP_CLOSE, fd, NULL, 0, 0);
+}
+
+inline bool kpl_io_uring_read(kpl_io_uring *io_uring, kpl_object_task *task, int32_t fd, void *buffer,
+    uint32_t byte_size, uint64_t offset) {
+    return kpl_io_uring_op(io_uring, task, IORING_OP_READ, fd, buffer, byte_size, offset);
+}
+
+inline bool kpl_io_uring_write(kpl_io_uring *uring, kpl_object_task *task, int32_t fd, void *buffer,
+    uint32_t byte_size, uint64_t offset) {
+    return kpl_io_uring_op(uring, task, IORING_OP_WRITE, fd, buffer, byte_size, offset);
+}
+
+inline bool kpl_io_uring_statx(kpl_io_uring *io_uring, kpl_object_task *task, int32_t dir_fd,
+    const char *path, int32_t flags, uint32_t mask, struct statx *statx_object) {
+    struct io_uring_sqe *sqe = kpl_io_uring_op(io_uring, task, IORING_OP_STATX, dir_fd, path, mask,
+        (uint64_t) statx_object);
+    if (sqe)
+        sqe->statx_flags = (uint32_t) flags;
+    return sqe;
+}
diff --git a/include/kpl/memory.h b/include/kpl/memory.h
index 5eb6d85..b61d861 100644
--- a/include/kpl/memory.h
+++ b/include/kpl/memory.h
@@ -9,7 +9,6 @@ typedef enum : int32_t {
     KPL_MEMORY_ALLOCATOR_SLAB_COUNT             = KPL_MEMORY_ALLOCATOR_MAX_SIZE_POW2 + 1
 } kpl_memory_allocator_enum;
 
-
 /*
  * Each slab is a linked list allocated with calloc and free
  * *buffer is allocated with mmap
diff --git a/include/kpl/object.h b/include/kpl/object.h
index 751088d..1f3ec04 100644
--- a/include/kpl/object.h
+++ b/include/kpl/object.h
@@ -48,9 +48,18 @@ inline void kpl_object_queue_enqueue(kpl_object_queue *object_queue, void *objec
  */
 void *kpl_object_queue_dequeue(kpl_object_queue *object_queue);
 
-typedef void kpl_task_fn(kpl_task *task, kpl_env *env);
+typedef void kpl_task_fn(kpl_object_task *task, kpl_env *env);
 
-typedef struct _kpl_task {
-    KPL_OBJECT_HEADER(_kpl_task);
-    // TODO
-} kpl_task;
+/*
+ * Coroutine state object
+ */
+typedef struct _kpl_object_task {
+    KPL_OBJECT_HEADER(_kpl_object_task);
+    _Atomic bool join_ready __attribute__((aligned(4))); // shares cache line with variables_size
+    int32_t variables_size, io_res; // variables_size is set once then only read
+    uint32_t io_flags; // io_res and io_flags are set by io_uring complete
+    kpl_object_task *join;
+    kpl_task_fn *task_fn;
+    kpl_result result;
+    kpl_class variables[]; // a task has no distinction between an argument or a local, both live here
+} kpl_object_task;
diff --git a/include/kpl_test.h b/include/kpl_test.h
index 70a99b2..6a423eb 100644
--- a/include/kpl_test.h
+++ b/include/kpl_test.h
@@ -29,21 +29,23 @@ typedef struct _kpl_test {
 
 void _kpl_test_add(kpl_test *_test); // do not call outside KPL_TEST macro
 
-#define _KPL_TEST_FN_DEFINITION(NAME) static void NAME(__attribute__((unused)) kpl_test *_test)
+#define _KPL_TEST_FN_NAME(NAME) kpl_test_ ## NAME
+
+#define _KPL_TEST_FN_DEFINITION(NAME) static void _KPL_TEST_FN_NAME(NAME)(__attribute__((unused)) kpl_test *_test)
 
 /*
  * Each test is a static function added to the test list and tree by a static gnu constructor function
  * The NAME becomes the function name _NAME becomes the constructor name, only one name can be used per file
  */
 #define KPL_TEST(NAME) _KPL_TEST_FN_DEFINITION(NAME); \
-static void __attribute__((constructor)) _ ## NAME(void) { \
+static void __attribute__((constructor)) kpl_test_constructor_ ## NAME(void) { \
     static kpl_test _test = { \
         .tree_left = NULL, .tree_right = NULL, .tree_duplicates = NULL, \
         .tree_height = KPL_TREE_TREE_HEIGHT_INIT, \
         .error_line_number = KPL_TEST_ERROR_LINE_NUMBER_INIT, \
         .list_next = NULL, \
         .source_file = __FILE__, .name = #NAME, .error_message = NULL, \
-        .test_fn = NAME \
+        .test_fn = _KPL_TEST_FN_NAME(NAME) \
     }; \
     _kpl_test_add(&_test); \
 } \
diff --git a/source/env.c b/source/env.c
index 587918e..4e4f463 100644
--- a/source/env.c
+++ b/source/env.c
@@ -1,6 +1,6 @@
 
 #include "kpl.h"
 
-inline void *kpl_thread_object_allocate(kpl_thread *thread, int32_t size);
+extern inline void *kpl_thread_object_allocate(kpl_thread *thread, int32_t size);
 
-inline void kpl_env_object_return(kpl_env *env, void *object);
+extern inline void kpl_env_object_return(kpl_env *env, void *object);
diff --git a/source/io.c b/source/io.c
new file mode 100644
index 0000000..0cd8480
--- /dev/null
+++ b/source/io.c
@@ -0,0 +1,101 @@
+
+#include "kpl.h"
+
+static void *kpl_io_uring_mmap(uint32_t size, int32_t fd, off_t offset) {
+    return mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, offset);
+}
+
+int32_t kpl_io_uring_init(kpl_io_uring *io_uring, int32_t queue_size_pow2) {
+    io_uring->params.flags |= IORING_SETUP_NO_SQARRAY;
+    io_uring->ring_queue_entries = 1 << queue_size_pow2;
+    io_uring->ring_fd = syscall(__NR_io_uring_setup, io_uring->ring_queue_entries, &io_uring->params);
+    if (io_uring->ring_fd < 0)
+        return -errno;
+    uint32_t ssize = io_uring->params.sq_off.array + io_uring->params.sq_entries * sizeof(uint32_t);
+    uint32_t csize = io_uring->params.cq_off.cqes + io_uring->params.cq_entries * sizeof(struct io_uring_cqe);
+    io_uring->sq_cq_size = MAX(ssize, csize);
+    io_uring->sqes_size = io_uring->params.sq_entries * sizeof(struct io_uring_sqe);
+    void *sring_ptr, *cring_ptr;
+    sring_ptr = cring_ptr = kpl_io_uring_mmap(io_uring->sq_cq_size, io_uring->ring_fd, IORING_OFF_SQ_RING);
+    io_uring->sqes = kpl_io_uring_mmap(io_uring->sqes_size, io_uring->ring_fd, IORING_OFF_SQES);
+    io_uring->sring_head = sring_ptr + io_uring->params.sq_off.head;
+    io_uring->sring_tail = sring_ptr + io_uring->params.sq_off.tail;
+    io_uring->sring_mask = sring_ptr + io_uring->params.sq_off.ring_mask;
+    io_uring->cring_head = cring_ptr + io_uring->params.cq_off.head;
+    io_uring->cring_tail = cring_ptr + io_uring->params.cq_off.tail;
+    io_uring->cring_mask = cring_ptr + io_uring->params.cq_off.ring_mask;
+    io_uring->cqes = cring_ptr + io_uring->params.cq_off.cqes;
+    io_uring->sring_ptr = sring_ptr;
+    io_uring->cring_ptr = cring_ptr;
+    return 0;
+}
+
+extern inline void kpl_io_uring_free(kpl_io_uring *io_uring);
+
+static int32_t kpl_io_uring_enter(int32_t ring_fd, uint32_t to_submit, uint32_t to_wait, uint32_t flags) {
+    int32_t ret = syscall(__NR_io_uring_enter, ring_fd, to_submit, to_wait, flags, NULL, 0);
+    return ret < 0 ? -errno : ret;
+}
+
+kpl_object_task *kpl_io_uring_get_completed(kpl_io_uring *io_uring) {
+    if (!io_uring->submitted_to_kernel)
+        return NULL;
+    uint32_t cring_head = __atomic_load_n(io_uring->cring_head, __ATOMIC_ACQUIRE);
+    if (cring_head == *io_uring->cring_tail) {
+        if (kpl_io_uring_enter(io_uring->ring_fd, 0, 1, IORING_ENTER_GETEVENTS) < 0)
+            return NULL;
+    }
+    struct io_uring_cqe *cqe = &io_uring->cqes[cring_head & *io_uring->cring_mask];
+    kpl_object_task *task = (kpl_object_task*) cqe->user_data;
+    task->io_res = cqe->res;
+    task->io_flags = cqe->flags;
+    cring_head++;
+    __atomic_store_n(io_uring->cring_head, cring_head, __ATOMIC_RELEASE);
+    io_uring->submitted_to_kernel--;
+    return task;
+}
+
+int32_t kpl_io_uring_submit(kpl_io_uring *io_uring) {
+    if (!io_uring->added_to_queue)
+        return -1;
+    int32_t ret = kpl_io_uring_enter(io_uring->ring_fd, io_uring->added_to_queue, 0, 0);
+    if (ret > 0) {
+        io_uring->added_to_queue -= ret;
+        io_uring->submitted_to_kernel += ret;
+    }
+    return ret;
+}
+
+struct io_uring_sqe *kpl_io_uring_op(kpl_io_uring *io_uring, kpl_object_task *task, int32_t op, int32_t fd,
+    const void *addr, uint32_t len, uint64_t offset) {
+    uint32_t sring_head = __atomic_load_n(io_uring->sring_head, __ATOMIC_ACQUIRE);
+    uint32_t sring_tail = *io_uring->sring_tail;
+    if (sring_tail - sring_head >= (uint32_t) io_uring->ring_queue_entries)
+        return NULL;
+    struct io_uring_sqe *sqe = &io_uring->sqes[sring_tail & *io_uring->sring_mask];
+    sring_tail++;
+    __atomic_store_n(io_uring->sring_tail, sring_tail, __ATOMIC_RELEASE);
+    io_uring->added_to_queue++;
+    memset(sqe, 0, sizeof(struct io_uring_sqe));
+    sqe->user_data = (uint64_t) task;
+    sqe->opcode = op;
+    sqe->fd = fd;
+    sqe->off = offset;
+    sqe->addr = (uint64_t) addr;
+    sqe->len = len;
+    return sqe;
+}
+
+extern inline bool kpl_io_uring_openat(kpl_io_uring *io_uring, kpl_object_task *task, int32_t dir_fd,
+    const char *path, int32_t flags, mode_t mode);
+
+extern inline bool kpl_io_uring_close(kpl_io_uring *io_uring, kpl_object_task *task, int32_t fd);
+
+extern inline bool kpl_io_uring_read(kpl_io_uring *io_uring, kpl_object_task *task, int32_t fd, void *buffer,
+    uint32_t byte_size, uint64_t offset);
+
+extern inline bool kpl_io_uring_write(kpl_io_uring *uring, kpl_object_task *task, int32_t fd, void *buffer,
+    uint32_t byte_size, uint64_t offset);
+
+extern inline bool kpl_io_uring_statx(kpl_io_uring *io_uring, kpl_object_task *task, int32_t dir_fd,
+    const char *path, int32_t flags, uint32_t mask, struct statx *statx_object);
diff --git a/test/io.c b/test/io.c
new file mode 100644
index 0000000..2df3144
--- /dev/null
+++ b/test/io.c
@@ -0,0 +1,20 @@
+
+#include "kpl_test.h"
+
+static const char *linux_version = "/proc/version";
+
+KPL_TEST(statx) {
+    kpl_io_uring io_uring = {};
+    KPL_ASSERT(kpl_io_uring_init(&io_uring, KPL_IO_URING_QUEUE_SIZE_POW2_MIN) == 0, "should return 0 for success");
+    struct statx statx_object = {};
+    kpl_object_task *task = calloc(1, sizeof(kpl_object_task) + sizeof(kpl_class) * 1);
+    task->io_res = 1;
+    KPL_ASSERT(kpl_io_uring_statx(&io_uring, task, AT_FDCWD, linux_version, 0, STATX_ALL, &statx_object),
+        "should add statx operation");
+    KPL_ASSERT(kpl_io_uring_submit(&io_uring) == 1, "should submit one io operation");
+    KPL_ASSERT(task == kpl_io_uring_get_completed(&io_uring), "should return a task");
+    KPL_ASSERT(task->io_res == 0, "io_res should have success stax return");
+    KPL_ASSERT(statx_object.stx_nlink > 0, "should have one or more hard link(s)");
+    free(task);
+    kpl_io_uring_free(&io_uring);
+}