jsrt/lib/rt/io.h

// IO natives. `__native_print_string` writes at once. The rest are async, on
// io_uring (see uring.h)
//
// A `__native_io_queue_<op>` queues an op and gives its handle. It makes no
// syscall, and no native waits. The host's loop (`main` below) puts queued
// ops into the SQ, submits them, waits until one op is done, and gives its
// handle to `jsrt_complete`. The getters read that op's result while
// `jsrt_complete` runs
//
// At most `cq_entries` ops are in flight, so the CQ cannot overflow. The rest
// stay queued until a later wait. So a queue never fails, and the program
// never sees a limit
//
// The ring is made at the first wait that has ops, so a program that does no
// async IO never makes it
#pragma once

#include "str.h"

#ifdef JSRT_IMPLEMENTATION
#include <errno.h>
#include <fcntl.h>
#include <linux/stat.h>
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "uring.h"

// ── sync ──

undefined_t __native_print_string(string_t s) {
    char *utf8 = __rt_string_utf8(s);
    puts(utf8);
    free(utf8);
    return (undefined_t){0};
}

// ── ops ──

typedef enum {
    JSRT_IO_TIMEOUT,
    JSRT_IO_OPEN,
    JSRT_IO_READ,
    JSRT_IO_WRITE,
    JSRT_IO_CLOSE,
    JSRT_IO_STAT,
    JSRT_IO_CANCEL,
} jsrt_io_kind;

typedef enum {
    JSRT_IO_FREE,       // in the free list
    JSRT_IO_QUEUED,     // in the backlog, not in the SQ yet
    JSRT_IO_FLIGHT,     // in the SQ or in the kernel
    JSRT_IO_DONE,       // in the done list, or the last op a wait gave
} jsrt_io_state;

// An op. Its handle is its index in the table, and its SQE's user_data
//
// The table moves when it grows, which only a queue does. So the kernel
// must not hold a pointer into it after a submit returns: `ts` is read
// before (SUBMIT_STABLE), and everything that the kernel reads or writes
// later is in `buf`, which does not move
typedef struct {
    uint8_t kind, state;
    bool cancelled;                 // cancelled while queued: never goes to the SQ
    uint32_t gen;                   // counts reuses of this slot (see CANCEL)
    int32_t next;                   // the next op in its list, or -1
    int32_t res;                    // the CQE's res: >= 0 on success, -errno on failure

    int32_t fd;                     // READ, WRITE, CLOSE
    int32_t flags, mode;            // OPEN
    uint32_t len;                   // READ, WRITE: the size of `buf`
    uint64_t off;                   // READ, WRITE: -1 is the file's position
    int32_t target;                 // CANCEL: the op to cancel...
    uint32_t target_gen;            // ...while it is still this use of its slot
    struct __kernel_timespec ts;    // TIMEOUT
    void *buf;                      // OPEN: path. READ, WRITE: data. STAT: statx, then path
} jsrt_io_op;

// A FIFO of ops, linked through `next`
typedef struct { int32_t head, tail; } jsrt_io_list;

static struct {
    jsrt_uring ring;
    bool ring_ready;
    unsigned in_flight;

    jsrt_io_op *ops;
    int32_t cap;
    int32_t free;                   // a stack, linked through `next`
    jsrt_io_list backlog, done;
    int32_t given;                  // the op the last wait gave, or -1
} jsrt_io = {
    .free = -1,
    .backlog = { -1, -1 },
    .done = { -1, -1 },
    .given = -1,
};

static void jsrt_io_push(jsrt_io_list *l, int32_t h) {
    jsrt_io.ops[h].next = -1;

    if (l->tail < 0) {
        l->head = h;
    } else {
        jsrt_io.ops[l->tail].next = h;
    }

    l->tail = h;
}

static int32_t jsrt_io_pop(jsrt_io_list *l) {
    int32_t h = l->head;

    if (h >= 0) {
        l->head = jsrt_io.ops[h].next;

        if (l->head < 0) {
            l->tail = -1;
        }
    }

    return h;
}

