🟨 Async I/O (files, sockets, timers, processes, fs events, threads wake-up)

Async is a multi-platform / event-driven asynchronous I/O library.

It exposes async programming model for common IO operations like reading / writing to / from a file or tcp socket.

Synchronous I/O operations could block the current thread of execution for an undefined amount of time, making it difficult to scale an application to a large number of concurrent operations, or to coexist with other even loop, like for example a GUI event loop. Such async programming model uses a common pattern, where the call fills an AsyncRequest with the required data. The AsyncRequest is added to an AsyncEventLoop that will queue the request to some low level OS IO queue. The event loop can then monitor all the requests in a single call to SC::AsyncEventLoop::run, SC::AsyncEventLoop::runOnce or SC::AsyncEventLoop::runNoWait. These three different run methods cover different integration use cases of the event loop inside of an applications.

The kernel Async API used on each operating systems are the following:

IOCP on Windows
kqueue on macOS
epoll on Linux
io_uring on Linux (dynamically loading liburing)

Note: If liburing is not available on the system, the library will transparently fallback to epoll.

If an async operation is not supported by the OS, the caller can provide a SC::ThreadPool to run it on a thread. See SC::AsyncFileRead / SC::AsyncFileWrite for an example.

Features

This is the list of supported async operations:

Async Operation	Description
AsyncSocketConnect	Starts a socket connect operation, connecting to a remote endpoint.
AsyncSocketAccept	Starts a socket accept operation, obtaining a new socket from a listening socket.
AsyncSocketSend	Starts a socket send operation, sending bytes to a remote endpoint.
AsyncSocketReceive	Starts a socket receive operation, receiving bytes from a remote endpoint.
AsyncSocketClose	Starts a socket close operation.
AsyncFileRead	Starts a file read operation, reading bytes from a file (or pipe).
AsyncFileWrite	Starts a file write operation, writing bytes to a file (or pipe).
AsyncFileClose	Starts a file close operation, closing the OS file descriptor.
AsyncLoopTimeout	Starts a Timeout that is invoked after expiration (relative) time has passed.
AsyncLoopWakeUp	Starts a wake-up operation, allowing threads to execute callbacks on loop thread.
AsyncLoopWork	Executes work in a thread pool and then invokes a callback on the event loop thread.
AsyncProcessExit	Starts monitoring a process, notifying about its termination.
AsyncFilePoll	Starts an handle polling operation.

Status

🟨 MVP
This is usable but needs some more testing and a few more features.

Description

An async operation is struct derived from AsyncRequest asking for some I/O to be done made to the OS.
Every async operation has an associated callback that is invoked when the request is fulfilled. If the start function returns a valid (non error) Return code, then the user callback will be called both in case of success and in case of any error.
If the function returns an invalid Return code or if the operation is manually cancelled with SC::AsyncRequest::stop, then the user callback will not be called.

Note: The memory address of all AsyncRequest derived objects must be stable for the entire duration of a started async request. This means that they can be freed / moved after the user callback is executed.

Some implementation details: SC::AsyncRequest::state dictates the lifetime of the async request according to a state machine.

Regular Lifetime of an Async request (called just async in the paragraph below):

An async that has been started, will be pushed in the submission queue with state == State::Setup.
Inside stageSubmission a started async will be do the one time setup (with setupAsync)
Inside stageSubmission a Setup or Submitting async will be activated (with activateAsync)
If activateAsync is successful, the async becomes state == State::Active.
- When this happens, the async is either tracked by the kernel or in one of the linked lists like activeLoopWakeUps
The Active async can become completed, when the kernel signals its completion (or readiness...):
- [default] -> Async is complete and it will be teardown and freed (state == State::Free)
- result.reactivateRequest(true) -> Async gets submitted again (state == State::Submitting) (3.)

Cancellation of an async: An async can be cancelled at any time:

Async not yet submitted in State::Setup --> it just gets removed from the submission queue
Async in submission queue but already setup --> it will receive a teardownAsync
Async in Active state (so after setupAsync and activateAsync) --> will receive cancelAsync and teardownAsync

Any other case is considered an error (trying to cancel an async already being cancelled or being teardown).

AsyncEventLoop

Asynchronous I/O (files, sockets, timers, processes, fs events, threads wake-up) (see Async) AsyncEventLoop pushes all AsyncRequest derived classes to I/O queues in the OS. Basic lifetime for an event loop is:

AsyncEventLoop eventLoop;
SC_TRY(eventLoop.create()); // Create OS specific queue handles
// ...
// Add all needed AsyncRequest
// ...
SC_TRY(eventLoop.run());
// ...
// Here all AsyncRequest have either finished or have been stopped
// ...
SC_TRY(eventLoop.close()); // Free OS specific queue handles

