c-asio

A memory safe focus C framework, combining c-raii, libuv, coroutine and other concurrency primitives.

Table of Contents

• Introduction
• Design
  • API layout
• Synopsis
• Usage
• Installation
• Contributing
• License

Introduction

Attain the behavior of C++ boost.cobalt and boost.asio without the overhead.

This library provides ease of use convenience wrappers for libuv combined with the power of c-raii, a high level memory management library similar to other languages, among other features. Like coroutine support, the otherwise needed callback, is now automatically back to the caller with results.

• All functions requiring allocation and passing pointers, now returns them instead, if needed.
• The general naming convention is to drop uv_ prefix and require only necessary arguments/options.
• This integration also requires the use of uv_main(int argc, char **argv) as the startup entry routine:

libuv example from https://github.com/libuv/libuv/tree/master/docs/code/

#include "asio.h"

int uv_main(int argc, char **argv) {
    printf("Now quitting.\n");
    yield();

    return coro_err_code();
}

#include <stdio.h>
#include <stdlib.h>
#include <uv.h>

int main() {
    uv_loop_t *loop = malloc(sizeof(uv_loop_t));
    uv_loop_init(loop);

    printf("Now quitting.\n");
    uv_run(loop, UV_RUN_DEFAULT);

    uv_loop_close(loop);
    free(loop);
    return 0;
}

This general means there will be a dramatic reduction of lines of code repeated, repeatedly.

Libuv guides/examples:

• Reading/Writing files as in uvcat/main.c - 62 line script.
• Buffers and Streams as in uvtee/main.c - 79 line script.
• Querying DNS as in dns/main.c - 80 line script.
• Spawning child processes as in spawn/main.c - 36 line script.
• Networking/TCP as in tcp-echo-server/main.c - 87 line script.

Reduced to:

#include "asio.h"

int uv_main(int argc, char **argv) {
    uv_file fd = fs_open(argv[1], O_RDONLY, 0);
    if (fd > 0) {
        string text = fs_read(fd, -1);
        fs_write(STDOUT_FILENO, text, -1);

        return fs_close(fd);
    }

    return fd;
}

#include "asio.h"

int uv_main(int argc, char **argv) {
    string text = nullptr;
    uv_file fd = fs_open(argv[1], O_CREAT | O_RDWR, 0644);
    if (fd > 0) {
        pipe_file_t *file_pipe = pipe_file(fd, false);
        pipe_out_t *stdout_pipe = pipe_stdout(false);
        pipe_in_t *stdin_pipe = pipe_stdin(false);
        while (text = stream_read(stdin_pipe->reader)) {
            if (stream_write(stdout_pipe->writer, text)
                || stream_write(file_pipe->handle, text))
                break;
        }

        return fs_close(fd);
    }

    return fd;
}

#include "asio.h"

int uv_main(int argc, char **argv) {
    string text = nullptr;
    fprintf(stderr, "irc.libera.chat is...\033[0K\n");
    dnsinfo_t *dns = get_addrinfo("irc.libera.chat", "6667", 3,
        kv(ai_flags, AF_UNSPEC),
        kv(ai_socktype, SOCK_STREAM),
        kv(ai_protocol, IPPROTO_TCP)
    );

    fprintf(stderr, "%s\033[0K\n", addrinfo_ip(dns));
    uv_stream_t *server = stream_connect_ex(UV_TCP, addrinfo_ip(dns), "irc.libera.chat", 6667);
    while (text = stream_read(server))
        fprintf(stderr, "\033[0K%s", text);

    return coro_err_code();
}

#include "asio.h"

void _on_exit(int64_t exit_status, int term_signal) {
    fprintf(stderr, "\nProcess exited with status %" PRId64 ", signal %d\n",
            exit_status, term_signal);
}

int uv_main(int argc, char **argv) {
    spawn_t child = spawn("mkdir", "test-dir", nullptr);
    if (!spawn_atexit(child, _on_exit))
        fprintf(stderr, "\nLaunched process with ID %d\n", spawn_pid(child));

    return coro_err_code();
}

#include "asio.h"

#define DEFAULT_PORT 7000
#define DEFAULT_BACKLOG 128

void new_connection(uv_stream_t *socket) {
    string data = stream_read(socket);
    stream_write(socket, data);
}