// A new op of `kind`, not in any list yet
static int32_t jsrt_io_new(jsrt_io_kind kind) {
    if (jsrt_io.free < 0) {
        int32_t cap = jsrt_io.cap ? jsrt_io.cap * 2 : 64;
        jsrt_io_op *ops = realloc(jsrt_io.ops, (size_t)cap * sizeof *ops);

        if (!ops) {
            fputs("jsrt: out of memory\n", stderr);
            abort();
        }

        // The new slots go on the free stack, the lowest on top
        for (int32_t h = cap - 1; h >= jsrt_io.cap; h--) {
            ops[h] = (jsrt_io_op){ .state = JSRT_IO_FREE, .next = jsrt_io.free };
            jsrt_io.free = h;
        }

        jsrt_io.ops = ops;
        jsrt_io.cap = cap;
    }

    int32_t h = jsrt_io.free;
    jsrt_io_op *op = &jsrt_io.ops[h];
    jsrt_io.free = op->next;
    *op = (jsrt_io_op){ .kind = kind, .state = JSRT_IO_QUEUED, .gen = op->gen + 1, .next = -1 };
    return h;
}

static int32_t jsrt_io_queue(int32_t h) {
    jsrt_io_push(&jsrt_io.backlog, h);
    return h;
}

// Ends op `h` with `res`, without the kernel
static int32_t jsrt_io_finish(int32_t h, int32_t res) {
    jsrt_io.ops[h].state = JSRT_IO_DONE;
    jsrt_io.ops[h].res = res;
    jsrt_io_push(&jsrt_io.done, h);
    return h;
}

static void jsrt_io_release(int32_t h) {
    jsrt_io_op *op = &jsrt_io.ops[h];
    free(op->buf);
    op->buf = NULL;
    op->state = JSRT_IO_FREE;
    op->next = jsrt_io.free;
    jsrt_io.free = h;
}

// The op that the last wait gave. A handle of any other op is an error
static jsrt_io_op *jsrt_io_given(int32_t h) {
    if (h < 0 || h != jsrt_io.given) {
        __rt_type_error("an IO result read outside its op's callback");
    }

    return &jsrt_io.ops[h];
}

// 0 if op `h` still has work for the kernel, else the -errno it ends with
//
// An op that was cancelled while queued has none. Nor has a cancel whose
// target ended, because then the target's slot can hold a new op
static int32_t jsrt_io_dead(int32_t h) {
    jsrt_io_op *op = &jsrt_io.ops[h];

    if (op->cancelled) {
        return -ECANCELED;
    }

    if (op->kind == JSRT_IO_CANCEL) {
        jsrt_io_op *t = &jsrt_io.ops[op->target];

        if (t->gen != op->target_gen || t->state != JSRT_IO_FLIGHT) {
            return -ENOENT;
        }
    }

    return 0;
}

static void jsrt_io_prep(struct io_uring_sqe *sqe, int32_t h) {
    jsrt_io_op *op = &jsrt_io.ops[h];
    sqe->user_data = (uint64_t)h;

    switch (op->kind) {
    case JSRT_IO_TIMEOUT:
        // A plain timer: `off` 0 means that no count of CQEs ends it
        sqe->opcode = IORING_OP_TIMEOUT;
        sqe->fd = -1;
        sqe->addr = (uint64_t)(uintptr_t)&op->ts;
        sqe->len = 1;
        break;

    case JSRT_IO_OPEN:
        sqe->opcode = IORING_OP_OPENAT;
        sqe->fd = AT_FDCWD;
        sqe->addr = (uint64_t)(uintptr_t)op->buf;
        sqe->len = (uint32_t)op->mode;
        sqe->open_flags = (uint32_t)op->flags;
        break;

    case JSRT_IO_READ:
    case JSRT_IO_WRITE:
        sqe->opcode = op->kind == JSRT_IO_READ ? IORING_OP_READ : IORING_OP_WRITE;
        sqe->fd = op->fd;
        sqe->addr = (uint64_t)(uintptr_t)op->buf;
        sqe->len = op->len;
        sqe->off = op->off;
        break;

    case JSRT_IO_CLOSE:
        sqe->opcode = IORING_OP_CLOSE;
        sqe->fd = op->fd;
        break;

    case JSRT_IO_STAT:
        sqe->opcode = IORING_OP_STATX;
        sqe->fd = AT_FDCWD;
        sqe->addr = (uint64_t)(uintptr_t)((char *)op->buf + sizeof(struct statx));
        sqe->len = STATX_SIZE;
        sqe->addr2 = (uint64_t)(uintptr_t)op->buf;
        break;

    case JSRT_IO_CANCEL:
        // A timeout is removed with its own op
        sqe->opcode = jsrt_io.ops[op->target].kind == JSRT_IO_TIMEOUT
            ? IORING_OP_TIMEOUT_REMOVE
            : IORING_OP_ASYNC_CANCEL;
        sqe->fd = -1;
        sqe->addr = (uint64_t)op->target;
        break;
    }
}

