Io

Runs proc as the first fiber of a new scheduler on this thread and returns when it has returned. Fibers it started that are still alive then are canceled, and run until they end. Everything that waits (sleep, accept, read, channels) runs inside it, and only the calling fiber waits, never the thread. The scheduler's memory is released when the call returns; the caller keeps what proc allocated. Do not call it from inside another io_run on the same thread.

Io :: #import "Extensions/Io";

done: bool;

main :: () {
    Io.io_run(() {
        Io.sleep(1);
        done = true;
    });

    assert(done);
}

Starts proc as a new fiber on the current scheduler and returns its handle. It first runs when the calling fiber blocks or yields, and not at all if the scheduler ends first. The new fiber starts with a copy of the caller's context and temporary storage of its own. Fibers share one OS thread, so they can use shared data without locks as long as they do not block in the middle of updating it.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        results := Io.make_channel(int, 1);
        defer Io.free_channel(results);

        Io.spawn((out: *Io.Channel(int)) { Io.send(out, 42); }, results);

        value, error := Io.recv(results); // waits for the spawned fiber
        assert(error == .NONE && value == 42);
    });
}

Lets the other ready fibers run first. .CANCELED if this fiber was canceled.

Io :: #import "Extensions/Io";

Io.io_run(() {
    log: [..] int;
    Io.spawn((out: *[..] int) { array_add(out, 2); }, *log);
    array_add(*log, 1);
    Io.yield();            // the spawned fiber runs now
    array_add(*log, 3);
    assert(log[0] == 1 && log[1] == 2 && log[2] == 3);
});

Blocks this fiber, not the thread, for at least milliseconds. Ends early with .TIMED_OUT when a with_deadline limit comes first.

Io :: #import "Extensions/Io";

Io.io_run(() {
    error := Io.sleep(5);
    assert(error == .NONE);
});

Makes the blocking calls of fiber return .CANCELED, now and from now on. The call it is blocked in, if any, returns at once. Does nothing for a fiber that has ended.

Io :: #import "Extensions/Io";

Io.io_run(() {
    sleeper := Io.spawn(() {
        error := Io.sleep(60_000);
        assert(error == .CANCELED);
    });

    Io.yield();             // let it start sleeping
    Io.cancel(sleeper);
});

Whether the running fiber was canceled, for loops that do not block.

Io :: #import "Extensions/Io";

Io.io_run(() {
    assert(!Io.is_canceled());
});

Whether the calling code runs on a fiber of a scheduler, that is inside io_run.

Io :: #import "Extensions/Io";

assert(!Io.in_scheduler());
Io.io_run(() { assert(Io.in_scheduler()); });

The handle of the running fiber.

Io :: #import "Extensions/Io";

Io.io_run(() {
    me := Io.current();
    assert(me.fiber != null);
});

Gives the memory of the running fiber's stack that no frame uses any more back to the system: the pages below the current frames that earlier, deeper calls touched. Call it before a long wait (an idle connection) so those pages do not stay resident; it costs a system call, so not in a hot path. The stack's contents below the caller are lost, which nothing can rely on anyway. Does nothing outside a fiber or on a platform without a way to discard pages.

A channel that holds up to capacity values, made with context.allocator.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(int, 2);
    defer Io.free_channel(channel);

    Io.send(channel, 1);
    value, error := Io.recv(channel);
    assert(error == .NONE && value == 1);
});

Frees a channel from make_channel. No fiber may be waiting on it.

Puts value into the channel, blocking while it is full. .CLOSED if the channel is (or becomes) closed, .CANCELED if the fiber is canceled while it waits.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(string, 1);
    defer Io.free_channel(channel);

    assert(Io.send(channel, "first") == .NONE);

    Io.close(channel);
    assert(Io.send(channel, "second") == .CLOSED);
});

A send that gives up instead of blocking: whether the value was taken.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(int, 1);
    defer Io.free_channel(channel);

    assert(Io.try_send(channel, 1));
    assert(!Io.try_send(channel, 2)); // full
});

Takes the oldest value, blocking while the channel is empty. .CLOSED once it is closed and drained, .CANCELED if the fiber is canceled while it waits.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(int, 1);
    defer Io.free_channel(channel);

    Io.spawn((c: *Io.Channel(int)) { Io.send(c, 7); }, channel);

    value, error := Io.recv(channel); // waits for the spawned fiber
    assert(error == .NONE && value == 7);
});

A recv that gives up instead of blocking: the value and whether there was one.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(int, 1);
    defer Io.free_channel(channel);

    _, found := Io.try_recv(channel);
    assert(!found);

    Io.send(channel, 5);
    value, found2 := Io.try_recv(channel);
    assert(found2 && value == 5);
});