int uv_main(int argc, char **argv) {
    uv_stream_t *client, *server;
    char addr[UV_MAXHOSTNAMESIZE] = nil;

    if (snprintf(addr, sizeof(addr), "0.0.0.0:%d", DEFAULT_PORT)) {
        server = stream_bind(addr, 0);
        while (server) {
            if (is_empty(client = stream_listen(server, DEFAULT_BACKLOG))) {
                fprintf(stderr, "Listen error %s\n", uv_strerror(coro_err_code()));
                break;
            }

            stream_handler(new_connection, client);
        }
    }

    return coro_err_code();
}

See branches for previous setup, main is an complete makeover of previous implementation approaches.

Similar approach has been made for C++20, an implementation in uvco. The tests presented there currently being reimplemented for C89 here, this project will be considered stable when completed and passing. And another approach in libasync mixing libco with libuv. Both approaches are Linux only.

Design

The intergration pattern for all libuv functions taking callback is as waitgroup_work.c example:

#define USE_CORO
#include "raii.h"

#define FIB_UNTIL 25

long fib_(long t) {
    if (t == 0 || t == 1)
        return 1;
    else
        return fib_(t-1) + fib_(t-2);
}

void_t fib(params_t req) {
    int n = req->integer;
    if (random() % 2)
        sleepfor(1);
    else
        sleepfor(3);

    long fib = fib_(n);
    fprintf(stderr, "%dth fibonacci is %lu in thrd: #%d\033[0K\n", n, fib, coro_thrd_id());

    return casting(fib);
}

void after_fib(int status, rid_t id) {
    fprintf(stderr, "Done calculating %dth fibonacci, result: %d\n", status, result_for(id).integer);
}

int main(int argc, char **argv) {
    rid_t data[FIB_UNTIL];
    int i;

    waitgroup_t wg = waitgroup_ex(FIB_UNTIL);
    for (i = 0; i < FIB_UNTIL; i++) {
        data[i] = go(fib, 1, casting(i));
    }
    waitresult_t wgr = waitfor(wg);

    if ($size(wgr) == FIB_UNTIL)
        for (i = 0; i < FIB_UNTIL; i++) {
            after_fib(i, data[i]);
        }

    return 0;
}

Every system thread has a run queue assigned, a tempararay FIFO queue, it holds coroutine tasks. This assignment is based on system cpu cores available, and set at startup, a coroutine thread pool set before uv_main is called.

When a go/coroutine is created, it's given a index result id from global array like struct, a coroutine id, and thread id. Then placed into a global run queue, a hashtable, the key being result id, for schedularing. The thread id determines which thread pool coroutine gets assigned to.

These three data structures are atomically accessable by all threads.

• The main thread determines and move coroutines from global queue to each thread queue.
• Each thread's scheduler manages it's own local run queue of coroutines by thread local storage.
  • It takes coroutine tasks from it's tempararay FIFO queue to local storage.
• Each coroutine is self containing, can be assigned to any thread, at any point within a yield execution.

All libuv functions outlined/api, is as example, but having a waitgroup of one. Demonstrating true libuv/Coroutine multi threading is disabled by how current tests and examples startup.

If the number of coroutines created before the first yield encountered, does not equal cpu core count plus one. Then main thread will move all coroutines to itself, set each coroutine thread id to itself, and mark whole system feature disabled.

• Codebase will need current libuv function wrapper implementations to have all arguments pass into coroutine creation, right now the initaliaztion process using a uv_loop_t thread handle of wrong coroutine thread pool, disabling is a tempararay fix.

The approach outlined and things still to be worked out, is as Go The Scheduler Saga, Queues, Fairness, and The Go Scheduler and Rust Making the Tokio scheduler 10x faster.

API

The documentation at boost.cobalt is a good staring point. The principles still apply, just done automatically, with some naming differences hereforth. Like boost.asio io_context is similar to uv_loop_t in libuv and others in:

• Coroutine Patterns: Problems and Solutions Using Coroutines in a Modern Codebase YouTube
• Introduction to C++ Coroutines Through a Thread Scheduling Demonstration YouTube
• C++ Coroutine Intuition YouTube
• Asynchrony with ASIO and coroutines YouTube

/* This library provides its own ~main~,
which call this function as an coroutine! */
C_API int uv_main(int, char **);

C_API uv_loop_t *asio_loop(void);

/* Same as `sleepfor`, but uses `uv_timer_start` to explicitly give up the current `coroutine`
for at least `ms` milliseconds. Other tasks continue to run during this ti