// Moves queued ops into the SQ while fewer than `cq_entries` are in flight,
// and submits them
static void jsrt_io_drain(void) {
    if (jsrt_io.backlog.head < 0) {
        return;
    }

    jsrt_uring *r = &jsrt_io.ring;

    if (!jsrt_io.ring_ready) {
        jsrt_uring_init(r, 128);
        jsrt_io.ring_ready = true;
    }

    while (jsrt_io.backlog.head >= 0 && jsrt_io.in_flight < r->cq_entries) {
        int32_t h = jsrt_io.backlog.head;
        int32_t dead = jsrt_io_dead(h);

        if (dead) {
            jsrt_io_pop(&jsrt_io.backlog);
            jsrt_io_finish(h, dead);
            continue;
        }

        struct io_uring_sqe *sqe = jsrt_uring_sqe(r);

        // The SQ is full: a submit empties it
        if (!sqe) {
            jsrt_uring_submit(r, false);
            continue;
        }

        jsrt_io_pop(&jsrt_io.backlog);
        jsrt_io_prep(sqe, h);
        jsrt_io.ops[h].state = JSRT_IO_FLIGHT;
        jsrt_io.in_flight++;
    }

    // No SQE may stay unsubmitted when the wait returns, because a queue can
    // then move the table (see jsrt_io_op)
    jsrt_uring_submit(r, false);
}

// ── natives ──

int32_t __native_io_queue_timeout(double ms) {
    int32_t h = jsrt_io_new(JSRT_IO_TIMEOUT);

    if (!(ms > 0)) {
        ms = 0;
    }

    jsrt_io.ops[h].ts.tv_sec = (int64_t)(ms / 1000);
    jsrt_io.ops[h].ts.tv_nsec = (int64_t)(fmod(ms, 1000) * 1e6);
    return jsrt_io_queue(h);
}

// `flags` and `mode` are open(2)'s, as numbers
int32_t __native_io_queue_open(string_t path, int32_t flags, int32_t mode) {
    int32_t h = jsrt_io_new(JSRT_IO_OPEN);
    jsrt_io.ops[h].buf = __rt_string_utf8(path);
    jsrt_io.ops[h].flags = flags | O_CLOEXEC;
    jsrt_io.ops[h].mode = mode;
    return jsrt_io_queue(h);
}

// Reads at most `len` bytes at `offset`, or at the file's position if
// `offset` is -1. The status is the number of bytes read
int32_t __native_io_queue_read(int32_t fd, int32_t len, double offset) {
    int32_t h = jsrt_io_new(JSRT_IO_READ);
    jsrt_io_op *op = &jsrt_io.ops[h];
    op->fd = fd;
    op->len = len > 0 ? (uint32_t)len : 0;
    op->off = offset < 0 ? (uint64_t)-1 : (uint64_t)offset;
    op->buf = __rt_alloc(op->len ? op->len : 1);
    return jsrt_io_queue(h);
}

// Writes `data` as UTF-8, at `offset` as for a read. The status is the number
// of bytes written, which can be fewer than all
int32_t __native_io_queue_write(int32_t fd, string_t data, double offset) {
    int32_t h = jsrt_io_new(JSRT_IO_WRITE);
    jsrt_io_op *op = &jsrt_io.ops[h];
    op->fd = fd;
    op->buf = __rt_string_utf8(data);
    op->len = (uint32_t)strlen(op->buf);
    op->off = offset < 0 ? (uint64_t)-1 : (uint64_t)offset;
    return jsrt_io_queue(h);
}