No more values will be sent. Fibers waiting to send get .CLOSED; receivers get what is buffered, then .CLOSED.

Io :: #import "Extensions/Io";

Io.io_run(() {
    channel := Io.make_channel(int, 2);
    defer Io.free_channel(channel);

    Io.send(channel, 1);
    Io.close(channel);

    value, error := Io.recv(channel);  // what was buffered still arrives
    assert(error == .NONE && value == 1);

    _, error = Io.recv(channel);
    assert(error == .CLOSED);
});

Adds count tasks to wait for.

Marks one task finished.

Blocks until the count is back to zero.

Takes the lock, waiting for whoever holds it. Pair it with unlock.

Io :: #import "Extensions/Io";

Io.io_run(() {
    mutex: Io.Fiber_Mutex;

    assert(Io.lock(*mutex) == .NONE);
    Io.sleep(1);               // safe to wait while holding it
    Io.unlock(*mutex);
});

Takes the lock if it is free, without waiting; returns whether it did.

Io :: #import "Extensions/Io";

mutex: Io.Fiber_Mutex;
assert(Io.try_lock(*mutex));
assert(!Io.try_lock(*mutex));
Io.unlock(*mutex);

Releases the lock, handing it to the fiber that has waited longest.

Wakes the fiber that has waited longest, if any.

Io :: #import "Extensions/Io";

Io.io_run(() {
    changed: Io.Condition;

    Io.spawn((changed: *Io.Condition) { Io.notify_one(changed); }, *changed);

    assert(Io.wait(*changed) == .NONE); // the spawned fiber notifies
});

Wakes every fiber waiting on the condition.

An IPv4 address from its four octets, a.b.c.d, and a port in host order. address_to_string turns it back into text.

Io :: #import "Extensions/Io";

address := address_v4(127, 0, 0, 1, 8080);

assert(address.family == .IPV4);
assert(address.bytes[0] == 127 && address.bytes[3] == 1);
assert(address_to_string(address) == "127.0.0.1:8080");

Reads an address literal: 1.2.3.4:80, :8080 (any IPv4 address), [::1]:80, 1.2.3.4, ::1. It does not look names up; resolve does.

Io :: #import "Extensions/Io";

address, ok := Io.parse_address("[::1]:8080");
assert(ok && address.family == .IPV6 && address.port == 8080);

1.2.3.4:80 or [::1]:80, in temporary storage.

Io :: #import "Extensions/Io";

address, _ := Io.parse_address("10.0.0.7:443");
assert(Io.address_to_string(address) == "10.0.0.7:443");

The system's error number behind the last failed IO call of the running fiber.

Listens on address (a literal such as ":8080", port 0 picks a free one; see local_address). reuse_port lets several listeners, one per thread, share the port; the kernel then spreads the connections between them.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        listener, error := Io.tcp_listen("127.0.0.1:0");
        assert(error == .NONE);
        defer Io.close(listener);

        address, _ := Io.local_address(listener);
        assert(address.port != 0); // the system picked a free port

        _, bad := Io.tcp_listen("not an address");
        assert(bad == .INVALID);
    });
}

Waits for the next connection.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    assert(Io.write(client, "hi") == 2);

    buffer: [8] u8;
    count, error := Io.read(server, buffer);
    assert(error == .NONE && string.{ count, buffer.data } == "hi");
});

Connects to address. The one with a name is connect("host:port").

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, error := Io.connect(address);
    assert(error == .NONE);
    Io.close(client);
});

Listens on the socket file path. The file must not exist yet; with replace an existing file at path is removed first (whatever it is, so use it only for a path you own). The socket file stays after close: remove it when the server is done.

Io :: #import "Extensions/Io";

#import "File";

Io.io_run(() {
    listener, error := Io.unix_listen("/tmp/jacket-io-unix.sock", replace = true);
    assert(error == .NONE);
    Io.close(listener);
});
file_delete("/tmp/jacket-io-unix.sock");

Connects to the Unix-domain socket at path.

Io :: #import "Extensions/Io";

#import "File";

Io.io_run(() {
    listener, _ := Io.unix_listen("/tmp/jacket-io-unix2.sock", replace = true);
    defer Io.close(listener);

    client, error := Io.connect_unix("/tmp/jacket-io-unix2.sock");
    assert(error == .NONE);
    Io.close(client);
});
file_delete("/tmp/jacket-io-unix2.sock");