Run modes

Event loop can be run in different ways to allow integrated it in multiple ways in applications that may already have an existing event loop (example GUI applications).

Run mode	Description
SC::AsyncEventLoop::run	Blocks until there are no more active queued requests. It's useful for applications where the eventLoop is the only (or the main) loop. One example could be a console based app doing socket IO or a web server. Waiting on requests blocks the current thread with 0% CPU utilization.
SC::AsyncEventLoop::runOnce	Blocks until at least one request proceeds, ensuring forward progress. It's useful for applications where the eventLoop events needs to be interleaved with other work. For example one possible way of integrating with a UI event loop could be to schedule a recurrent timeout timer every 1/60 seconds where calling GUI event loop updates every 60 seconds, blocking for I/O for the remaining time. Waiting on requests blocks the current thread with 0% CPU utilization.
SC::AsyncEventLoop::runNoWait	Process active requests if they exist or returns immediately without blocking. It's useful for game-like applications where the event loop runs every frame and one would like to check and dispatch its I/O callbacks in-between frames. This call allows poll-checking I/O without blocking.

AsyncLoopTimeout

Starts a Timeout that is invoked after expiration (relative) time has passed.

// Create a timeout that will be called after 200 milliseconds
// AsyncLoopTimeout must be valid until callback is called
AsyncLoopTimeout timeout;
timeout.callback = [&](AsyncLoopTimeout::Result&)
{
    console.print("My timeout has been called!");
};
// Start the timeout, that will be called 200 ms from now
SC_TRY(timeout.start(eventLoop, Time::Milliseconds(200)));

AsyncLoopWakeUp

Starts a wake-up operation, allowing threads to execute callbacks on loop thread.
SC::AsyncLoopWakeUp::callback will be invoked on the thread running SC::AsyncEventLoop::run (or its variations) after SC::AsyncLoopWakeUp::wakeUp has been called.

Note: There is no guarantee that after calling AsyncLoopWakeUp::start the callback has actually finished execution. An optional SC::EventObject passed to SC::AsyncLoopWakeUp::start can be used for synchronization

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
// This code runs on some different thread from the one calling SC::AsyncEventLoop::run.
// The callback is invoked from the thread calling SC::AsyncEventLoop::run
AsyncLoopWakeUp wakeUp; // Memory lifetime must be valid until callback is called
wakeUp.callback = [&](AsyncLoopWakeUp::Result&)
{
    console.print("My wakeUp has been called!");
};
SC_TRY(wakeUp.start(eventLoop));

An EventObject can be wait-ed to synchronize further actions from the thread invoking the wake up request, ensuring that the callback has finished its execution.

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
// This code runs on some different thread from the one calling SC::AsyncEventLoop::run.
// The callback is invoked from the thread calling SC::AsyncEventLoop::run
AsyncLoopWakeUp wakeUpWaiting; // Memory lifetime must be valid until callback is called
wakeUpWaiting.callback = [&](AsyncLoopWakeUp::Result&)
{
    console.print("My wakeUp has been called!");
};
EventObject eventObject;
SC_TRY(wakeUpWaiting.start(eventLoop, &eventObject));
eventObject.wait(); // Wait until callback has been fully run inside event loop thread
// From here on we know for sure that callback has been called

AsyncLoopWork

Executes work in a thread pool and then invokes a callback on the event loop thread.
AsyncLoopWork::work is invoked on one of the thread supplied by the ThreadPool passed during AsyncLoopWork::start. AsyncLoopWork::callback will be called as a completion, on the event loop thread AFTER work callback is finished.

    // This test creates a thread pool with 4 thread and 16 AsyncLoopWork.
    // All the 16 AsyncLoopWork are scheduled to do some work on a background thread.
    // After work is done, their respective after-work callback is invoked on the event loop thread.
 
    static constexpr int NUM_THREADS = 4;
    static constexpr int NUM_WORKS   = NUM_THREADS * NUM_THREADS;
 
    ThreadPool threadPool;
    SC_TEST_EXPECT(threadPool.create(NUM_THREADS));
 
    AsyncEventLoop eventLoop;
    SC_TEST_EXPECT(eventLoop.create());
 
    AsyncLoopWork works[NUM_WORKS];
 
    int         numAfterWorkCallbackCalls = 0;
    Atomic<int> numWorkCallbackCalls      = 0;
 
    for (int idx = 0; idx < NUM_WORKS; ++idx)
    {
        works[idx].work = [&]
        {
            // This work callback is called on some random threadPool thread
            Thread::Sleep(50);                 // Execute some work on the thread
            numWorkCallbackCalls.fetch_add(1); // Atomically increment this counter
            return Result(true);
        };
        works[idx].callback = [&](AsyncLoopWork::Result&)
        {
            // This after-work callback is invoked on the event loop thread.
            // More precisely this runs on the thread calling eventLoop.run().
            numAfterWorkCallbackCalls++; // No need for atomics here, callback is run inside loop thread
        };
        SC_TEST_EXPECT(works[idx].start(eventLoop, threadPool));
    }
    SC_TEST_EXPECT(eventLoop.run());
 
    // Check that callbacks have been actually called
    SC_TEST_EXPECT(numWorkCallbackCalls.load() == NUM_WORKS);
    SC_TEST_EXPECT(numAfterWorkCallbackCalls == NUM_WORKS);