int32_t __native_io_queue_close(int32_t fd) {
    int32_t h = jsrt_io_new(JSRT_IO_CLOSE);
    jsrt_io.ops[h].fd = fd;
    return jsrt_io_queue(h);
}

// Gives the file's size, as `__native_io_result_number`
int32_t __native_io_queue_stat(string_t path) {
    int32_t h = jsrt_io_new(JSRT_IO_STAT);
    char *utf8 = __rt_string_utf8(path);
    size_t size = strlen(utf8) + 1;
    char *buf = __rt_alloc(sizeof(struct statx) + size);
    memcpy(buf + sizeof(struct statx), utf8, size);
    free(utf8);
    jsrt_io.ops[h].buf = buf;
    return jsrt_io_queue(h);
}

// Cancels op `target`, which then ends with -ECANCELED if it did not end
// first. The cancel's own status is 0 if it found the op, else -ENOENT or
// -EALREADY
int32_t __native_io_queue_cancel(int32_t target) {
    if (target < 0 || target >= jsrt_io.cap) {
        __rt_type_error("a cancel of an op that does not exist");
    }

    int32_t h = jsrt_io_new(JSRT_IO_CANCEL);
    jsrt_io_op *t = &jsrt_io.ops[target];

    switch (t->state) {
    case JSRT_IO_QUEUED:
        t->cancelled = true;
        return jsrt_io_finish(h, 0);
    case JSRT_IO_FLIGHT:
        jsrt_io.ops[h].target = target;
        jsrt_io.ops[h].target_gen = t->gen;
        return jsrt_io_queue(h);
    default:
        return jsrt_io_finish(h, -ENOENT);
    }
}

// The handle of an op that is done, or -1 if no op is queued or in flight.
// The op that the last call gave is freed now
static int32_t jsrt_io_next(void) {
    if (jsrt_io.given >= 0) {
        jsrt_io_release(jsrt_io.given);
        jsrt_io.given = -1;
    }

    jsrt_io_drain();

    int32_t h = jsrt_io_pop(&jsrt_io.done);

    while (h < 0) {
        if (jsrt_io.in_flight == 0) {
            return -1;
        }

        uint64_t user_data;
        int32_t res;

        if (!jsrt_uring_cqe(&jsrt_io.ring, &user_data, &res)) {
            jsrt_uring_submit(&jsrt_io.ring, true);
            continue;
        }

        jsrt_io.in_flight--;
        h = (int32_t)user_data;

        // A timer that runs out ends with -ETIME, which is its success
        bool timer = jsrt_io.ops[h].kind == JSRT_IO_TIMEOUT;
        jsrt_io.ops[h].state = JSRT_IO_DONE;
        jsrt_io.ops[h].res = timer && res == -ETIME ? 0 : res;
    }

    jsrt_io.given = h;
    return h;
}

// >= 0 if the op succeeded (a count of bytes, an fd, or 0), else -errno
int32_t __native_io_status(int32_t h) {
    return jsrt_io_given(h)->res;
}

// The data of a read, decoded as UTF-8. Empty for any other op, or a failure
string_t __native_io_result_string(int32_t h) {
    jsrt_io_op *op = jsrt_io_given(h);
    bool data = op->kind == JSRT_IO_READ && op->res > 0;
    return __rt_string_from_utf8(op->buf, data ? (size_t)op->res : 0);
}

// The size of a stat's file. The status for any other op
double __native_io_result_number(int32_t h) {
    jsrt_io_op *op = jsrt_io_given(h);

    if (op->kind == JSRT_IO_STAT && op->res >= 0) {
        return (double)((struct statx *)op->buf)->stx_size;
    }

    return op->res;
}

#ifndef JSRT_NO_HOST
// The Linux flavor's host: run the program, then give it each op as it
// completes, until none is left
int main(void) {
    jsrt_start();

    for (int32_t h; (h = jsrt_io_next()) >= 0;) {
        jsrt_complete(h);
    }

    return 0;
}
#endif

#endif // JSRT_IMPLEMENTATION