The address this socket is bound to; with tcp_listen(":0"), which port it got.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        listener, _ := Io.tcp_listen("127.0.0.1:0");
        defer Io.close(listener);

        address, error := Io.local_address(listener);
        assert(error == .NONE);
        assert(address.family == .IPV4 && address.port != 0);
    });
}

The far end of a connected stream. On the accepting side this is the client's address and its ephemeral port; on the connecting side it is the server's.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        listener, _ := Io.tcp_listen("127.0.0.1:0");
        defer Io.close(listener);

        address, _ := Io.local_address(listener);
        client, connect_error := Io.connect(address);
        assert(connect_error == .NONE);
        defer Io.close(client);

        server, accept_error := Io.accept(listener);
        assert(accept_error == .NONE);
        defer Io.close(server);

        far, _ := Io.peer_address(client);
        assert(far.port == address.port); // the client's peer is the listener
        near, _ := Io.local_address(client);
        seen, _ := Io.peer_address(server);
        assert(seen.port == near.port); // the server sees the client's own port
    });
}

Waits for data and reads what is there, up to buffer.count bytes. 0 means the peer closed its end. An error comes with 0 bytes.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    Io.write(client, "hello");

    buffer: [16] u8;
    count, error := Io.read(server, buffer);
    assert(error == .NONE && string.{ count, buffer.data } == "hello");
});

Waits until read would not wait: bytes have arrived, the peer closed its end, or the connection failed (read then says which). Nothing is consumed and no buffer is needed, so a fiber can wait for a quiet connection and take its memory only when there is something to read. Ends with .TIMED_OUT or .CANCELED like any other wait, and .CLOSED if the socket is closed meanwhile.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    Io.write(client, "x");

    assert(Io.wait_readable(server) == .NONE); // data is waiting, nothing consumed
    buffer: [4] u8;
    count, _ := Io.read(server, buffer);
    assert(count == 1);
});

Reads until buffer is full. .END_OF_FILE if the peer closed first, with the bytes read.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    Io.write(client, "abcd");

    buffer: [4] u8;
    count, error := Io.read_full(server, buffer);
    assert(error == .NONE && count == 4);
});

Sends some of data: at least one byte unless data is empty. Returns how many. Use write to send everything.

Sends all of data, waiting as often as it takes. On an error it returns how many bytes went out.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    count, error := Io.write(client, u8.[1, 2, 3]);
    assert(error == .NONE && count == 3);
});

Closes the sending side; the peer reads the end of the stream once it has what was sent.

Io :: #import "Extensions/Io";

Io.io_run(() {
    listener, _ := Io.tcp_listen("127.0.0.1:0");
    defer Io.close(listener);

    address, _ := Io.local_address(listener);
    client, _ := Io.connect(address);
    server, _ := Io.accept(listener);
    defer Io.close(client);
    defer Io.close(server);

    Io.write(client, "done");
    Io.shutdown_write(client);

    data, _ := Io.read_all(Io.reader_of(server));
    assert(cast(string) data == "done");
    free(data.data);
});

Looks host up (an address literal, or a name for the system's resolver, which runs on a helper thread so only this fiber waits) and gives its first address, IPv4 preferred.

Io :: #import "Extensions/Io";

Io.io_run(() {
    address, error := Io.resolve("127.0.0.1", 8080);
    assert(error == .NONE && address.port == 8080);

    host, _ := Io.resolve("localhost", 80);
    assert(host.family != .NONE);
});

Two connected ends. What one writes the other reads. close both when done (they free themselves once both are closed).

Io :: #import "Extensions/Io";

Io.io_run(() {
    client, server := Io.make_pipe();
    defer Io.close(client);
    defer Io.close(server);

    Io.write(client, "ping");

    buffer: [8] u8;
    count, error := Io.read(server, buffer);
    assert(error == .NONE && string.{ count, buffer.data } == "ping");
});

Opens path. Close the file with close.

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    file, error := Io.open_file("/tmp/jacket-io-open.txt", .WRITE);
    assert(error == .NONE);
    Io.write(*file, "hello");
    Io.close(*file);

    _, error = Io.open_file("/tmp/jacket-io-missing/none.txt");
    assert(error == .NOT_FOUND);
});
file_delete("/tmp/jacket-io-open.txt");

Reads at offset; the file's position does not move.

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    Io.write_file("/tmp/jacket-io-at.txt", "0123456789");

    file, _ := Io.open_file("/tmp/jacket-io-at.txt");
    defer Io.close(*file);

    buffer: [3] u8;
    Io.read_at(*file, buffer, 4);
    assert(string.{ 3, buffer.data } == "456");
});
file_delete("/tmp/jacket-io-at.txt");