AsyncProcessExit

Starts monitoring a process, notifying about its termination. Process library can be used to start a process and obtain the native process handle.

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
Process process;
SC_TRY(process.launch({"executable", "--parameter"}));
ProcessDescriptor::Handle processHandle;
SC_TRY(process.handle.get(processHandle, Result::Error("Invalid Handle")));
AsyncProcessExit processExit; //  Memory lifetime must be valid until callback is called
processExit.callback = [&](AsyncProcessExit::Result& res)
{
    ProcessDescriptor::ExitStatus exitStatus;
    if(res.get(exitStatus))
    {
        console.print("Process Exit status = {}", exitStatus.status);
    }
};
SC_TRY(processExit.start(eventLoop, processHandle));

AsyncSocketAccept

Starts a socket accept operation, obtaining a new socket from a listening socket.
The callback is called with a new socket connected to the given listening endpoint will be returned.
Socket library can be used to create a Socket but the socket should be created with SC::SocketFlags::NonBlocking and associated to the event loop with SC::AsyncEventLoop::associateExternallyCreatedTCPSocket.
Alternatively SC::AsyncEventLoop::createAsyncTCPSocket creates and associates the socket to the loop.

Note: To continue accepting new socket SC::AsyncResult::reactivateRequest must be called.

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
// Create a listening socket
constexpr uint32_t numWaitingConnections = 2;
SocketDescriptor   serverSocket;
uint16_t           tcpPort = 5050;
SocketIPAddress    nativeAddress;
SC_TRY(nativeAddress.fromAddressPort("127.0.0.1", tcpPort));
SC_TRY(eventLoop.createAsyncTCPSocket(nativeAddress.getAddressFamily(), serverSocket));
SC_TRY(SocketServer(serverSocket).listen(nativeAddress, numWaitingConnections));
// Accept connect for new clients
AsyncSocketAccept accept;
accept.callback = [&](AsyncSocketAccept::Result& res)
{
    SocketDescriptor client;
    if(res.moveTo(client))
    {
        // ...do something with new client
        console.printLine("New client connected!");
        res.reactivateRequest(true); // We want to receive more clients
    }
};
SC_TRY(accept.start(eventLoop, serverSocket));
// ... at some later point
// Stop accepting new clients
SC_TRY(accept.stop());

AsyncSocketConnect

Starts a socket connect operation, connecting to a remote endpoint.
Callback will be called when the given socket is connected to ipAddress.
Socket library can be used to create a Socket but the socket should be created with SC::SocketFlags::NonBlocking and associated to the event loop with SC::AsyncEventLoop::associateExternallyCreatedTCPSocket.
Alternatively SC::AsyncEventLoop::createAsyncTCPSocket creates and associates the socket to the loop.

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
SocketIPAddress localHost;
SC_TRY(localHost.fromAddressPort("127.0.0.1", 5050)); // Connect to some host and port
AsyncSocketConnect connect;
SocketDescriptor   client;
SC_TRY(eventLoop.createAsyncTCPSocket(localHost.getAddressFamily(), client));
connect.callback = [&](AsyncSocketConnect::Result& res)
{
    if (res.isValid())
    {
        // Do something with client that is now connected
        console.printLine("Client connected");
    }
};
SC_TRY(connect.start(eventLoop, client, localHost));

AsyncSocketSend

Starts a socket send operation, sending bytes to a remote endpoint. Callback will be called when the given socket is ready to send more data.
Socket library can be used to create a Socket but the socket should be created with SC::SocketFlags::NonBlocking and associated to the event loop with SC::AsyncEventLoop::associateExternallyCreatedTCPSocket or though AsyncSocketAccept.
Alternatively SC::AsyncEventLoop::createAsyncTCPSocket creates and associates the socket to the loop.