Writes all of data at offset; the file's position does not move.

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    file, _ := Io.open_file("/tmp/jacket-io-write-at.txt", .READ_WRITE);
    defer Io.close(*file);

    Io.write_at(*file, "abcdef", 0);
    Io.write_at(*file, "XY", 2);

    data, _ := Io.read_file("/tmp/jacket-io-write-at.txt");
    assert(cast(string) data == "abXYef");
    free(data.data);
});
file_delete("/tmp/jacket-io-write-at.txt");

The size of the file in bytes.

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    Io.write_file("/tmp/jacket-io-size.txt", "12345");

    file, _ := Io.open_file("/tmp/jacket-io-size.txt");
    defer Io.close(*file);

    size, error := Io.file_size(*file);
    assert(error == .NONE && size == 5);
});
file_delete("/tmp/jacket-io-size.txt");

The whole content of a file, in memory from context.allocator (free it with free).

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    Io.write_file("/tmp/jacket-io-whole.txt", "contents");

    data, error := Io.read_file("/tmp/jacket-io-whole.txt");
    assert(error == .NONE && cast(string) data == "contents");
    free(data.data);
});
file_delete("/tmp/jacket-io-whole.txt");

Creates or replaces a file with data.

Io :: #import "Extensions/Io";
#import "File";

Io.io_run(() {
    assert(Io.write_file("/tmp/jacket-io-wf.txt", "hello") == .NONE);
});
file_delete("/tmp/jacket-io-wf.txt");

Whether wait_readable works on reader, which is when its source supplies a wait_proc (a TCP stream does; a file or a pipe end does not). When it is false, wait_readable returns .UNSUPPORTED.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        near, far := Io.make_pipe();
        defer Io.close(near);
        defer Io.close(far);

        assert(!Io.can_wait_readable(Io.reader_of(near)));
    });
}

Reads to the end of the source into memory from context.allocator (free it with free). size_hint is the expected length, when it is known. On any error the bytes read so far are freed.

Io :: #import "Extensions/Io";

Io.io_run(() {
    near, far := Io.make_pipe();
    defer Io.close(near);
    defer Io.close(far);
    Io.write(Io.writer_of(far), "everything");
    Io.shutdown_write(far);

    data, error := Io.read_all(Io.reader_of(near));
    assert(error == .NONE && cast(string) data == "everything");
    free(data.data);
});

The file as a Reader. Reading advances the file's position. The reader points at the Io_File, which must outlive it.

Io :: #import "Extensions/Io";
#import "File";

path := "/tmp/jacket-io-reader-of.txt";
Io.io_run(() {
    Io.write_file("/tmp/jacket-io-reader-of.txt", "hello");

    file, _ := Io.open_file("/tmp/jacket-io-reader-of.txt");
    defer Io.close(*file);

    data, error := Io.read_all(Io.reader_of(*file));
    assert(error == .NONE && cast(string) data == "hello");
    free(data.data);
});
file_delete(path);

The file as a Writer. Writing advances the file's position. The writer points at the Io_File, which must outlive it.

Io :: #import "Extensions/Io";
#import "File";

path := "/tmp/jacket-io-writer-of.txt";
Io.io_run(() {
    file, _ := Io.open_file("/tmp/jacket-io-writer-of.txt", .WRITE);
    defer Io.close(*file);

    count, error := Io.write(Io.writer_of(*file), "hello");
    assert(error == .NONE && count == 5);
});
file_delete(path);

A line reader over source that accepts lines of up to limit bytes (not counting the line ending). The limit stops a peer that never sends a newline from filling memory.

Io :: #import "Extensions/Io";

Io.io_run(() {
    near, far := Io.make_pipe();
    defer Io.close(near);
    defer Io.close(far);
    Io.write(Io.writer_of(far), "one\r\ntwo\n");
    Io.shutdown_write(far);

    lines := Io.make_line_reader(Io.reader_of(near), limit = 1024);
    defer Io.free_line_reader(*lines);

    line, error := Io.read_line(*lines);
    assert(error == .NONE && line == "one");
});

Frees the reader's buffer. Lines it returned are invalid afterwards.

Io :: #import "Extensions/Io";

Io.io_run(() {
    near, far := Io.make_pipe();
    defer Io.close(near);
    defer Io.close(far);

    lines := Io.make_line_reader(Io.reader_of(near));
    Io.free_line_reader(*lines);
});

The next line without its line ending (\n or \r\n), or the error that ended the reading: .END_OF_FILE once the source has ended and every line was handed out, .INVALID for a line longer than the limit, or whatever the source's read reported. After an error, stop reading and free the reader.

Io :: #import "Extensions/Io";

Io.io_run(() {
    near, far := Io.make_pipe();
    defer Io.close(near);
    defer Io.close(far);
    Io.write(Io.writer_of(far), "GET a\nGET b\nlast");
    Io.shutdown_write(far);

    lines := Io.make_line_reader(Io.reader_of(near));
    defer Io.free_line_reader(*lines);

    whole := 0;

    while true {
        line, error := Io.read_line(*lines);
        if error != .NONE break;
        if lines.unterminated break; // "last" was cut off: do not act on it
        whole += 1;
    }

    assert(whole == 2);
});

Reads a document to the end of reader and decodes it as a T: the same value or error as from_json(T, text) for the same bytes, however they arrive (a pipe written in small pieces, a socket). The pieces are gathered in one buffer that is freed before this returns. A reader that fails gives a .READ error naming the Io_Error; a document over options.max_size is a .LIMIT error and the rest of the stream is left unread. The reader is not closed.

Io :: #import "Extensions/Io";
#import "Extensions/Json";

Point :: struct { x: int; y: int; }

Io.io_run(() {
    near, far := Io.make_pipe();
    defer Io.close(near);
    defer Io.close(far);

    Io.write(Io.writer_of(far), "{\"x\": 1, \"y\": 2}");
    Io.shutdown_write(far);

    point, error := Io.read_json(Io.reader_of(near), Point);
    assert(error.kind == .NONE && point.x == 1 && point.y == 2);
});

The one way to bound a wait. Limits every IO call (and sleep) in body to milliseconds from now: one that is still waiting then returns .TIMED_OUT, and so does a call made after the limit, even when data is ready. 0 is a limit that has already passed. Deadlines nest; the earlier one wins, and the previous deadline is back when body ends. A negative milliseconds adds no limit (NO_DEADLINE), so a timeout setting that may be off needs no second code path. Channel and lock operations are not IO and ignore it.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        client, server := Io.make_pipe();
        buffer: [8] u8;
        error: Io.Io_Error;

        Io.with_deadline(20, #code {
            _, error = Io.read(client, buffer);       // nothing is ever sent
        });

        assert(error == .TIMED_OUT);
    });
}

Runs body clear of the running fiber's deadline and cancel, for cleanup that has to happen after one of them fired: telling a server to stop a query, a goodbye message. body has its own limit of milliseconds (negative: none), and with_deadline inside it nests as usual. When body ends, the outer deadline is back, and the fiber is canceled if it was before or a cancel came during body.

Io :: #import "Extensions/Io";

main :: () {
    Io.io_run(() {
        client, server := Io.make_pipe();
        buffer: [8] u8;
        error: Io.Io_Error;
        sent: s64;

        Io.with_deadline(0, #code {
            _, error = Io.read(client, buffer);                // .TIMED_OUT: the deadline has passed
            Io.with_grace(1000, #code { sent, _ = Io.write(client, "stop"); });
        });

        assert(error == .TIMED_OUT);
        assert(sent == 4);
    });
}

The name of a backend as JAI_IO_BACKEND spells it, such as "epoll", or "none" for .NONE. The string is a constant.

assert(name_of_backend(.EPOLL) == "epoll");
assert(name_of_backend(.NONE) == "none");

The backend the running scheduler uses: .IO_URING, .EPOLL, .KQUEUE, .IOCP, .IORING, or .NONE where IO is not implemented. Set JAI_IO_BACKEND to io_uring, epoll, kqueue, iocp or ioring before io_run to ask for one; a backend the system cannot offer is replaced (see backend_reason).

Io :: #import "Extensions/Io";


Io.io_run(() {
    assert(Io.backend() != .NONE);
    assert(Io.backend_name().count > 0);
});

backend() as text: "io_uring", "epoll", "kqueue", "iocp", "ioring" or "none".

Why the running scheduler does not use the backend that was asked for or is the first choice on this system (for example "io_uring is disabled here"). Empty when it does.

What the engine found it could use.

The number of processors, for sizing io_run_threads.

Io :: #import "Extensions/Io";


assert(Io.processor_count() >= 1);

Starts count schedulers, each on a thread of its own, runs proc as the first fiber of every one, and returns when all have finished. Fibers never move between them. With reuse_port listeners (tcp_listen(..., reuse_port = true)) each thread accepts its own connections.

Io :: #import "Extensions/Io";
Atomics :: #import "Atomics";

started: s32;