// Assuming an already created (and running) AsyncEventLoop named `eventLoop`
// and a connected or accepted socket named `client`
// ...
const char sendBuffer[] = {123, 111};
 
// The memory pointed by the span must be valid until callback is called
Span<const char> sendData = {sendBuffer, sizeof(sendBuffer)};
 
AsyncSocketSend sendAsync;
sendAsync.callback = [&](AsyncSocketSend::Result& res)
{
    if(res.isValid())
    {
        // Now we could free the data pointed by span and queue new data
        console.printLine("Ready to send more data");
    }
};
 
SC_TRY(sendAsync.start(eventLoop, client, sendData));

AsyncSocketReceive

Starts a socket receive operation, receiving bytes from a remote endpoint. Callback will be called when some data is read from socket.
Socket library can be used to create a Socket but the socket should be created with SC::SocketFlags::NonBlocking and associated to the event loop with SC::AsyncEventLoop::associateExternallyCreatedTCPSocket or though AsyncSocketAccept.
Alternatively SC::AsyncEventLoop::createAsyncTCPSocket creates and associates the socket to the loop.

// Assuming an already created (and running) AsyncEventLoop named `eventLoop`
// and a connected or accepted socket named `client`
// ...
AsyncSocketReceive receiveAsync;
char receivedData[100] = {0};
receiveAsync.callback = [&](AsyncSocketReceive::Result& res)
{
    Span<char> readData;
    if(res.get(readData))
    {
        // readData now contains a slice of receivedData with the received bytes
        console.print("{} bytes have been read", readData.sizeInBytes());
    }
    // Ask to reactivate the request if we want to receive more data
    res.reactivateRequest(true);
};
SC_TRY(receiveAsync.start(eventLoop, client, {receivedData, sizeof(receivedData)}));

AsyncSocketClose

Starts a socket close operation. Callback will be called when the socket has been fully closed.

// Assuming an already created (and running) AsyncEventLoop named `eventLoop`
// and a connected or accepted socket named `client`
// ...
AsyncSocketClose asyncClose;
 
asyncClose.callback = [&](AsyncSocketClose::Result& result)
{
    if(result.isValid())
    {
        console.printLine("Socket was closed successfully");
    }
};
SC_TRY(asyncClose.start(eventLoop, client));
 

AsyncFileRead

Starts a file read operation, reading bytes from a file (or pipe). Callback will be called when the data read from the file (or pipe) is available.
Use the start overload with ThreadPool / Task parameters to execute file read on a background thread. This is important on APIs with blocking behaviour on buffered file I/O (all apis with the exception of io_uring).

File library can be used to open the file and obtain a file (or pipe) descriptor handle.

Note: Pipes or files opened using Posix O_DIRECT or Windows FILE_FLAG_WRITE_THROUGH & FILE_FLAG_NO_BUFFERING should instead avoid using the Task parameter for best performance.

When not using the Task remember to:

Open the file descriptor for non-blocking IO (SC::FileDescriptor::OpenOptions::blocking == false)
Call SC::AsyncEventLoop::associateExternallyCreatedFileDescriptor on the file descriptor

// Assuming an already created (and running) AsyncEventLoop named `eventLoop`
// ...
 
// Assuming an already created threadPool named `eventLoop
// ...
 
// Open the file
FileDescriptor fd;
FileDescriptor::OpenOptions options;
options.blocking = true; // AsyncFileRead::Task enables using regular blocking file descriptors
SC_TRY(fd.open("MyFile.txt", FileDescriptor::ReadOnly, options));
 
// Create the async file read request and async task
AsyncFileRead asyncReadFile;
asyncReadFile.callback = [&](AsyncFileRead::Result& res)
{
    Span<char> readData;
    if(res.get(readData))
    {
        console.print("Read {} bytes from file", readData.sizeInBytes());
        res.reactivateRequest(true); // Ask to receive more data
    }
};
char buffer[100] = {0};
asyncReadFile.buffer = {buffer, sizeof(buffer)};
// Obtain file descriptor handle and associate it with event loop
SC_TRY(fd.get(asyncReadFile.fileDescriptor, Result::Error("Invalid handle")));
 
// Start the operation on a thread pool
AsyncFileRead::Task asyncFileTask;
SC_TRY(asyncReadFile.start(eventLoop, threadPool, asyncFileTask));
 
// Alternatively if the file is opened with blocking == false, AsyncFileRead can be omitted
// but the operation will not be fully async on regular (buffered) files, except on io_uring.
//
// SC_TRY(asyncReadFile.start(eventLoop));

AsyncFileWrite