main :: () {
    Io.io_run_threads(2, () {
        Atomics.atomic_add(*started, 1);
    });

    assert(started == 2);
}

The value a library stored for key on the running scheduler, or null. Keys are addresses of something the library owns (a global works). This is the per-thread variable the language does not have: the Http client keeps its connection pool here.

Io :: #import "Extensions/Io";


Io.io_run(() {
    counter: int;
    Io.set_scheduler_local(*counter, *counter, null);
    assert(Io.scheduler_local(*counter) == *counter);
});

Stores value for key on the running scheduler; drop (if not null) gets it when the scheduler ends. Replaces an earlier value for the key without dropping it.

Makes SIGINT and SIGTERM set the flag interrupted reads, instead of ending the process. The handler is installed once for the whole program; calling this again does nothing.

True once SIGINT or SIGTERM arrived after catch_interrupts, or request_interrupt ran.

Io :: #import "Extensions/Io";

Io.clear_interrupt();
assert(!Io.interrupted());
Io.request_interrupt();
assert(Io.interrupted());
Io.clear_interrupt();

Sets the flag by hand, as a signal would (tests and programs that stop themselves).

Clears the flag (tests that run several servers in one process).

Windows pages are 4 KiB on x64 and arm64 alike.

A stack of at least bytes bytes (rounded up to whole pages), or a zero base.

Gives the memory of the stack's whole pages that lie DISCARD_MARGIN bytes or more below frame back to the system. The pages stay mapped (and the guard untouched); their next use finds zeros on macOS and Linux and unspecified contents on Windows. frame is an address in the frame of the running procedure: everything below it must be dead, and the margin is left for the calls that do this.

-errno when a call returned -1, else result.

Makes fd non-blocking and close-on-exec.

Makes fd a socket the engine can use: close-on-exec, no SIGPIPE, and non-blocking when the backend wants readiness (epoll, kqueue). io_uring does its own waiting and wants sockets that block.

The pending error of a socket (SO_ERROR), as a positive errno or 0.

A pipe whose read end is non-blocking when nonblocking (the backend waits for readiness); the write end always blocks, which a one-byte wake-up never does for long.

Closes a handle of any kind (sockets and files share one call on POSIX).

The options a listening socket gets before bind: the address may be reused after a restart, and with reuse_port several listeners (one per thread) may share it.

The value of an environment variable, or "" if it is unset. buffer is scratch space that the Windows version fills.

Whether a sys_recv would not wait, without taking anything: 0 if a byte, the end of the stream or an error is there (the receive reports which), -E_AGAIN if not.

Reads at offset, or at the current position when it is negative.

libc is declared in stacks.jai.

open and fcntl are variadic in C, which matters on Apple arm64 (variadic arguments go on the stack).

The backend a user asked for with JAI_IO_BACKEND, or .NONE if it is unset or unknown.

Starts op. Returns true if it already finished (its result is set, nothing will be reported later), false if the engine will report it through finish_op.

Asks the engine to end op early. The operation still completes, with the system's cancel error as its result, and only then does its fiber wake.

Submits what was queued and waits up to timeout nanoseconds (forever if negative, not at all if 0) for completions, which wake their fibers.

The engine reports op as complete.

Cancels the engine operation fiber is blocked in. Its fiber wakes when the operation completes.

Describes an operation for the running fiber.

Runs op for the running fiber and blocks it until the operation completes, is canceled, or the fiber's deadline passes. The result stays in op.result.

A portable error for a positive platform error number.

The error of a plain system call that returned result (negative: -errno), for the running fiber, and remembers the code for platform_code.

Completions that belong to the engine itself rather than a fiber: the wake-up pipe resolver threads write to (resolver.jai) and the eventfd of the file pool (files_pool.jai).

Puts the kernel in charge of waking op, which is queued on fd. True if that failed and the operation finished with the error; false if it waits.

Whether op waits for the descriptor to become writable.

poll_attempt plus what epoll remembers about the result (see the header).

One non-blocking try. True: finished, op.result is set. False: it must wait for readiness (op.wants_write says which).

Serves the operations waiting on fd now that it may be ready, then arms it again if any remain.

Ends every operation waiting on fd with result.

fd is about to be closed: fail what waits on it and forget what the kernel was told.

Makes sure the kernel will report fd when someone is waiting on it. 0, or the negative error number if the kernel refused (the waiters are left for the caller to fail).

Hands the changes queued so far to the kernel.

Whether epoll can be had at all; it always can on a Linux this module supports.

Registers fd for what its waiters need, once and edge-triggered: the registration stays and only grows (a MOD to add a direction, which also reports the descriptor if it is ready already, so nothing is missed between the failed try and here).