Starts a file write operation, writing bytes to a file (or pipe). Callback will be called when the file is ready to receive more bytes to write.
Use the start overload with ThreadPool / Task parameters to execute file read on a background thread. This is important on APIs with blocking behaviour on buffered file I/O (all apis with the exception of io_uring).

File library can be used to open the file and obtain a blocking or non-blocking file descriptor handle.

Note: Pipes or files opened using Posix O_DIRECT or Windows FILE_FLAG_WRITE_THROUGH & FILE_FLAG_NO_BUFFERING should instead avoid using the Task parameter for best performance.

When not using the Task remember to:

Open the file descriptor for non-blocking IO (SC::FileDescriptor::OpenOptions::blocking == false)
Call SC::AsyncEventLoop::associateExternallyCreatedFileDescriptor on the file descriptor

// Assuming an already created (and running) AsyncEventLoop named `eventLoop`
// ...
 
// Assuming an already created threadPool named `threadPool`
// ...
 
// Open the file (for write)
FileDescriptor::OpenOptions options;
options.blocking = true; // AsyncFileWrite::Task enables using regular blocking file descriptors
FileDescriptor fd;
SC_TRY(fd.open("MyFile.txt", FileDescriptor::WriteCreateTruncate, options));
 
// Create the async file write request
AsyncFileWrite asyncWriteFile;
asyncWriteFile.callback = [&](AsyncFileWrite::Result& res)
{
    size_t writtenBytes = 0;
    if(res.get(writtenBytes))
    {
        console.print("{} bytes have been written", writtenBytes);
    }
};
// Obtain file descriptor handle
SC_TRY(fd.get(asyncWriteFile.fileDescriptor, Result::Error("Invalid Handle")));
asyncWriteFile.buffer = StringView("test").toCharSpan();;
 
// Start the operation in a thread pool
AsyncFileWrite::Task asyncFileTask;
SC_TRY(asyncWriteFile.start(eventLoop, threadPool, asyncFileTask));
 
// Alternatively if the file is opened with blocking == false, AsyncFileRead can be omitted
// but the operation will not be fully async on regular (buffered) files, except on io_uring.
//
// SC_TRY(asyncWriteFile.start(eventLoop));

AsyncFileClose

Starts a file close operation, closing the OS file descriptor. Callback will be called when the file is actually closed.
File library can be used to open the file and obtain a blocking or non-blocking file descriptor handle.

// Assuming an already created (and running) AsyncEventLoop named eventLoop
// ...
 
// Open a file and associated it with event loop
FileDescriptor fd;
FileDescriptor::OpenOptions options;
options.blocking = false;
SC_TRY(fd.open("MyFile.txt", FileDescriptor::WriteCreateTruncate, options));
SC_TRY(eventLoop.associateExternallyCreatedFileDescriptor(fd));
 
// Create the file close request
FileDescriptor::Handle handle;
SC_TRY(fd.get(handle, Result::Error("handle")));
AsyncFileClose asyncFileClose;
asyncFileClose.callback = [&](AsyncFileClose::Result& result)
{
    if(result.isValid())
    {
        console.printLine("File was closed successfully");
    }
};
SC_TRY(asyncFileClose.start(eventLoop, handle));

AsyncFilePoll

Starts an handle polling operation. Uses GetOverlappedResult (windows), kevent (macOS), epoll (Linux) and io_uring (Linux). Callback will be called when any of the three API signals readiness events on the given file descriptor. Check File System Watcher for an example usage of this notification.

Implementation

Library abstracts async operations by exposing a completion based mechanism. This mechanism currently maps on kqueue on macOS and OVERLAPPED on Windows.

It currently tries to dynamically load io_uring on Linux doing an epoll backend fallback in case liburing is not available on the system. There is not need to link liburing because the library loads it dynamically and embeds the minimal set of static inline functions needed to interface with it.

The api works on file and socket descriptors, that can be obtained from the File and Socket libraries.

Memory allocation

The entire library is free of allocations, as it uses a double linked list inside SC::AsyncRequest.
Caller is responsible for keeping AsyncRequest-derived objects memory stable until async callback is called.
SC::ArenaMap from the Containers can be used to preallocate a bounded pool of Async objects.

Roadmap

🟩 Usable Features:

Implement option to do blocking poll check without dispatching callbacks (needed for efficient gui event loop integration)
More comprehensive test suite, testing all cancellations
FS operations (open stat read write unlink copyfile mkdir chmod etc.)
UDP Send/Receive
DNS Resolution

🟦 Complete Features:

TTY with ANSI Escape Codes

💡 Unplanned Features:

Signal handling

Table of Contents