Tries to make the ring; returns why it could not, or "" if engine now uses io_uring.

A free submission entry. With none free, the queued ones go to the kernel first, and completions are reaped so that a full completion queue cannot hold submission back.

Operations still waiting on a descriptor would keep the file alive past the close, so they are canceled first. The cancel is hard-linked so that finding nothing to cancel does not stop the close.

accept through one multishot request per listener instead of one request per connection. Returns false (and does nothing) when this call has to use a one-shot accept instead: another fiber is already waiting on the listener, or the request is being canceled.

The listener fd is being closed: sockets accepted ahead of time are closed, and the slot goes when the kernel has ended its request (the close cancels it).

close(now = true): the descriptor is closed without waiting, so the multishot request has to be canceled by hand or it would keep the socket open.

Wakes the scheduler that owns port from any thread.

Finishes op at once with the Win32 error error (negated into the result).

After an overlapped call: the packet will come if it succeeded or is pending.

Starts op. True: it finished now (the result is set). False: its packet will arrive.

The file offset goes into the OVERLAPPED. A negative one means the end of the file (for appending handles, which only have FILE_APPEND_DATA access) or the start.

Takes packets for up to timeout nanoseconds (forever if negative), finishing operations.

The packet of op arrived with this NTSTATUS and byte count.

Whether this system has a usable IoRing: "" if so, else why not. The answer is kept.

Reads the first two bytes of the running executable through ring and waits for the event, then checks that the ring accepts a write entry.

Makes this engine's ring. Returns why it cannot, or "" if files now use it.

Waits for the ring's completion event and tells the scheduler's port.

A file read or write in the ring. True: it failed at once. False: its completion will come.

Puts op in the submission queue. False if the ring refused it.

Hands the queued entries to the kernel.

The Win32 error an HRESULT from the ring stands for (0 for success).

Takes the completions that are in the ring, finishes their operations, and lets waiting ones in.

Blocks the thread until every operation in the ring has finished, for close of a file.

Reads the pool's settings from the environment into the engine.

A non-negative integer from the environment, or fallback.

Takes the engine's part in the pool, starting the workers if it is the first. Null if the pool cannot be had (the caller then runs the operation inline).

One scheduler less; the last one ends the workers.

Hands the read or write op to a worker. False if the pool is not available.

A worker: takes operations off the queue, makes the call, reports to the owner.

Finishes what the workers have completed.

Blocks the thread until every operation handed to the pool has been taken back.

The engine is going away: wait for the workers, release the wake-up, leave the pool.

Called with the link's mutex held.

Keeps a READ on the eventfd outstanding while operations are out.

The READ on the eventfd completed.

Starts Winsock for this process (reference counted; win_sockets_stop undoes it).

The extension function guid names, for a socket of this provider, or null.

The Win32 error behind an NTSTATUS, the way a completion packet reports failures.

Win32 errors with no E_* constant of their own that mean one of the portable errors.

The job for host that is queued or running, if any.

Stops fiber waiting for its lookup. The job stays: a helper may be running it, and others may wait for it. It is freed when its answer is collected.

Starts helpers until there are wanted (at most RESOLVER_MAX_THREADS). A helper that cannot be started is left out; the caller checks helpers_started.

Makes sure a READ on the pipe is outstanding while lookups are, and takes answers that are already there.

The READ on the pipe completed.

Takes the finished jobs and wakes their fibers.

Waits for the lookups still under way, ends the helpers and closes the pipe.

Runs on a helper thread: no context is needed for plain C calls.

libc is declared in stacks.jai.

open and fcntl are variadic in C, which matters on Apple arm64 (variadic arguments go on the stack).

Why a blocking call returned early or failed. platform_code() gives the system's own error number for the last failure of the running fiber, which is how OTHER stays useful.

Where a fiber is in its life: waiting in the pool, ready, running, blocked on a timer, channel or mutex, or finished. It is the scheduler's own bookkeeping; programs rarely need it.

One fiber. Treat it as opaque: use Fiber_Handle to refer to one from outside the scheduler.

A name for one fiber, returned by spawn. It stays safe to use after the fiber has ended: the generation tells a reused fiber slot from the one the handle named, so cancel and join on a stale handle do nothing.

A bounded queue of T between fibers. send blocks while it is full, recv while it is empty. With capacity 0 a send and a receive meet: each waits for the other.

Waits for a number of tasks, as many as were added and have not yet been done.

Io :: #import "Extensions/Io";

Io.io_run(() {
    group: Io.Wait_Group;

    for 1..3 {
        Io.add(*group);
        Io.spawn((group: *Io.Wait_Group) {
            Io.sleep(1);
            Io.done(group);
        }, *group);
    }

    assert(Io.wait(*group) == .NONE); // returns once all three are done
});

A lock for fibers. A fiber that cannot take it waits without blocking the thread, so unlike Thread's Mutex it is safe to hold across a sleep or a channel operation. It is not recursive, and waiters get it in the order they asked.

Fibers waiting until another fiber says that something changed. Unlike channels and locks, wait honours with_deadline. A condition carries no value and remembers nothing: a notify with nobody waiting is lost, so wait in a loop that checks the state itself.

Io :: #import "Extensions/Io";

changed: Io.Condition;
ready: bool;

main :: () {
    Io.io_run(() {
        error: Io.Io_Error;

        // Nothing changes `ready` here, so the wait ends at the deadline.
        Io.with_deadline(20, #code {
            while !ready {
                error = Io.wait(*changed);           // .TIMED_OUT, .CANCELED
                if error != .NONE break;
            }
        });

        assert(error == .TIMED_OUT);
    });
}

The IP version of an Address: IPV4, IPV6, or NONE for an Address that was never filled in (and for what parse_address returns when the text is not an address).

Io :: #import "Extensions/Io";

address, ok := parse_address("[::1]:8080");
assert(ok && address.family == .IPV6 && address.port == 8080);

address, ok = parse_address("127.0.0.1:80");
assert(ok && address.family == .IPV4);

address, ok = parse_address("not an address");
assert(!ok && address.family == .NONE);

An IP address and port. bytes is in network order; an IPv4 address uses the first four.

A listening socket.

A connected socket, from accept or connect.

One direction of a pipe.

One end of a pipe made by make_pipe.

How open_file opens a file.

An open file. read and write use and advance position; read_at and write_at do not. It has a position, so every procedure takes it by pointer.

Reads what is there, up to buffer.count bytes, waiting for at least one: 0 means the end.

Writes at least one byte of bytes (all of it for a sink that never waits) and returns how many.

Waits until read would not have to wait: a byte is there, the source ended, or it failed (the read that follows says which). Nothing is consumed, so it needs no buffer.

Closes the source. now asks for a close that does not wait, for code that cannot block.

A source of bytes: a pointer to its data and the procedures to use on it. reader_of makes one for a file, a TCP stream or a pipe end, and a struct of your own can fill in read_proc (and optionally close_proc and wait_proc). A Reader is a handle: copy it freely, it owns nothing, and the source must stay alive while it is used. close on it closes the source.

Counter :: struct {
    next: u8;
}

count_up :: (data: *void, buffer: [] u8) -> s64, Io_Error {
    counter := cast(*Counter) data;

    for 0..buffer.count - 1 {
        buffer[it] = counter.next;
        counter.next += 1;
    }

    return buffer.count, .NONE;
}

main :: () {
    counter: Counter;
    source := Reader.{ data = *counter, read_proc = count_up };

    bytes: [4] u8;
    count, error := read(source, bytes);

    assert(error == .NONE && count == 4);
    assert(bytes[0] == 0 && bytes[3] == 3);
}

A sink for bytes, the counterpart of Reader: a pointer to its data, a write_proc and an optional close_proc. writer_of makes one for a file, a TCP stream or a pipe end. It is a handle that owns nothing; close on it closes the sink.

Reads lines from source. Make one with make_line_reader, read with read_line and give its buffer back with free_line_reader. It does not close source.

The system facility a scheduler waits on: IO_URING or EPOLL on Linux, KQUEUE on macOS, IOCP or IORING on Windows. The scheduler picks the best one the machine offers; the JAI_IO_BACKEND environment variable names another (io_uring, epoll, kqueue, iocp, ioring). NONE is a system with no IO, where every operation fails with .UNSUPPORTED. Behaviour seen through the API is the same on all of them; they differ in speed and in the kernel versions they need.

What the engine found it could use. Backends fill in what applies.

What set_scheduler_local is given to release a value when its scheduler ends. It runs on the scheduler's thread after every fiber has ended, so it must not block: close sockets with close(stream, now = true).

One stored value of set_scheduler_local; the scheduler keeps a list of them.

One listening socket's multishot accept. A fiber in accept waits on it; connections that arrive while nobody waits are queued for the next call.

The answer a parked accept is waiting for; lives on that fiber's stack.

What the Windows engine keeps. The fields exist everywhere so that Engine has one shape.

IORING_HANDLE_REF and IORING_BUFFER_REF: a kind (0: a raw handle or address) and the value.