DEV Community: Szymon Gibała

Async without async

Szymon Gibała — Sun, 29 Mar 2026 06:19:09 +0000

To start with a short TLDR: this article is my exploration of
implementing an asynchronous networking application, without using
async Rust.

Background

Over the past months (if not years at this point) I have been playing
around with some sane approaches of implementing consensus algorithms,
and perhaps more general distributed systems.

As part of this journey I am seeking ways to have more control over
the whole "application framework". This recently led me to ask a
question: Can I have a performant, IO heavy application without using
async Rust?

Why not async?

In many ways async support in Rust is great. If you are just writing a
web application, the async and await keywords really make it very easy
to write the code as you would with sync Rust. However, everything
comes at a cost - async Rust brings in the complexity of the whole
async runtime, and hides a lot of what is going on from our sight.

One of the reasons for it is that Async Rust and accompanying runtimes
are built to be a generalized solution, to support a lot of different
cases and be robust in many different ways. To be able to do this some
complexity arises naturally, which is then well hidden from us by async
and friends. Not all this complexity is needed for every use case,
and since there is no such thing as a free lunch, there may come the
time to pay for it.

Another thing we sacrifice by using an async runtime is control.
It is programmed in a specific way, with some knobs that we are able
to tweak and some we are not. Until we understand the code thoroughly
and grasp the possible code paths, there will always be a black box
aspect to it.

What I have learned over the years is that sometimes it is better to
ditch a one-size-fits-all, batteries-included solution, and build
something simpler, use-case-specific, sacrificing some time but sparing
yourself a lot of complexity, and retaining full control and
better understanding of the system.

Part of this exploration is to answer the question whether it is worth
it in this case.

Objective

Not using async Rust is not a goal in itself, but only a means to
an end. The main objective remains to build a proof of concept of a
simple system that could be used to implement more sophisticated
software on top of it.

The goal is exploration, but there are a few constraints I want to
satisfy.

Keep it simple
- The foundation needs to be simple, easy to reason about, troubleshoot and understand. Let the complexity arise from the problems that applications on top of it will be solving, not from its fundamental parts. Given that I want it to be single threaded, at least for as long as it is not a performance limitation.
Keep it real
- This application aims to be proof of concept, of something that could be turned into a functional system. For me this implies:
  - No busy waiting - I do not want to burn the CPU when nothing is going on.
  - No added latency - When IO is ready, it should be processed, not wait until a few milliseconds sleep between loop iterations finishes.
Not just request trigger
- It is not a REST API I want to use it for, therefore applications built on top need to have a way of "triggering" some logic not only when a request arrives. To be more specific here, I am thinking of time based triggers, be it intervals or timeouts, there needs to be a way to run some logic based on those, and not just incoming IO.

Since I am ditching async Rust, and IO is still at the core of the
application, the first step is to figure out how to handle it without
the magic of Tokio.
Let's take a look at the possibilities.

Handling IO

If I were asked to write the simplest echo server to handle just one
connection I would end up with something like this:

let listener = std::net::TcpListener::bind("127.0.0.1:9000")?;
let (mut connection, addr) = listener.accept()?;
loop {
    let mut buffer = [0u8; 1024];
    let bytes_read = connection.read(&mut buffer)?;
    if bytes_read == 0 {
        // Connection closed by client
        break;
    }
    connection.write_all(&buffer[..bytes_read])?;
}

And that is the first, likely simplest approach to handling IO -
blocking IO. The application will block on the connection.read call
until there is something to read.

Now if I would have to handle multiple connections there are a few
ways to extend it.

I suppose that the most intuitive one is to just handle each connection
in a separate thread and keep accepting in the main one:

let listener = std::net::TcpListener::bind("127.0.0.1:9000")?;
loop {
    let (mut connection, addr) = listener.accept()?;
    let _handle = std::thread::spawn(move || -> anyhow::Result<()> {
        loop {
            let mut buffer = [0u8; 1024];
            let bytes_read = connection.read(&mut buffer)?;
            if bytes_read == 0 {
                // Connection closed by client
                break;
            }
            connection.write_all(&buffer[..bytes_read])?;
        }
        Ok(())
    });
}

Clearly, this approach is more versatile than handling only one
connection, but it is also clear that this approach violates one of my
objectives -- being single threaded.

Note: a variation of this could be process per connection, which
is used by some systems. Still they often use async IO anyway.

The other option that we have allows us to keep our single thread, all
we need is making sockets non-blocking and adding a bit more code:

let mut connections = Vec::new();
let listener = std::net::TcpListener::bind("127.0.0.1:9000")?;
listener.set_nonblocking(true)?;
loop {
    // accept all pending connections
    for cnn_res in listener.incoming() {
        match cnn_res {
            Ok(connection) => {
                // set socket non-blocking
                connection.set_nonblocking(true)?;
                connections.push(connection);
            }
            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                // No more incoming connections at the moment
                break;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }
    // Go through all connections. Read, write, and retain
    // only not closed connections.
    let mut new_connections = Vec::new();
    for connection in connections.iter_mut() {
        let mut drop = false;
        let mut buffer = [0u8; 1024];
        loop {
            match connection.read(&mut buffer) {
                Ok(bytes_read) => {
                    if bytes_read == 0 {
                        // Connection closed by client
                        drop = true;
                        break;
                    }
                    // If we would like to do it properly, we should
                    // handle would block errors as well
                    connection.write_all(&buffer[..bytes_read])?;
                }
                Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    // No data to read at the moment
                    break
                }
                Err(e) => {
                    return Err(e.into());
                }
            }
        }
        if !drop {
            new_connections.push(connection.try_clone()?);
        }
    }
    connections = new_connections;
}

However, a careful observer can immediately see that this violates
another constraint, as the loop will just keep spinning burning all
the CPU cycles. We could avoid busy looping by adding a short sleep
between iterations, but that is added latency I want to avoid as
well.

With all of those out of the way, and not suitable, we need to do a
full circle and go back to async. Not necessarily async Rust but
async IO nonetheless.

Asynchronous IO (or not really)

As the name suggests, asynchronous IO is not synchronous.

But what it really means is a bit complicated.

Async IO can work in different ways, and I am not sure if there is
a real, correct definition of what is async and what is not. In
general, when we talk about async, it is understood as something that
happens "in the background" and there is some notification when
"things are ready".

Different systems work in different ways. With io_uring, IO
happens in kernel space and user space application receives
notification when the work is completed, while with epoll the
application still does the dirty work of IO syscalls, and just
receives the notification when there is progress to be made.

Deeper tangent:
I would say that async is in the eye of the beholder.
One can argue that epoll is not "real" async since the application
only receives the event and all work still happens synchronously
(in a non-blocking way, let's say).
However, if you go with this thinking then Rust tokio isn't really
async, since it is also the application that does the IO.
"But io_uring is a real async!" You may object. In its case it is not
the application that does IO, but the kernel itself. It must be a
true async then!
However, if you look at it from the perspective of the CPU (or even
the kernel), it all happens on the same silicon (perhaps on different
cores, but that is not for us to decide), so is it "async" after all?

Different operating systems have different APIs for async IO, to name
a few:

kqueue - MacOS, BSD
epoll and io_uring - Linux
IOCP - Windows

There are other, older mechanisms on Linux as well, such as poll and
select, but these days epoll is likely most prevalent, with io_uring
being the newest and slowly getting more adoption.

Since penguins dominate the server world, I focused on Linux
and took a deeper look into Epoll.

Epoll

I knew about epoll from the first time I asked myself a question
"But how does async, actually work?" that led me into the deep rabbit
hole of various kernel mechanisms, down to the realm of CPU interrupts
(if you never went there, I highly recommend that journey). However,
not being a C programmer, I never used it "directly".

Most of the "async" web dev libraries in all languages rely on it,
but hide it carefully under a few layers of abstraction, mainly because
they are meant to work on all OSes and not just on Linux. But, let's
get to the point...

Epoll as a whole
is an API in the Linux kernel that allows a process to register
interest in IO events for a set of file descriptors.

There are 4 syscalls listed under the epoll man page:

epoll_create
epoll_create1
epoll_ctl
epoll_wait

Names are somewhat self-explanatory, so I will not copy-paste
definitions from the
man page,
feel free to check it out on your own.

I will not leave you empty-handed, however, and give you a quick intuition of how things work:
Epoll is about events. Instead of constantly checking if there is
any IO to be done, the user space application receives "notification",
when there is "progress" to be made. Since behind this mechanism is the
kernel, while waiting for the events the waiting thread can "go to
sleep" and get woken up when the IO event arrives, hence not wasting
CPU cycles by spinning around checking all connections, and also not
adding latency with an actual sleep.

One might ask: how do we know when to stop reading or writing then?
Well, if you ask the socket politely it will tell you. As long as it is
a non-blocking socket, as async IO is usually used in conjunction
with those.

And by socket telling you what I mean is returning EAGAIN or
EWOULDBLOCK.

A look inside

For each epoll instance created in userspace there is an
eventpoll
allocated on the kernel side. It contains a red-black tree of
epitems
keyed by file pointer and file descriptor. When we register
interest, the new tree node is inserted, and a callback is added
to file descriptors' wait queue. This callback is where the magic
happens, as whenever we call epoll_wait our thread will be parked
(if no interests are ready), and it is this callback's job to wake
it up (if the interest mask matches). Additionally when this happens
the reference to epitem from the tree is inserted into
eventpoll's ready list.

Now, to the more interesting part: how to actually use it.
My goal here is to get a real glimpse of epoll in all its glory,
not covered by the compatibility layers and easy to use abstractions.

Fine, fine... Using libc is not the lowest one can go, but it is good
enough for today...

First things first, I need something that listens on a port and
accepts connections. No epoll magic here, no async IO, just good
old C:

#define _GNU_SOURCE
#include <netinet/in.h>
#include <stdio.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>

int main(void) {
    int socket_fd = socket(AF_INET, SOCK_STREAM, 0);   // create TCP socket
    if (socket_fd < 0) { perror("socket"); return 1; } // check return value

    int yes = 1;
    setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));

    struct sockaddr_in addr = {          // bind address
        .sin_family = AF_INET,           // IPv4
        .sin_port   = htons(9000),       // port 9000
        // Convert from host byte order to network byte order
        .sin_addr   = { htonl(INADDR_LOOPBACK) } // localhost
    };

    if (bind(socket_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) { 
        perror("bind"); return 1; 
    }

    if (listen(socket_fd, SOMAXCONN) < 0) { 
        perror("listen"); 
        return 1;
    }

    printf("Listening on 127.0.0.1:9000...\n");

    struct sockaddr_in clientaddr;
    socklen_t clientaddr_size = sizeof(clientaddr);
    int connection_fd = accept(socket_fd, (struct sockaddr*)&clientaddr, &clientaddr_size);
    if (connection_fd < 0) {
        perror("accept");
        return 1;
    }
    printf("accepted connection from %s:%d\n", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port));

    char buf[4096];

    for (;;) {
        ssize_t r = read(connection_fd, buf, sizeof buf); // read some
        if (r == 0) {      // peer closed
            printf("read 0 bytes, closing fd\n");
            close(connection_fd);
            break;
        } else if (r < 0) {
            perror("read");  // read error
            close(connection_fd);
            break;
        } else {            // got r bytes
            printf("read %zd bytes\n", r);
            ssize_t off = 0;
            while (off < r) { 
                // echo back, in a blocking way
                ssize_t w = send(connection_fd, buf + off, (size_t)(r - off), MSG_NOSIGNAL);
                if (w < 0) {
                    perror("send");  // write error
                    close(connection_fd);
                    break;
                }
                off += w;    // advance
            }
        }
    }

    return 0;
}

This will do the work, as a simple echo server. However, as an example
of single thread blocking IO (just in a different language), it can
only handle one connection at a time.

Since async IO only makes sense with non-blocking sockets, the first
step is to make the listening socket as such:

#include <fcntl.h>

static int set_nonblocking(int fd) {
    // F_GETFL - return (as the function result) the file access mode and
    // the file status flags; arg is ignored.
    int fl = fcntl(fd, F_GETFL, 0);
    if (fl == -1) return -1;
    // Preserve the current flags and add O_NONBLOCK.
    return fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

...
    int yes = 1;
    setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
    // new line - setting socket_fd as non-blocking
    if (set_nonblocking(socket_fd) < 0) { 
        perror("fcntl"); 
        return 1; 
    }
...

Now, this breaks the echo server, since accept will no longer block,
but return the error instead:

Listening on 127.0.0.1:9000...
accept4: Resource temporarily unavailable

Which basically means that: "if the socket was blocking, the call would
block". Now it is the time we need to create an epoll instance and
register read interest (EPOLLIN) for socket_fd on it:

#include <sys/epoll.h>
...
if (listen(socket_fd, SOMAXCONN) < 0) { 
    perror("listen"); 
    return 1; 
}

// Create epoll instance and return its fd
int ep = epoll_create1(EPOLL_CLOEXEC);
if (ep < 0) { 
    perror("epoll_create1"); 
    return 1; 
}

// watch socket_fd for READ events, and register it with epoll instance
struct epoll_event ev = { .events = EPOLLIN, .data.fd = socket_fd };
if (epoll_ctl(ep, EPOLL_CTL_ADD, socket_fd, &ev) < 0) { 
    perror("epoll_ctl ADD lfd"); 
    return 1; 
}
...

Next, instead of just calling accept and handling the connection
directly in the main thread, we call epoll_wait inside the loop. When
the socket is able to accept a connection, epoll_wait returns,
putting an event into the buffer we pass to it. We then iterate
through new events, checking if the associated file descriptor is
the listening socket -- in which case we accept all new connections
and add them to epoll -- or regular connection socket otherwise.

...
struct epoll_event events[128];      // event buffer
char buf[4096];                      // I/O buffer

printf("Listening on 127.0.0.1:9000...\n");

for (;;) {
    // epoll_wait is a system call. It will populate the events array and 
    // return the number of events that were triggered.
    // We also specify max events to return - calculated based on the size of the events array.
    int n = epoll_wait(ep, events, (int)(sizeof events / sizeof events[0]), -1);
    if (n < 0) {
        if (errno == EINTR) continue; // interrupted -> retry
        perror("epoll_wait");
        break;
    }
    printf("epoll_wait wake up: %d events\n", n);
    for (int i = 0; i < n; i++) {     // handle ready fds
        int fd = events[i].data.fd;   // fd associated with the event
        uint32_t e = events[i].events;// event flags - readable, writable etc

        if (fd == socket_fd) { // socket_fd is readable, that may mean new connections to accept
            for (;;) { // accept until EAGAIN or EWOULDBLOCK
                struct sockaddr_in clientaddr;
                socklen_t clientaddr_size = sizeof(clientaddr);
                // now, accept with SOCK_NONBLOCK so that the new fd is already
                // set to non-blocking, so we can avoid fcntl call
                int connection_fd = accept4(socket_fd, (struct sockaddr*)&clientaddr, &clientaddr_size, SOCK_NONBLOCK);
                if (connection_fd < 0) {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) break; // no more connections
                    else perror("accept4");
                    break;
                }
                printf("accepted connection from %s:%d\n", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port));
                // We have accepted a new connection. Now we add it to the
                // epoll instance to monitor it for events.
                struct epoll_event cev = {
                    // Edge-triggered epoll will notify only once when fd is
                    // ready, it will not keep notifying until we read all data.
                    .events = EPOLLIN | EPOLLET | EPOLLHUP | EPOLLERR,  // read, edge-triggered, hup/err
                    .data.fd = connection_fd
                };
                // Add new connection to epoll
                if (epoll_ctl(ep, EPOLL_CTL_ADD, connection_fd, &cev) < 0) {
                    perror("epoll_ctl ADD connection_fd");
                    close(connection_fd);
                }
            }

        } else { // handle existing connection
            if (e & (EPOLLHUP | EPOLLERR)) { // hangup/error
                printf("epollhup | epollerr, closing fd\n");
                // fd is removed from the epoll when the last reference
                // is closed. We never dup it so it is the only one.
                close(fd);
                continue;
            }

            if (e & EPOLLIN) { // readable
                if (handle_connection_io(fd, buf, sizeof buf) < 0) {
                    break;
                }
            }
        }
    }
}

Epoll interest can be registered as level-triggered or edge-triggered.
Level-triggered is a default option and it will keep notifying while
the interest "is fulfilled". So if I register a TCP connection
socket with read interest epoll_wait will keep waking with an
event until I read all available data from that socket.
Edge-triggered (EPOLLET option) on the other hand will notify
(at least once) only when the interest "becomes fulfilled", so in
the example above, only when new data arrives to the socket.
More details can be found on already known to you
man page.

Here I add TCP connections to epoll as edge-triggered, however, in this
case it does not really matter, since I read all available data each
time and we are not working around any constraints. I also do not care
about writes as they are done in best effort fashion.

Gotchas with writes

Write interest is slightly more complicated. If we were to use
level-triggered epoll, we would get wake up events as long as the
socket is writable, which if we do not have anything to write will
be all the time, hence the application will never "sleep". One
option here is to register write interest only when there is data
to be written, and then remove it. It is not a problem with
edge-triggered epoll, however, here we need to be mindful that we
only get notified when the socket state changes to be writable,
therefore if the socket was already writable, and we have new data
to send, we will not be notified, so either again, we re-arm the
epoll with write interest only when we have data to write, or
whenever we have new data we attempt to write to socket immediately,
and stop when the write would block.

Handling existing connections will change slightly as well, as now we
also need to handle EAGAIN and EWOULDBLOCK since connection
sockets are now non-blocking as well:

int handle_connection_io(int fd, char *buf, size_t buf_size) {
    for (;;) { // drain the connection until EAGAIN or EWOULDBLOCK
        ssize_t r = read(fd, buf, buf_size); // read some
        if (r == 0) { // peer closed
            printf("read 0 bytes, closing fd\n");
            close(fd);
            return 0;
        } else if (r < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) break; // drained
            perror("read");
            close(fd);
            return 0;
        } else { // got r bytes
            printf("read %zd bytes\n", r);
            ssize_t off = 0;
            while (off < r) { // echo back
                ssize_t w = send(fd, buf + off, (size_t)(r - off), MSG_NOSIGNAL);
                if (w < 0) {
                    // Write what we can in best effort manner. We would need
                    // EPOLLOUT and some kind of buffer per connection to do it
                    // properly in non-blocking way.
                    if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                    perror("send");
                    close(fd);
                    return -1; // signal to skip to next_event
                }
                off += w;
            }
        }
    }
    return 0;
}

The write part is a bit simplified since I do not want to store an extra
state of what has been successfully written and what was not (I could
register connection fds with write interest as well (EPOLLOUT) and get
notified when there is some progress to be made writing).

Now to compile it and run

clang main.c
./a.out

And connect from two separate terminal windows:

nc  127.0.0.1 9000

If I now start writing to the connections, I can see messages being echoed
back, and server logs show its hard work.

Listening on 127.0.0.1:9000...
epoll_wait wake up: 1 events
accepted connection from 127.0.0.1:50496
epoll_wait wake up: 1 events
read 5 bytes
epoll_wait wake up: 1 events
accepted connection from 127.0.0.1:50870
epoll_wait wake up: 1 events
read 6 bytes
epoll_wait wake up: 1 events
read 6 bytes

It can handle multiple connections, runs in a single thread, does not add
any artificial latency, and is not busy looping.

This checks the requirements. So, as all software does eventually, it is
time to rewrite it in Rust.

This full code is available on Github.

Rust

Since I am not planning to become C wizard anytime soon, to build the
foundation for something more complex and come back to the idea of
"async without async", there actually needs to be some async that I
ditch, so Rust it is.

Do not expect fireworks, though, just a bit more "flashy" echo
server...

Mio

To not libc myself into oblivion or unsafe my way to hell, I decide to
take an easier path. Path well trodden by others, the secret async
source behind Tokio - Mio.

Mio not only wraps around Epoll with a nice, easy-to-use API, but
also does so over other OSes' async APIs, making our app
cross-platform!

We can now forget about epoll's naked glory. However, the
overall approach is the same as in C:

Create a Poll instance (which on Linux uses Epoll)
Register listener and accepted connections as a Source
Wait for events, and handle IO in non-blocking way

And to be fair, there is not much more to it, since Mio is handling
all the dirty work behind the scenes.

Poll's API is quite similar to what we saw in C code, but without
making your hands dirty with direct syscall calls.
As with C code, the first thing to do is register a listener socket.
To use it with Poll, it needs to be wrapped with
mio::net::TcpListener,
which provides the aforementioned Source trait implementation
expected by the
Registry::register(...) method (Registry lives inside Poll):

...
let listener = std::net::TcpListener::bind(addr)
    .context(format!("Failed to bind to address {}", addr))?;
listener
    .set_nonblocking(true)
    .context("Failed to set listener to non-blocking mode")?;

let poll = mio::Poll::new().context("Failed to create Poll instance")?;

let mut listener = mio::net::TcpListener::from_std(listener);
let listener_token = mio::Token(0);

poll.registry()
    .register(&mut listener, listener_token, mio::Interest::READABLE)
    .context("Failed to register listener with Poll")?;
...

While in C events are associated with a specific file descriptor, to be
cross platform Mio uses
Token, which is
a wrapper around usize and allows us to map the event back to the
Source, for example a specific TCP connection, or as is the purpose
of listener_token, to TCP listener.

With Poll initialized, we can wait for events and process them:

fn wait_for_events(&mut self, duration: std::time::Duration) -> anyhow::Result<()> {
    let mut events = mio::Events::with_capacity(1024);
    self.poll
        .poll(&mut events, Some(duration))
        .context("Failed to poll events")?;

    for event in events.iter() {
        self.handle_io_for_event(event)?;
    }

    Ok(())
}

And to process them, we again come back to Tokens, as each
mio::Event
is associated with the token used when registering event::Source:

fn handle_io_for_event(&mut self, event: &mio::event::Event) -> anyhow::Result<()> {
    let token = event.token();

    if token == self.listener_token {
        if !event.is_readable() {
            tracing::warn!("Listener event is not readable");
            return Ok(());
        }

        loop {
            if self.free_tokens.is_empty() {
                tracing::warn!("Maximum connections reached, cannot accept new connection");
                break;
            }

            // Accept on mio::net::TcpListener already sets socket to
            // non-blocking mode and returns mio::net::TcpStream.
            let (mut stream, addr) = match self.listener.accept() {
                Ok((stream, addr)) => (stream, addr),
                // EAGAIN and EWOULDBLOCK are both mapped to std::io::ErrorKind::WouldBlock
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    // No more connections to accept at the moment
                    break;
                }
                Err(ref e) if is_transient_listener_error(e) => {
                    tracing::warn!(error =? e, "transient error accepting connection, retrying");
                    continue;
                }
                Err(e) => {
                    return Err(anyhow::anyhow!("Failed to accept new connection: {:?}", e));
                }
            };

            tracing::info!("Accepted new connection from {}", addr);
            let token = self
                .free_tokens
                .pop_front()
                .expect("we checked above that free_tokens is not empty");

            // On Linux Mio by default uses edge-triggered epoll
            let result = self.poll.registry().register(
                &mut stream,
                token,
                mio::Interest::READABLE.add(mio::Interest::WRITABLE),
            );
            match result {
                Ok(_) => {}
                Err(err) => {
                    self.free_tokens.push_back(token);
                    return Err(anyhow::anyhow!(
                        "Failed to register new connection with Poll: {:?}",
                        err
                    ));
                }
            }

            self.connections_by_token
                .insert(token, Connection::new(stream));
        }
    } else if let Some(connection) = self.connections_by_token.get_mut(&token) {
        match connection.process_io_event(event) {
            Ok(_) => {}
            Err(err) => {
                tracing::error!(
                    "Error processing IO event for token {:?}. Dropping connection. Error: {:?}",
                    token,
                    err
                );
                self.connections_by_token.remove(&token);
                self.free_tokens.push_back(token);
            }
        }
    } else {
        tracing::warn!("Received event for unknown token: {:?}", token);
    }

    Ok(())
}

It is analogous to our previous echo server, differentiating
between events to the listening socket and connection sockets, with
the difference that here we compare Tokens instead of file
descriptors.

Unlike in the C implementation, however, here we handle writes properly
by registering both READABLE and WRITABLE interests on the TCP
stream from an established connection. For that to work, writes are
initially appended to an in-memory buffer and then written to the
connection whenever we can make progress:

pub struct Connection {
    stream: mio::net::TcpStream,
    write_buffer: Vec<u8>,
}

impl Connection {
    fn new(stream: mio::net::TcpStream) -> Self {
        Connection {
            stream,
            write_buffer: Vec::new(),
        }
    }

    fn process_io_event(&mut self, event: &mio::event::Event) -> anyhow::Result<()> {
        let mut try_write = false;
        if event.is_readable() {
            let data = read_from_stream(&mut self.stream)
                .context("failed to read data from tcp stream")?;
            // dump to logs just for our example
            tracing::debug!(
                "Read from connection: {} bytes: {}",
                data.len(),
                String::from_utf8_lossy(&data)
            );
            // Since we want to write data back, if we read something we will
            // try to write it to connection immediately.
            self.write_buffer.extend_from_slice(&data);
            try_write = data.len() > 0;
        }
        if event.is_writable() || try_write {
            let written = self.write_pending()?;
            tracing::debug!("Wrote {} bytes to connection", written);
        }

        Ok(())
    }

    fn write_pending(&mut self) -> anyhow::Result<usize> {
        let n_written = write_to_stream(&mut self.stream, &self.write_buffer)
            .context("failed to write data to tcp stream")?;
        self.write_buffer.drain(0..n_written);
        Ok(n_written)
    }
}

When reading and writing to the connection, we need to remember to
handle WouldBlock errors as "cannot do more, wait for next epoll
event":

use std::io::{Read, Write};

pub fn read_from_stream<T: Read>(stream: &mut T) -> anyhow::Result<Vec<u8>> {
    let mut all_bytes = Vec::new();
    let mut read_buffer: [u8; 1024] = [0; 1024];

    let mut interrupted = false;

    loop {
        match stream.read(&mut read_buffer) {
            Ok(read) => {
                if read == 0 {
                    return Err(anyhow::anyhow!("Connection closed by peer"));
                }
                all_bytes.extend_from_slice(&read_buffer[..read]);
            }
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                // No more data to read at the moment
                break;
            }
            // Retry once if we hit interrupted error
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted && !interrupted => {
                interrupted = true;
                continue;
            }
            Err(e) => {
                return Err(anyhow::anyhow!("Failed to read from TCP stream: {}", e));
            }
        }
    }

    Ok(all_bytes)
}

pub fn write_to_stream<T: Write>(stream: &mut T, buffer: &[u8]) -> anyhow::Result<usize> {
    let mut interrupted = false;
    let mut written = 0;

    loop {
        if buffer.is_empty() || written >= buffer.len() {
            break;
        }

        match stream.write(&buffer[written..]) {
            Ok(n) => {
                if n == 0 {
                    return Err(anyhow::anyhow!("Connection closed by peer"));
                }
                written += n;
            }
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                // Cannot write more at the moment
                break;
            }
            // Retry once if we hit interrupted error
            Err(e) if e.kind() == std::io::ErrorKind::Interrupted && !interrupted => {
                interrupted = true;
                continue;
            }
            Err(e) => {
                return Err(anyhow::anyhow!("Failed to write to TCP stream: {}", e));
            }
        }
    }

    Ok(written)
}

That gives us all necessary ingredients to be async in non-async
world, and completes the echo server. However, if you paid attention
there is one more objective to be taken care of.

Time driven action

A lot of applications have to do more than just handling IO, and do
some work periodically, be it send some heartbeat or request some
data from another system. Let's consider those as time-based work, as
opposed to request-based work that is triggered by IO events.

It is then time to add a killer feature to the mighty echo server.
Every 5 seconds it is going to send its "status" to all connected
clients. Yes, that's it.

In async Rust one could simply do some
tokio::select
magic with a
interval timer
as one of the branches.

I still want the application to stay single threaded, so a separate
sleeping thread is also not an option. Fortunately, Epoll (or Mio) has
exactly what we need. With epoll_wait we can specify the timeout for
how long we want to wait for events, before the function will return
and presumably come back to the next iteration of the loop. Mio Poll
exposes the same functionality in poll method as well.

fn poll(&mut self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()>

I already pass timeout through wait_for_events function. And that
brings me to the last piece of this puzzle, which is the aforementioned
loop that will send the status periodically, and wait for the IO
events when idle:

pub fn run(&mut self, exit: Arc<AtomicBool>) -> anyhow::Result<()> {
    tracing::info!(
        "Echo server starting up, listening on {:?}",
        self.listener.local_addr()?
    );

    let mut status_timer = std::time::Instant::now();

    loop {
        tracing::debug!("Echo server main loop iteration starting");
        if exit.load(std::sync::atomic::Ordering::Relaxed) {
            tracing::info!("Echo server exiting as requested");
            return Ok(());
        }

        if status_timer.elapsed() >= self.status_interval {
            let num_connections = self.connections_by_token.len();
            tracing::info!(
                num_connections,
                "Broadcasting server status to all connections"
            );
            let message = format!("Server status: {} active connections\n", num_connections);
            self.broadcast_message(message.as_bytes())?;
            status_timer = std::time::Instant::now();
        }

        let time_remaining = self
            .status_interval
            .checked_sub(status_timer.elapsed())
            .unwrap_or_default();
        self.wait_for_events(time_remaining)?;
    }
}

To set things straight, broadcasting the message is non-blocking as
well. All it does is extend the write buffer for all connections
and try to progress the write as much as it can until hitting
the WouldBlock error. In case of fatal connection errors, we just
drop them:

fn broadcast_message(&mut self, message: &[u8]) -> anyhow::Result<()> {
    let mut connections_to_drop = Vec::new();

    for (token, connection) in self.connections_by_token.iter_mut() {
        connection.write_buffer.extend_from_slice(message);
        match connection.write_pending() {
            Ok(_) => {}
            Err(err) => {
                tracing::error!(
                    "Error writing broadcast message to connection: {:?}. Dropping connection. Error: {:?}",
                    token,
                    err
                );
                connections_to_drop.push(*token);
            }
        }
    }

    for token in connections_to_drop {
        self.connections_by_token.remove(&token);
        self.free_tokens.push_back(token);
    }

    Ok(())
}

Just extending the buffer would not be enough, and we would not get
any event if the connection would currently be writable, since the
state did not change.
This is the thing I mentioned here.

There is not a lot more magic than in C implementation, but for the
sake of completeness, let's run it.

You can find full source code on GitHub.

RUST_LOG=echo_server_rust=debug cargo r
2026-03-23T05:55:00.083791Z  INFO echo_server_rust::echo_server: Echo server starting up, listening on 127.0.0.1:9000
2026-03-23T05:55:00.083820Z DEBUG echo_server_rust::echo_server: Echo server main loop iteration starting

And if we connect from two terminal windows, we can see that the echo
server is cruising.

nc 127.0.0.1 9000
test
test
Server status: 1 active connections
Server status: 2 active connections

2026-03-23T05:59:00.484727Z DEBUG echo_server_rust::echo_server: Echo server main loop iteration starting
2026-03-23T05:59:00.484814Z  INFO echo_server_rust::echo_server: Broadcasting server status to all connections num_connections=2
2026-03-23T05:59:02.280579Z DEBUG echo_server_rust::echo_server: Read from connection: 5 bytes: test

2026-03-23T05:59:02.280732Z DEBUG echo_server_rust::echo_server: Wrote 5 bytes to connection

Summary - Good Bad and Ugly

This concludes the echo server, and thus the proof of concept.
As intended, I have achieved "async IO" without using async Rust.

Well, if we come back to the rant in
Asynchronous IO section, maybe it
is not so async. Perhaps this article should be named "Non blocking,
event driven IO without async", but hey, that is not very catchy.
Anyway, I am digressing...

It, however, begs the question: was it worth it?
If I were about to give you advice on writing echo servers, you are
probably better off just using async, and that is likely true
for most simple applications.

With this simple example it is hard to argue the case of async without
async, yet I am going to give it a shot.

Bad

Writing an echo server in async Rust could probably be done in 1/3 of the
lines of code. For all IO operations instead of catching would block
errors, we would simply await. Mio Poll, although still there,
would be conveniently hidden inside the belly of the Tokio runtime.

Ugly

Things would get a bit ugly if we were to go deep into some more
complex network protocol. Let's consider that we need to perform a
simple handshake. With Tokio we could simply:

send_handshake(connection).await;
let ack = wait_for_ack(connection).await;
confirm_connection(connection).await;

What happens under the hood, is that the compiler would convert
this code into a state machine, with each await marking the state
transition, where it can yield.

This convenience is gone when we drop async Rust. For it to
work with our framework, we would need to build this kind of state
machine ourselves (unless we chose to block the thread), a greatly
simplified example could be:

enum OutgoingConnectionState {
    Initiated,
    SentHandshake,
    AwaitingAck,
    Connected,
}
fn handle_io_event(...) -> ... {
...
match connection.state {
    Initiated => {
        try_send_handshake(connection)
        return SentHandshake
    },
    SentHandshake => {
        try_progress_write(connection)
        if !has_pending_write(connection) {
            return AwaitingAck
        } else {
            return SentHandshake
        }
    },
    AwaitingAck => {
        if let Some(ack) = try_read_handshake_ack(connection) {
            return Connected
        }
        return AwaitingAck
    }
}
...
}

Each "state" logic could be called multiple times in case our "hands"
are so big that they do not fit into send queues, or acks did not
arrive fully at once, therefore there can be much more logic hidden
inside function calls.

Without async, buffers and queues become your friends.

Good

Enough with the beating, though. I was supposed to argue for, not
against, my own creation!

One thing we do not see in this code are Mutexes. Since everything
is single threaded and there are no
Tasks ready to jump
around different threads with the next await, no
Send + Sync + 'static, no Pins and Arcs, we do not annoy the
borrow checker, nor ourselves. There is no async runtime, so when
things get hot, it should be easier to troubleshoot and debug.

The whole system is much more deterministic, IO can be easier to
separate from the application logic (think some kind of event loop),
and perhaps a small thing, but there is no function coloring and
async spreading over the whole codebase.

Of course some of those things could be achieved with async as well.

Conclusion

To wrap up these lengthy conclusions, I am yet to experiment with this
approach in more sophisticated systems and see whether it has some
juice in it.

How it turns out, maybe you will be able to find out in the next
~~episode~~ article. Until then I hope you enjoyed this experiment and
perhaps even learned something. If you have any questions, comments, or
suggestions, feel free to reach out. Thanks for reading!

Add context to errors

Szymon Gibała — Thu, 16 Nov 2023 07:59:39 +0000

It is Friday evening. Bob just wants to wrap up the small bug fix in the service developed by his team, be done for the week, and pack up for his weekend trip.

Bob is tired but determined. He quickly glances through the code, it is nice and readable, so with a quick search, and a few jumps to definition, he finds the suspect. He shoots a few lines of code before asking any questions, and... Done!

With good old ls Bob identifies the tests directory, and since it is a Rust project he knows it will contain integration tests. His fingers dance on the keyboard again as he quickly adds the test case, balancing on the edge of razor-sharp focus and exhaustion.

He formats the code, slams ctrl + s, opens the terminal, and types cargo test before the cursor has a chance to blink once. He can already see all the green oks, his mind can no longer stand against the gravity of the weekend. Code compiles painfully slowly, as he stretches his fingers for the last (git) push. There are no errors, not even a warning, tests start-up and...

thread 'test_whatever' panicked at Os { code: 2, kind: NotFound, message: "No such file or directory" }', tests/source_code.rs:291:14
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Bob scratches his head once, and again a bit harder... He takes a deep breath and blows off fallen hair from his keyboard while he exhales slowly, shrinking in his chair like a deflated balloon. A look at a clock reveals 8 pm. Bob stares at the code, but he does not see much, while all his brain cells converge around trying to answer a single, important question: "What f***** file?"

Convenient does not mean good

Since I started with the Rust example let me continue on this path for now, however, the problem is not language-specific.

As with everything Rust gives many options when it comes to dealing with errors. Most notably we can simply unwrap() causing the thread to panic or return from the function early when an error occurs with the ? operator (of course there are some constraints to it about which you can read in the documentation).

While unwrap()ing most often is used with caution (rightfully so), the ? operator is almost as convenient especially when paired with a dynamic error object in the form of Box<dyn std::error::Error>, and given how clean it makes the code look its use is ubiquitous.

When it comes to test code both practices are fine, at least in the majority of cases. While in the "surface code" unwraping or returning Box<dyn std::error::Error> early with ? is a good way to keep it concise, the problem begins -- as usual -- when we add more layers.

Let's consider a very simple test for a single function:

#[test]
fn test_do_something() {
    let out = do_something().unwrap();
    assert_eq!(&out, "something");
}

If our do_something function fails, we can find it out very easily, since as
part of the error we directly get the line of code that unwrapped:

---- tests::test_do_something stdout ----
thread 'tests::test_do_something' panicked at 'called `Result::unwrap()` on an `Err` value: Something went wrong', src/main.rs:15:34
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

If the test is lengthy and more complex, a good practice in my experience would be to still add some more information on what exactly failed. A month from now, you will not remember what line 344 does, you will have to navigate there, read some code, and load up your mind with long discarded context. To smooth those experiences out expect is your friend:

#[test]
fn test_do_something() {
    let out = do_something()
        .expect("failed to do something");
    assert_eq!(&out, "something");
}

---- tests::test_do_something stdout ----
thread 'tests::test_do_something' panicked at 'failed to do something: Something went wrong', src/main.rs:16:14
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Code bases, however, tend to grow and become more complicated, there are more moving pieces and interdependencies, and to fight the growing probability of errors creeping into cracks between our modules and their interactions we often reach for integration tests. Those are rarely that simple.

It is there, where we are going to have setup code, perhaps spinning up a database, creating temporary files, populating fake data to test filtering logic, or just having a lot of helper functions.

And here we come back to Bob's story.

Let's consider another example. This time we have a bunch of setup code and some functions are used in multiple places. In a real-world project this test together with all the helpers could have hundreds of lines of code:

type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

fn setup_test() -> Result<()> {
    initial_setup()?;
    populate_data()?;
    let something = do_something()?;
    Ok(())
}

fn initial_setup() -> Result<()> {
    let something = do_something()?;
    // Some initial setup code
    Ok(())
}

fn populate_data() -> Result<()> {
    let something = do_something()?;
    // Some more setup code
    Ok(())
}

fn do_something() -> Result<String> {
    Err(String::from("Something went wrong").into())
}

#[test]
fn test_do_something() {
    setup_test().unwrap();
    // Test logic...
}

---- tests::test_do_something stdout ----
thread 'tests::test_do_something' panicked at 'called `Result::unwrap()` on an `Err` value: "Something went wrong"', src/main.rs:34:22
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Now, we can only suspect that do_something failed somewhere, and it holds true only if we know that this particular error Something went wrong comes only from the do_something function. Imagine that instead, it is some standard library error such as No such file or directory, we would find ourselves in Bob's shoes scratching hair out from our scalp. Not even the RUST_BACKTRACE=full would come to our rescue.

Context for context

To make Bob's life (and likely their own) easier, people a long time ago figured out that adding context to errors at different layers makes them much easier to debug, and in general understand what happens in the application. This article is by no means any kind of tutorial and is more of a rant, but I feel obligated to provide at least a basic example of how to achieve it.

Since I do not cover the basics of Rust error handling here if you are new to Rust you can have a look at great talk by Tim McNamara that covers the error handling journey.

With Rust, the easiest way is probably to use anyhow crate that allows us to call .context(...) on any Result type where Err part implements std::error::Error trait, and turn it into anyhow::Result (or Result<T, anyhow::Error>) wrapping the error with any additional information.

Using anyhow we can transform the last example:

use anyhow::Context;

fn setup_test() -> anyhow::Result<()> {
    initial_setup()
        .context("failed to do initial setup")?;
    populate_data()
        .context("failed to populate data")?;
    let something = do_something()
        .context("failed to do something during setup")?;
    Ok(())
}

fn initial_setup() -> anyhow::Result<()> {
    let something = do_something()?;
    // Some initial setup code
    Ok(())
}

fn populate_data() -> anyhow::Result<()> {
    let something = do_something()?;
    // Some more setup code
    Ok(())
}

fn do_something() -> anyhow::Result<String> {
    Err(anyhow::Error::msg("Something went wrong"))
}

#[test]
fn test_do_something() {
    setup_test().expect("failed to setup test");
    // Test logic...
}

and have a much saner error message to work with:

---- tests::test_do_something stdout ----
thread 'tests::test_do_something' panicked at 'failed to setup test: failed to do initial setup

Caused by:
    Something went wrong', src/main.rs:39:22
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

While very convenient and straightforward, it may not be the solution for all cases, for example when we want to retain a custom error type. Sadly, for this scenario, I am yet to find a low-friction way to eat a cookie and have a cookie. Some reasonable practices I have found would be:

Introduce an error type for each module (or layer of abstraction) and extend them appropriately on different levels (leveraging the source method of the std::error::Error trait). It is not the most convenient approach and requires quite a bit of "boring code", however, it might be the most precise one. Crates like thiserror can help a lot here reducing the amount of boilerplate code.
With anyhow we can relatively easily downcast the error, so it may be a suitable solution in certain cases.
Wrapping the Result in Result, in order to have a way to represent a specific application error and all other errors that do not need special handling (here you can use anyhow again):
```
pub enum MyAppError {
    ...
}

pub fn do_something() -> anyhow::Result<std::result::Result<(), MyAppError>>
```
In this case, if we encounter some unexpected IO error we can add context to it with anyhow, but if there is an application-specific error, that we want the calling code to handle we would return Ok(Err(MyAppError {...})).

For more practical resources and deep dives rather than philosophical ranting, you can take a look at:

Jane Lusby's talk about error handling.

Luca Palmieri's talk on error handling with an example in a web application

matklad's Study of std::io::Error

It (obviously) is not just about tests

While I started my rant from the test code, I later realized that the value of adding context to errors may not be as obvious to everyone as it is to me. Then I remembered that too many times in my life I have seen this:

err := doSomething()
if err != nil {
    return err
}

and it is not a rant about Go's error handling, but cryptic errors during development or even worse production. Bear in mind that it is often other people who will troubleshoot your code and the error on line 537 of magic.go won't tell them anything. To spare those SREs a few swear words we often do not even need to add a line of code...

err := doSomething()
if err != nil {
    return fmt.Errorf("failed to do something: %w", err)
}

Say what you want about Go error handling, but its simplicity however verbose is actually great in practice. While it definitely lacks the precision of Rust type system, you just wrap the error and spare yourself hours of overthinking what is the best approach for error handling in my application as well as fighting the conveniet temptation of simply ? it away to the caller. But, that is a rant for another day...

Conclusion: helpful error messages are helpful

From my experience of troubleshooting other people's software (as well as my own ¯\_(ツ)_/¯), it is better to err on the side of more context than less. While I am not suggesting writing an essay in every error context, or even that, we need to add it in every possible place, I will take a bit redundant error such as:

failed to load config: failed to read config file: failed to parse config from JSON: invalid character 'a' looking for the beginning of object key string

over simply:

invalid character 'a' looking for the beginning of the object key string

We certainly want to strive to keep errors concise, however, from two extremes I chose the former.

Thanks for sticking with me on this rant. The whole point here is that we all could be more mindful about using things like the ? operator, if err != nil { return err }, and any other ways of just pushing errors up the call stack. Look at error messages while you writing or reviewing the code and ask yourself "will I or someone who never touched this code get at least a sense of the problem seeing this error message?".

Implementing Practical Byzantine Fault Tolerance - part 3

Szymon Gibała — Sat, 14 Oct 2023 13:18:30 +0000

Part 3: Protocol Implementation

This is the final part of my brief description of pBFT protocol implementation. The first one gave a general overview of the protocol, and the second one covered a high-level view of the project and its different parts. In this one, we are going to take a very quick tour of the normal case implementation and look more closely at view changes.

Normal Case Operations

I have covered more important parts of normal case operations in part one together with a handy diagram featured in the paper, and the code for the majority of that is fairly simple. There are still some gotchas to be careful, however.

In the previous part, I introduced PbftState where all consensus-related data lives, and PbftExecutor which implements the logic. For the normal case operations besides things like checking replica state, validation, and making sure sequence falls between watermarks, we simply add messages to the log and act when we collect 2f + 1 of them.

// Excerpt from fn process_prepare

// Since we are not guaranteed to receive pre-prepare prior to receiving first prepare messages,
// we might need to create a new entry in the log based on the prepare message.
let entry = log
    .entry(idx)
    .or_insert(RequestConsensusState::new(&message_meta));

// If we received message for a given view and sequence number with different digest,
// we reject it.
if message_meta.digest != entry.digest {
    return Err(Error::PrepareForViewAndSequenceDoesNotMatchDigest {
        view: message_meta.view,
        sequence: message_meta.sequence,
        expected: entry.digest.clone(),
        actual: message_meta.digest,
    });
}

// If we already have Prepare message from this replica, we do not add it again.
if !entry
    .prepare
    .iter()
    .any(|m| m.replica_id == prepare.replica_id)
{
    entry.prepare.push(prepare.clone());
}

// Check if replica is prepared
if entry.is_prepared(self.config.node_config.nodes.len())
    && state.message_store.has_message(message_meta.sequence)
{
    // Broadcast Commit if not done already...
}

Since different messages may arrive at different times, we need to make sure to not rely on the order. For example Prepare message from some replicas may arrive before PrePrepare from the leader, or Prepare for sequence n + 1 may arrive before the one for n.

When the replica finally collects 2f + 1 Commit messages, it is ready to apply the operation. However, this time the order matters. Since the whole protocol is about agreeing on message ordering in a distributed system, we need to make sure that it is preserved when modifying the state. So if the message n was not yet applied, we cannot apply n + 1 even if we have a sufficient number of Commits.

NOTE: In case it is not clear -- applying messages in different order on different replicas may lead to a different state. Say message n sets fluffiness: 10 and message n + 1 sets fluffiness: 0. If one replica preserves the order, but the other applies n + 1 prior to n they will no longer return the same fluffiness (which as you might imagine is a pretty serious problem).

fn apply_messages(
    &self,
    state: &mut PbftState,
) -> Result<(Vec<SignedCheckpoint>, Vec<ClientResponse>)> {
    // Start from state.last_applied
    let last_applied = &mut state.last_applied_seq;

    // Because we are using a BTreeMap, we can iterate over it in order of
    // sequence number.
    let start = state
        .consensus_log
        .iter()
        .position(|(idx, _)| idx.sequence > *last_applied);
    if start.is_none() {
        return Ok((vec![], vec![]));
    }
    ...
    for (idx, entry) in state.consensus_log.iter().skip(start.unwrap()) {
        // Entry is not committed locally, we should not apply it, and we
        // cannot proceed further.
        if !entry.is_committed_local(self.config.node_config.nodes.len()) {
            // It is possible to have entries with the same sequence in
            // different views due to the View Change protocol, this may
            // result in entry never being committed, hence we cannot simply
            // break the loop here.
            continue;
        }
        // In case we have applied a message with the same sequence, we skip
        // it. This can happen due to view change -- having a consensus 
        // entry for the same message (sequence) in different views.
        if idx.sequence == *last_applied {
            continue;
        }
        if idx.sequence > *last_applied + 1 {
            break;
        }

        // Apply messages...
    }

The other thing to be careful with is that there might be log entries with the same sequence but different a view number (if the view change occurred).

If all those things add up, however, we can at last apply the operation to the state:

...
// Take a write lock
let mut state_machine = self.state_machine.write().unwrap();
...

// Make sure that the digest matches - this should always
// be the case, hence we simply assert.
assert!(entry.digest == store_msg.digest());

let result = state_machine.apply_operation(store_msg.operation());
*last_applied += 1;
assert!(*last_applied == idx.sequence);

// Stop View Change timer if this message started it
if let Some(timer) = &state.timer {
    // The message that started the timer was applied,
    // so we can stop the timer.
    if timer.trigger_digest == store_msg.digest() {
        self.reset_timer(&mut state.timer);
    }
}

store_msg.set_opreation_result(result.clone());
...

The process of applying the operation is finished by storing the result (so that it can be returned in case of request retransmission), and sending the response to clients (KV Service nodes at /api/v1/client/response)

Checkpoints

A concept related to applying messages is checkpoints. Since they are crucial for view change protocol that we are inevitably getting closer to, let me share a sentence or two about them.

NOTE: See section "4.3 Garbage Collection" in the paper.

All replicas are going to periodically take the checkpoint of their state and share the prof with others. As the paper section title suggests this is useful for reducing the number of stored protocol messages. Additionally, they play an important role in view change protocol, as a stable boundary from which the new view can begin.

Checkpoints are taken every set amount of applied messages. For this implementation, it is going to be 10 by default. Since everything lives in memory the task is exponentially easier than in a real system as we do not need to copy any bytes on a disk (and since we also do not really worry about load, memory usage bla bla bla...).

The state machine checkpoint method is very naive. It serializes state to JSON and bluntly returns it so that we can easily produce a digest based on that.

fn checkpoint(&self, _sequence: u64) -> Result<String, Error> {
    let checkpoint =
        serde_json::to_string(&self.store).map_err(Error::CheckpointCreateError)?;
    Ok(checkpoint)
}

Using a BTreeMap to store the state instead of HashMap ensures that after serialization keys are ordered, therefore given the same state each replica is going to produce the same digest, which might have not been the case with HashMap.

NOTE: For a real system, as suggested in the paper, some copy-on-write mechanisms could be used to not duplicate the whole data on disk, which would be both space and compute-heavy.

Even then taking a checkpoint could be a quite costly operation (since we need to compute the digest of the whole state) so making them rare makes a lot of sense. However, it can result in a lot more messages being passed around during the view change so it is a balance to be found on per system basis.

Now that the checkpoint is created, the replica stores it, calculates the digest and broadcasts the Checkpoint message to other nodes:

// Part of:  fn apply_messages(...) ...
if *last_applied % self.config.checkpoint_frequency == 0 {
    let checkpoint = state_machine.checkpoint(idx.sequence)?;
    let digest = md5::compute(checkpoint.as_bytes());

    let checkpoint_digest = CheckpointDigest(digest.0);
    state.checkpoints.insert(idx.sequence, checkpoint.clone());
    state
        .checkpoint_digests
        .insert(idx.sequence, checkpoint_digest.clone());

    // Checkpoints are later broadcasted
    checkpoints.push(
        Checkpoint {
            replica_id: self.config.node_config.self_id,
            sequence: idx.sequence,
            digest: checkpoint_digest,
        }
        .sign(&self.keypair)?,
    )
}

You might have already spotted the pattern that more fun begins whenever the replica has 2f + 1 protocol messages for the same sequence, and in this case also with matching Checkpoint digest. Such a checkpoint in the eyes (or whatever else) of the replica becomes stable.

A stable checkpoint essentially means that at least f + 1 honest replicas saved the state up to that point. They therefore no longer need all the other protocol messages used to arrive at the consensus, which can now be discarded, this way freeing some space. At this point, we also shift watermarks setting the low one to a stable checkpoint sequence:

fn process_checkpoint_message(&self, state: &mut PbftState, checkpoint: SignedCheckpoint) {
...
    if entry.is_stable(self.config.node_config.nodes.len()) {
        debug!(
            seq = checkpoint.sequence,
            "checkpoint reached -- updating watermarks and discarding messages"
        );
        state.set_watermarks(checkpoint.sequence);
        self.discard_protocol_messages(state, checkpoint.sequence);
    }
}

With the state nicely checkpointed, we move to the last part of the protocol...

View Changes

Perhaps the most ~~complicated~~ fun part of the protocol is view changes. They are also quite important as they are necessary for good old fault tolerance -- if the leader goes down, someone needs to take over.

In the paper author’s words: "The view-change protocol provides liveness by allowing the system to make progress when the primary fails".
See section "4.4 View Change" for details.

The whole thing starts when the client request reaches the backup replica. Replica being a team player forwards requests to the leader first, but it does have limited patience. Regardless if the request reaches the leader or not, the backup already started the countdown.

If the particular request ends up being applied on time, the replica stops the timer and forgets the leader’s misstep.

When the time runs out, however, the backup flips to the ViewChange state, stops accepting messages other than Checkpoint, ViewChange, and NewView, and broadcasts its own ViewChange message for view v + 1.

So far so good, now let's look at the contents of the ViewChange message.

pub struct ViewChange {
    pub replica_id: NodeId,
    pub view: u64,
    // Last stable checkpoint for given replica. It is an Option in case we do
    // not have any checkpoints yet.
    pub last_stable_checkpoint: Option<ViewChangeCheckpoint>,
    // Proof for each prepared message (by sequence), containing at least 2f+1
    // Prepare messages from different replicas for a given message.
    // Each proof contains the pre-prepare message and the prepare messages by
    // public key of different replicas.
    pub prepared_proofs: HashMap<u64, PreparedProof>,
}

It quite obviously needs to contain the view number to which we propose the change, replica ID to introduce itself, and a bunch of other less straightforward stuff...

As a part of the View Change message, the replica needs to inform the leader of its current log state and provide proof that it is not fabricated. This boils down to two elements:

last_stable_checkpoint (if any) - is simply the most recent checkpoint that the replica considers stable as per rules described in the previous section. Together with it, the replica needs to deliver proof, that it actually is stable, and it does so by sending Checkpoint messages it used to conclude the stability (2f + 1 valid Checkpoint messages from different replicas):
```
pub struct ViewChangeCheckpoint {
    pub sequence: u64,
    pub digest: CheckpointDigest,
    // Map public key to signed checkpoint message by the given replica.
    pub checkpoint_proofs: HashMap<String, SignedCheckpoint>,
}
```
prepared_proofs - is a set of proofs for each sequence number that prepared after the last stable checkpoint sequence -- meaning the messages for which the replica received 2f + 1 valid Prepare messages with sequence higher than the one of last_stable_checkpoint. Such proof for every single message consists of PrePrepare and at least 2f + 1 Prepares (again, signed ones, from different replicas) -- which essentially is what made the replica conclude that the message is prepared:
```
pub struct PreparedProof {
    pub pre_prepare: SignedPrePrepare,
    pub prepares: HashMap<String, SignedPrepare>,
}
```

While theoretically only the leader needs this information, the message is broadcasted to everyone, so that during the next step, when the leader is ready to take over and sends NewView to backups, they can all verify he is legit.

Proofs and signatures

Let's take a quick detour to discuss the signatures. You might have noticed that in several places instead of ex. Checkpoint or Prepare we use SignedCheckpoint or SignedPrepare.

Those are precisely what allows replicas to prove and verify that it has legit messages from other replicas (e.g. for the ViewChange message), and not just fabricated ones, as each replica is going to sign protocol messages with its private key.

To implement that we use a SignedMessage wrapper type:

pub struct SignedMessage<T> {
    pub message: T,
    pub signature: Vec<u8>,
    pub pub_key: [u8; 32],
}

and simply sign the JSON representation of the message:

impl<T: Serialize> SignedMessage<T> {
    pub fn new(message: T, keypair: &ed25519_dalek::Keypair) -> Result<Self> {
        let serialized = serde_json::to_string(&message).map_err(
            crate::error::Error::serde_json_error("failed to serialize message"),
        )?;
        let signature = keypair.sign(serialized.as_bytes()).to_bytes().to_vec();
        Ok(Self {
            message,
            signature,
            pub_key: keypair.public.to_bytes(),
        })
    }

    pub fn verify(&self) -> Result<bool> {
        let serialized = serde_json::to_string(&self.message).map_err(
            crate::error::Error::serde_json_error("failed to serialize message"),
        )?;

        let pub_key = ed25519_dalek::PublicKey::from_bytes(&self.pub_key).map_err(
            crate::error::Error::ed25519_error("failed to parse public key from bytes"),
        )?;

        let signature = &ed25519_dalek::Signature::from_bytes(&self.signature).map_err(
            crate::error::Error::ed25519_error("failed to parse signature from bytes"),
        )?;

        Ok(pub_key.verify(serialized.as_bytes(), signature).is_ok())
    }

    pub fn pub_key_hex(&self) -> String {
        hex::encode(self.pub_key)
    }
}

Which can later be verified with pub_key. It also allows the node to make sure that the public key matches the replica ID in the message.

NOTE: While this may be a bit redundant given that replicas sign the whole request payload before sending it out, I have chosen to add additional signatures to individual protocol messages on top to make it easier to verify proofs without a need for pushing the whole request payload through different layers, storing it, decoding for verification, and sending together with ViewChange or NewView messages to other replicas.

New View

When the frustrated client broadcasted its request to all replicas, and all frustrated replicas got tired of waiting for the message to be applied and send out ViewChanges to each other, the new leader is ready to emerge.

As we established earlier (in part one), the leader is determined based on a view number, hence if we are transitioning from view v to v + 1, the appropriate replica will know it is its turn to take over. It can do so -- not surprisingly -- when it receives 2f + 1 valid ViewChange messages from different replicas and does it by broadcasting the NewView message. Beforehand, however, it needs to go through received ViewChanges and sort a few things out.

NOTE: To be clear: since the new leader needs at least 2f + 1 ViewChange messages, it is not enough if only one backup triggers the view change, so one might wonder: how is it triggered on all of them (or at least 2f + 1)?
This comes from the client behavior (see Client Request section in part one), which if tired of waiting for the responses is going to broadcast the request to all replicas participating in the consensus. This in turn will trigger a view change timer on all of them, and if the leader is down or disconnected should result in a fairly smooth transition to the new view.

Transitioning to the next view can be tricky as there might be a bunch of messages floating around in different stages of execution. For some, the PrePrepare could have been created, but only some replicas received it before triggering ViewChange, others might already be prepared but not yet committed, while a few can already be applied to the state but are not yet part of the checkpoint.

In order to keep the execution coherent there are a few things that a new leader needs to sort out before handling new messages. It starts out by determining min-s and max-s:

min-s - is the sequence number of the latest checkpoint, from all the latest checkpoints received in different ViewChanges. This number serves as a base on which all (or more precisely honest majority) agree. That is because those were stable which guarantees that the 2f + 1 of replicas agreed on it.
max-s - is the sequence number of highest prepared proofed and sent as part of ViewChanges. This one determines the last message that replicas agreed on before transitioning into the view change and that anything after that point can (and will) be forgotten, as consensus was not yet achieved.

Having those two numbers new leader can create the NewView message consisting of:

New view number.
Set of 2f + 1 valid (and signed) ViewChange messages from different replicas - as proof to others that it is legit.
Set of PrePrepare messages for all messages between min-s and max-s (the sequence numbers for those would not change, just a view they are in), or if such set would be empty (essentially min-s == max-s) the leader creates a PrePrepare message for a special NULL request, which is a no-op (does not change state in any way).

pub struct NewView {
    pub view: u64,
    // Proof of View Change messages received from different replicas
    pub view_change_messages: HashMap<String, SignedViewChange>,
    // Pre-prepare messages for those that were prepared in previous view, but
    // were not included in the last stable checkpoint
    // (or a single PrePrepare for NULL message).
    pub pre_prepares: Vec<SignedPrePrepare>,
}

Afterward, the NewView is ready to be broadcast, and the new leader officially transitions to its role. Before things get running smoothly however, replicas need to receive the NewView, ensure it is correct (verify those sweet proofs...), store new PrePrepare messages, and essentially push them through the whole protocol (by creating Prepares and so on).

The last_applied property, together with the fact that the sequence number for those requests did not change guarantees that they are not executed (or applied to the state) for a second time in case they were already committed before the view change.

Summary

This concludes my little write-up about implementing the Practical Byzantine Fault Tolerance Algorithm (finally :sweating:). If you got that far thank you for reading, and I hope you found something useful here or learned a thing or two - I certainly did.

If you spot any errors, have some suggestions, or just something you want to share, feel free to reach out, comment, or whatever, I appreciate it. And for now, I am happy to close this chapter in my side projects book and move on to the next one...

Implementing Practical Byzantine Fault Tolerance - part 2

Szymon Gibała — Thu, 12 Oct 2023 06:38:29 +0000

Part 2: Implementation Overview

This is the second part of the article describing a learning implementation of Practical Byzantine Fault Tolerance Algorithm.

In the first part I gave a general overview of the protocol, and how it unfolds for the base case. As it turns out, implementing pBFT even for non-production scenarios is not a one-evening project. This post outlines the assumptions, and decisions I have made for the implementation, together with the general structure of the project and some more interesting parts.

The full code can be found on GitHub.

NOTE: Since the whole code base became a bit too big, after several attempts I decided to make this article more of an overview than a very detailed write-up. I hope it gives enough of a background to allow anyone interested to find their way through the codebase.

Implementation Assumptions

To have something to base the implementation on I had to make some decisions on what to build and how, that are not strictly related to the protocol. The aim was to keep things "fairly" complete but with a bias towards simplicity. With this in mind, I settled on the following:

The application itself is a simple Key-Value store.
All communication is done through HTTP with messages in JSON format.
There is only one binary exposing both Key Value service as well as internal pBFT endpoints -- in a real-world project it would be good to separate them so that the consensus layer could handle different end-user applications.
Core pBFT logic, however, is separated from pbft-core library.
Message authentication is done with Ed25519 signatures
Other replicas, their addresses, and public keys are distributed as a configuration file (for development a built-in config can be used to quickly spin up nodes locally).
Key-Value store and pBFT clients are not authenticated in any way.

Who is the client?

To avoid confusion early on, let's set things straight. There will be a notion of two different clients in our system:

The client from pBFT perspective (what paper refers to as "client") -- In this section I will refer to it as pbft-client, but later anytime I mention "client", think pbft-client.
The client who uses the Key-Value Store -- I will refer to it as kv-client.

What is all that about?

In our case, the client of the consensus layer (pbft-client) will actually be the Key-Value (KV) Service itself, as it is going to translate kv-client requests into pBFT operation and send them to pBFT leader. Because of that the KV Service will also expose an endpoint to which pBFT replicas are going to send their responses. When it gathers enough confirmations that the request was executed, it will respond to the kv-client with the result.

The main reasons for such an implementation are:

Shift complexity to the backend (KV Service).
kv-client can be oblivious to the consensus mechanism that is going on behind the scenes.
pbft-client needs to be reachable by all pBFT nodes, so it is easier if it is a server listening for connections on some port.

With this (hopefully) clear, we can do a quick tour through the implementation.

Project Overview

PBFT State

The hearth of our pBFT node is the PbftState. It is going to store anything from the current replica state, and the sequence number of the last applied message, to all ClientRequest and ProtocolMessages. For our implementation all of it will live in memory, however, in the real-world scenario this would need to be backed by persistent storage.

pub enum ReplicaState {
    Replica,
    Leader { sequence: u64 },
    ViewChange,
}

pub struct PbftState {
    pub(crate) replica_state: ReplicaState,
    pub(crate) view: u64,

    pub(crate) high_watermark: u64,
    pub(crate) low_watermark: u64,
    /// watermark_k defines the range between low and high watermarks
    pub(crate) watermark_k: u64,

    pub(crate) last_applied_seq: u64,

    pub(crate) message_store: MessageStore,
    pub(crate) consensus_log: ConsensusLog,
    pub(crate) checkpoint_log: CheckpointLog,

    pub(crate) view_change_log: ViewChangeLog,

    pub(crate) timer: Option<ViewChangeTimer>,

    // We are going to store checkpoints in JSON format so that we can easily
    // take digest of them. In a real system, they would also not live in memory
    // but rather be stored on disk. Also for that reason we separate
    // checkpoints and their digests.
    pub(crate) checkpoints: BTreeMap<u64, String>,
    pub(crate) checkpoint_digests: BTreeMap<u64, CheckpointDigest>,
}

Logs where we store different ProtocolMessages are BTreeMaps so that we can easily iterate over them in order determined by either a sequence number, view number, or both combined into ConsensusLogIdx.

/// ConsensusLog is a map of a combination of view and sequence numbers. It
/// determines a state of a consensus for a specific request.
type ConsensusLog = BTreeMap<ConsensusLogIdx, RequestConsensusState>;
/// ViewChangeLog is a map of view number to a list of view change messages
type ViewChangeLog = BTreeMap<u64, Vec<SignedViewChange>>;
/// ConsensusLog is a map of sequence number to a checkpoint consensus state.
/// It keeps track of Checkpoint messages for specific sequence numbers.
type CheckpointLog = BTreeMap<u64, CheckpointConsensusState>;
...
pub struct ConsensusLogIdx {
    pub view: u64,
    pub sequence: u64,
}
...

The last crucial part is RequestConsensusState which is going to store all things necessary for the node to determine where the particular message is in terms of its journey through the protocol. This means PrePrepare, all Prepare, and Commit messages:

pub struct RequestConsensusState {
    pub digest: MessageDigest,
    pub pre_prepare: Option<SignedPrePrepare>,
    pub prepare: Vec<SignedPrepare>,
    pub commit: Vec<SignedCommit>,

    // Those will be flipped to true when the state is reached initially, so
    // that the replica does not broadcast the same message multiple times.
    pub reported_prepared: bool,
    pub reported_committed_local: bool,
}

For all details check out pbft_state.rs in the source code.

PBFT Executor

While PbftState is going to store all the things, PbftExecutor is going to do all the things (or at least a lot of them).

After a few different approaches I settled on the one where the Executor runs an event loop that processes incoming broadcasts, and internal events (e.g. when the view change timer expires), but still handles requests coming from the client directly instead of putting them into the queue. This choice requires us to use Mutex for the PbftState (which we could avoid if running a single event loop) but simplifies the API by not making it asynchronous.

This gives us the following structs:

pub struct EventOccurance {
    pub event: Event,
    pub attempt: u32,
}

pub type EventQueue = tokio::sync::mpsc::Sender<EventOccurance>;

pub struct PbftExecutor {
    event_tx: EventQueue,
    // HACK: We wrap event_rx in Mutex<Option<...>> so that we we can swap it
    // and take the full ownership of the receiver without mutable referencing
    // the Executor itself.
    event_rx: Arc<Mutex<Option<tokio::sync::mpsc::Receiver<EventOccurance>>>>,

    // backup_queue is used if the main queue is full.
    // For real system, this could be backed by some persistant storage to ofload
    // the memory. Here we are going to simply use unbouded channel.
    backup_queue: tokio::sync::mpsc::UnboundedSender<EventOccurance>,
    backup_rx: Arc<Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<EventOccurance>>>>,

    node_id: NodeId,
    config: Config,
    pbft_state: Arc<Mutex<PbftState>>,
    state_machine: Arc<RwLock<dyn StateMachie>>,

    keypair: Arc<ed25519_dalek::Keypair>,

    broadcaster: Arc<dyn PbftBroadcaster>,
}

The Executor's interface boils down to a few methods:

impl PbftExecutor {

    pub fn handle_client_request(&self, request: ClientRequest) -> Result<ClientRequestResult> {
        ...
    }

    pub fn queue_protocol_message(&self, sender_id: u64, msg: ProtocolMessage) {
        self.queue_event(Event::ProtocolMessage(NodeId(sender_id), msg).into())
    }

    pub fn queue_request_broadcast(&self, sender_id: u64, msg: ClientRequestBroadcast) {
        self.queue_event(Event::RequestBroadcast(NodeId(sender_id), msg).into())
    }

    pub async fn run(&self, mut rx_cancel: tokio::sync::broadcast::Receiver<()>) -> () {
        // HACK: hack to take full ownership of the receiver without needing a
        // mutable reference to the executor itself.
        let event_rx = self.event_rx.lock().unwrap().take();
        if event_rx.is_none() {
            error!("event channel is not present, cannot start executor event loop");
            return ();
        }
        let mut event_rx = event_rx.unwrap();

        loop {
                tokio::select! {
                    _ = rx_cancel.recv() => {
                        info!("received cancel signal");
                        return ();
                    }
                    event = event_rx.recv() => {
                        // Handle event...
                    }
                }
        }
    }
}

and Event is defined as an enum of three variants:

pub enum Event {
    RequestBroadcast(NodeId, ClientRequestBroadcast),
    ProtocolMessage(NodeId, ProtocolMessage),
    ViewChangeTimerExpired(MessageDigest),
}

The Executor is doing all the dirty work, from handling client requests, processing protocol messages, applying operations to the state, or watching for view changes.

In part one I describe some steps involved in doing those, if you are interested in the implementation, I hope that the links above give a good enough starting point to not get lost. After all, it boils down to doing some validation, adding entries to maps, and checking some conditions.

Configuration

Before we go any further, it is worth looking at the node configuration. Besides some obvious things one may want to tweak without touching the source code like checkpoints frequency or view change timeout, the most important part for the consensus is information about other nodes.

As I briefly mentioned in the assumptions, nodes know about each other from the config, which contains their IDs, addresses, and public keys. self_id determines the ID of this particular node, and private_key_path points to the file containing the private key used for signing messages, which is loaded at the startup.

pub struct Config {
    pub node_config: PbftNodeConfig,
    pub checkpoint_frequency: u64,
    pub view_change_timeout: Duration,
    pub response_urls: Vec<String>,
    pub executor_config: ExecutorConfig,
}

pub struct PbftNodeConfig {
    pub self_id: NodeId,
    pub private_key_path: PathBuf,
    pub nodes: Vec<NodeConfig>,
}

pub struct NodeConfig {
    pub id: NodeId,
    pub addr: String,
    pub public_key: String,
}

There is also a response_urls field, which informs our nodes where to send the client response. Since the pbft-client is going to be only the KV Service, I decided to send responses to all KV Nodes every time, as opposed to passing response_url in ClientRequest on a per-request basis. The reasoning behind it is that kv-client may connect to different KV Nodes (as there are no sessions etc.), and hence send the request to one node, but retry it to the other.

Broadcaster

Knowing where the addresses of fellow replicas come from we can look into Broadcaster component. Invoked by the PbftExecutor it handles: broadcasting client requests between the nodes, protocol messages, as well as delivering responses to clients.

The implementation is simple. We create a Future for each request to be sent, put it into FuturesUnordered, and spawn tokio::task to await their execution:

fn send_client_responses(&self, responses: Vec<ClientResponse>) {
    let mut futs = FuturesUnordered::new();
    for resp in responses {
        for url in &self.response_urls {
            let client = self.client.clone();
            let self_id = self.node_self_id;
            let keypair = self.keypair.clone();
            let resp = resp.clone();
            let url = url.clone();

            futs.push(Box::pin(async move {
                Broadcaster::send_with_retires(client, self_id, &keypair, &resp, url.as_str())
                    .await
            }));
        }
    }

    tokio::spawn(async move {
        while let Some(out) = futs.next().await {
            match out {
                Ok(_) => {}
                Err(err) => {
                    error!(error = ?err, "failed to send response to client in all attempts");
                }
            }
        }
    });
}

It is not ideal in way that our pBFT code does not know if the request was received successfully, but allows us to keep the method non-async, and therefore not spread it throughout the whole codebase.

NOTE: A better implementation here could be to create a request queue, and make broadcaster distribute requests over a pool of workers. This would allow us to better keep track of what is going on, have sane limits, and retry policy... but I did not implement it.

There are a few more things going on in the Broadcaster, but probably the most important one is signing messages.

When sending the message, the Broadcaster serializes the payload to JSON, and signs it with the replica private key. Then it adds two headers to the request, one containing Replica ID, and the other hex-encoded signature. This way other replicas can later easily find the public key to verify the signature.

pub async fn send_msg<T: Serialize>(
    client: &reqwest::Client,
    self_id: NodeId,
    keypair: &ed25519_dalek::Keypair,
    msg: &T,
    url: &str,
) -> Result<()> {
    let body = serde_json::to_vec(msg).map_err(BroadcastError::serde_error(
        "failed to serialize request body",
    ))?;

    let signature = keypair.sign(&body).to_bytes().to_vec();
    let signature_hex = hex::encode(signature);

    let res = client
        .post(url)
        .header(REPLICA_ID_HEADER, self_id.0)
        .header(REPLICA_SIGNATURE_HEADER, signature_hex.to_string())
        .header(reqwest::header::CONTENT_TYPE, "application/json")
        .body(body)
        .send()
        .await
        .map_err(BroadcastError::send_error(url.to_string()))?;

    if res.status().is_success() {
        Ok(())
    } else {
        Err(BroadcastError::UnexpectedStatusError {
            url: url.to_string(),
            status_code: res.status(),
        })
    }
}

State Machine

Last but not least we have the StateMachine, which is a trait exposing methods to apply operations and create a checkpoint:

pub trait StateMachie: Send + Sync {
    fn apply_operation(&mut self, operation: &Operation) -> OperationResult;
    fn checkpoint(&self, sequence: u64) -> Result<String, Error>;
}

And simple Key-Value store that implements it:

pub struct InMemoryKVStore {
    store: BTreeMap<String, String>,
}

It is just a simple map, so there is nothing really to talk about. In case you want to check it out the source code is very simple.

API

Now that we have covered the core components implementing the pBFT let's take a peek at an API. Since it is an entrypoint to the application it has to wrap nicely around the protocol to make it usable.

As I have mentioned in the first part, for simplicity both the application and consensus layer will be a part of a single application, and therefore API as well. Paths for different responsibilities are going to be grouped, but it all will run as a single server. We can quickly glimpse at routes configuration, to get a better idea (for details see api.rs in kv-node).

let kv_router = Router::new()
    .route("/", post(handle_kv_set))
    .route("/", get(handle_kv_get))
    .route("/local", get(handle_kv_get_local));

// KV Nodes receives responses from pBFT replicas here
let consensus_client_router =
    Router::new().route("/response", post(handle_client_consensus_response));

// KV Node translates the request to pBFT operation and sends it here
let consensus_ext_router = Router::new()
    // Request operation to execute
    .route("/operation", post(handle_consensus_operation_execute));

// pBFT nodes talking to each other
let consensus_int_router = Router::new()
    // Client Request + PrePrepare broadcasted by the leader
    .route("/execute", post(handle_consensus_pre_prepare))
    // Any other consensus message -- Prepare, Commit, ViewChange, NewView
    .route("/message", post(handle_consensus_message))
    // Debuging endpoint to dump current pBFT state
    .route("/state", get(handle_state_dump));

// Combine routers
let app = Router::new()
    .route("/api/v1/health", get(health_handler))
    .nest("/api/v1/kv", kv_router)
    .nest("/api/v1/client", consensus_client_router)
    .nest("/api/v1/consensus", consensus_ext_router)
    .nest("/api/v1/pbft", consensus_int_router)
    .with_state(self.ctx.clone());

The other important part of the API is verifying that the request comes from the replica it claims to. We have looked at how the Broadcaster signs the request body and adds two headers to the request. Now it is the time to do the reverse. With custom axum extractor the node verifies the signature prior to deserializing JSON body to whatever type the handler expects:

pub struct JsonAuthenticated<T: DeserializeOwned> {
    pub sender_id: u64,
    pub data: T,
}

pub struct JsonAuthenticatedExt<T: DeserializeOwned>(pub JsonAuthenticated<T>);

#[async_trait]
impl<S, T: DeserializeOwned> FromRequest<S, Body> for JsonAuthenticatedExt<T>
where
    S: VerifySignrature + Send + Sync,
{
    type Rejection = axum::response::Response;

    async fn from_request(
        req: axum::http::Request<Body>,
        state: &S,
    ) -> Result<Self, Self::Rejection> {
        let (parts, body) = req.into_parts();

        let signature = get_replica_signature(&parts.headers)?;
        let peer_id = get_sender_replica_id(&parts.headers)?;

        // this wont work if the body is an long running stream -- it is fine
        let bytes = hyper::body::to_bytes(body)
            .await
            .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())?;

        state.verify_signature(peer_id, &signature, &bytes)?;

        let data = serde_json::from_slice(&bytes)
            .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()).into_response())?;

        Ok(JsonAuthenticatedExt(JsonAuthenticated {
            sender_id: peer_id,
            data: data,
        }))
    }
}

Verifying the signature boils down to hex-decoding public key and the signature, and doing the verification:

pub fn verify_request_signature(
    &self,
    replica_id: u64,
    signature: &str,
    msg: &[u8],
) -> Result<(), crate::error::Error> {
    if replica_id > self.nodes_config.nodes.len() as u64 {
        return Err(crate::error::Error::InvalidReplicaID {
            replica_id: NodeId(replica_id),
        });
    }
    let peer = &self.nodes_config.nodes[replica_id as usize];

    let pub_key_raw = hex::decode(peer.public_key.as_bytes()).map_err(
        crate::error::Error::hex_error("failed to decode public key from hex"),
    )?;

    let public_key = PublicKey::from_bytes(&pub_key_raw).map_err(
        crate::error::Error::ed25519_error("failed to parse public key from bytes"),
    )?;

    let signature_raw = hex::decode(signature.as_bytes()).map_err(
        crate::error::Error::hex_error("failed to decode signature from hex"),
    )?;

    let signature = Signature::from_bytes(&signature_raw).map_err(
        crate::error::Error::ed25519_error("failed to parse signature from bytes"),
    )?;

    let is_ok = public_key.verify(msg, &signature).is_ok();
    if !is_ok {
        return Err(crate::error::Error::InvalidSignature);
    }
    Ok(())
}

Now we simply use the extractor for the handlers where we want to verify the signature. If the message is not signed, or the signature is not right, the request is rejected before we reach the handler:

async fn handle_consensus_message(
    ctx: axum::extract::State<HandlerContext>,
    JsonAuthenticatedExt(protocol_msg): JsonAuthenticatedExt<ProtocolMessageBroadcast>,
) -> impl axum::response::IntoResponse

Summary

This covers the main components of the application, the rest "is just" protocol implementation... The last part is going to look into the remaining parts of the protocol that I did not touch on yet such as checkpoints and the view change.

Implementing Practical Byzantine Fault Tolerance

Szymon Gibała — Thu, 05 Oct 2023 05:58:39 +0000

Part 1: The Foundation

Practical Byzantine Fault Tolerance is one of the seminal papers in the world
of Distributed Systems. It is one of the first practical solutions proposed to the Byzantine Fault Problem (also known as the Byzantine Generals problem).

NOTE: To keep this post focused instead of describing everything from the ground up I will assume readers' familiarity with some of those base concepts, or being comfortable to move on despite not knowing them.
If you are not familiar with some or all, do not worry, everyone starts somewhere! If you like reading papers I can send you straight to the source. If you prefer more casual articles there is plenty out there too with some good videos on top. Or... You can just ask Chat GPT to explain it.

Since this paper is very fundamental to the world of Distributed Systems, especially when we venture beyond the walls of trusted systems running in our data center to things like blockchain where nodes are operated by different parties, and so many later algorithms are built on top of it, I reckoned that understanding it well will help me a lot with understanding others. And what is the better way to understand something than to implement it?

In this post, I will lay some groundwork for understanding the protocol, make some high-level design choices to make implementation fairly simple, and try to model basic data structures.

Since this is a learning project, do not expect optimal design, handling all the edge cases, or production-grade code. The goal is to implement a complete protocol for a more or less happy path.

If you are already familiar with pBFT you can probably quickly skim this part. Otherwise, I strongly encourage you to look at the paper, follow along, and read up on any details that I omit.

Base Protocol

A good start to understanding pBFT is the base case -- achieving consensus on a single value. Let's ignore unresponsive leaders, view changes just for now, and try to understand the core.

NOTE: Following the paper we will be mostly looking at sections "4.1 The Client' and "4.2 Normal-Case Operation". We will also peak at the basics as they unfold.

The paper features a simple diagram, that pretty much illustrates the whole protocol for a single message (see "Figure 1: Normal Case Operation"), here's my recreation of it with a bit more details:

To understand it, however, we need some background knowledge about how pBFT works, what are the constraints, etc:

pBFT assumes that at a given time (or in each view) there is exactly one replica that acts as a leader.
Since the whole purpose of pBFT is to achieve consensus in distributed system while tolerating the Byzantine Fault it can do so as long as at most ((n-1) / 3) (less than one third) of the replicas are misbehaving. In other words, if we want to tolerate f faulty replicas, we need at least 2f + 1 honest ones.

NOTE: From Section 3 of the paper "Service Properties":
"The algorithm provides both safety and liveness assuming no more than ((n-1) / 3) replicas are faulty. Safety means that the replicated service satisfies linearizability [...]: it behaves like a centralized implementation that executes operations atomically one at a time. Safety requires the bound on the number of faulty replicas because a faulty replica can behave arbitrarily, e.g., it can destroy its state."
From the above formula we can see that we need at least 4 replicas in the pBFT system to tolerate 1 dishonest one. For non-Byzantite Fault-tolerant consensus protocols such as Raft or Paxos the minimum number of replicas is usually 3, and the number needs to be odd so that they cannot be split into two even groups. Since in basic pBFT, a leader is predetermined for each view, there is no need for an election, hence we do not need a majority for it.

We dipped our toes in the theory enough to establish some ground rules, now let's come back to our diagram, look at each "Phase" in a bit more detail, and start writing some code.

NOTE: This and following post will include some snippets of code but it is far from the full implementation, so trying to copy-paste from it is rather pointless. For the full code base refer to the repository.

As many great adventures through layers of code begin with someone wanting to do something let's start this one here as well, with the client request.

Client Request

As we see on the diagram, the client sends the request to a single replica. That replica ideally is a leader, but if that is not the case, the request should be forwarded to the leader by any backup node.

Leader Election

In pBFT there is no leader election such as in non-BFT protocols like RAFT or Paxos. Instead, replicas become leader in a predetermined order based on their ID and current view. The leader remains in charge for the whole duration of the view, in other words: as long as protocol progresses (requested operations are being executed).
The simplest algorithm for selecting the leader would be: leader_id = view_number % number_of_replicas.

After the client request successfully reaches the node, the client's watch begins. For it to consider the request executed, it expects to receive matching responses from at least 2f + 1 different replicas directly (the last section on the diagram: reply).

If the client gets tired of waiting and smells that something might be fishy, it will send the request again, this time to all of the replicas.

NOTE: This part is later important for triggering the view changes...

You might have already spotted a common problem that may occur here: multiple executions of the same request. Fear not, however, as the solution for that is rather simple. The client will simply include a sequence number (or timestamp) with each request, which together with some client ID, will uniquely identify each request, hence will prevent multiple executions.

To keep things more straightforward, my implementation is using request UUID as a measure of identifying requests.

With all that I can propose an initial representation of the ClientRequest. The service behind pBFT will be a simple key-value store, but I will at least try to separate the consensus layer from the application just a bit.

NOTE: To spare screen real estate code in this article will be stripped to the minimum required to get the main points across. I omit things like attributes, helper methods, boilerplate, and other details. Again, the complete code can be found in the GitHub repository.

pub struct ClientRequestId(pub uuid::Uuid);

pub struct ClientRequest {
    pub request_id: ClientRequestId,
    pub operation: Operation,
}

For the key-value store operations can be defined as a simple enum:

pub enum Operation {
    Noop, // Used by the protocol for NULL messages
    Set { key: String, value: String },
    Get { key: String },
}

The last thing we need here is a response that the client receives from honest replicas that processed the request:

pub struct ClientResponse {
    pub request: ClientRequest,
    pub result: OperationResultSequenced,
}

pub struct OperationResultSequenced {
    pub result: OperationResult,
    /// Sequence number of the operation assigned by the pbft leader when
    /// initially accepting the operation
    pub sequence_number: u64,
}

pub enum OperationResult {
    Noop,
    Set { key: String, value: String },
    Get { key: String, value: Option<String> },
}

The response consist of:

Operation result.
Sequence number (or simply sequence) - assigned as part of processing request through pBFT, determines the exact order of operation execution.
Client request - it could include just the request ID, but for this implementation, it might as well be included as a whole.

The client also has to identify which replica sends the response and verify its authenticity, which I am going to cover later (in part 2) with message signatures and API implementation.

Let's wrap up this part by modeling protocol messages.

pBFT consensus

Coming back to the diagram, we can see that the core of the protocol is composed of three phases:

Pre-Prepare
Prepare
Commit

Let's zoom in on those one by one.

Pre-Prepare

When the leader receives a client request it initiates the consensus for the operation. This process consists of:

Computing the digest of the client request.
Assigning a sequence number to it.
Storing the request, digest, and newly composed PrePrepare message.
Broadcasting PrePrepare and the client request to all replicas.

ProtocolMessage can be represented as an enum with all its possible variants:

pub enum ProtocolMessage {
    PrePrepare(PrePrepare),
    Prepare(Prepare),
    Commit(Commit),
    Checkpoint(Checkpoint),
    ViewChange(ViewChange),
    NewView(NewView),
}

The PrePrepare includes metadata about the associated request:

// Simple md5 digest
pub struct MessageDigest(pub [u8; 16]);

pub struct MessageMeta {
    pub view: u64,
    pub sequence: u64,
    pub digest: MessageDigest,
}

pub struct PrePrepare {
    pub metadata: MessageMeta,
}

As mentioned in the paper, PrePrepare does not include the client request itself to keep it smaller. This is important later on during view changes when a new leader might need to broadcast a whole bunch of PrePrepares in a new view for previously sent requests.

The authors also mention the possibility of using separate protocols for smaller pBFT messages and bigger client requests indicating that those are sent completely separately - we are not going that far.

Since initially both PrePrepare and the client request are easier to process together, they are going to be broadcasted together as well:

pub struct ClientRequestBroadcast {
    pub request: ClientRequest,
    pub sequence_number: u64,
    pub pre_prepare: PrePrepare,
}

With this done, prepare yourself for the next phase...

Prepare

Upon receiving the PrePrepare backup performs some validation such as:

Verify signature and message digest.
Check that the message is in the current view.
Ensure it did not previously accept a different PrePrepare for the same sequence and view.
Check that the sequence number is between high and low watermarks.

If it is all good replica decides to accept the message and enters the prepare phase by:

Storing received PrePrepare and the client request.
Creating and storing its own Prepare message for the same request.
Broadcasting the Prepare to all replicas.

The Prepare message will include the same metadata as PrePrepare with a single addition of replica_id. This is because we only expect one PrePrepare for the consensus round coming from the leader, but each replica is going to create, and send its own Prepare:

pub struct Prepare {
    pub replica_id: u64,
    pub metadata: MessageMeta,
}

All (or at least most) replicas will be doing it at more or less the same time -- bombarding each other with Prepare messages. The replica is considered prepared for a given message when its log contains:

Client request.
PrePrepare message for a given request.
2f + 1 Prepare messages from different replicas (including itself) matching the PrePrepare.

NOTE: In the paper we see it state more precisely as: prepared(m, v, n, i), where m is a message, v is a view, n is a sequence number, and i is replica id.

Whether different protocol messages are "matching" is determined by the metadata - view, sequence number, and digest - in other words, if they are associated with the same client request.

This process ensures that the majority of honest replicas will agree on the same message order (determined by sequence number) in a given view. Or to put it simply: agree on the same message for a given combination of view and sequence number.

NOTE: sequence numbers increase independently of the view, meaning that if we achieved consensus for sequences 1-10 in view 1 and transition to view 2, the sequence numbers will not reset but will continue to grow, e.g. 11, 12, 13, etc.

When the replica is prepared it is time to move to the commit phase.

Commit

When the replica reaches the prepared state for a given message, it, again, broadcasts a message to all other nodes, this time it is Commit.

The process here is very similar to the prepare phase in that, replicas will validate received messages, and insert them into the log.

Similarly to the prepared condition of the previous phase, in this one, we define a committed-local property, which is true for the replica if it got 2f + 1 Commit messages from different peers.

With all the similarities the Commit message consists of the exact same properties as Prepare:

pub struct Commit {
    pub replica_id: u64,
    pub metadata: MessageMeta,
}

There is another cluster-wide property described by the paper - committed(m, v, n) - which is true when at least f + 1 non-faulty replicas reached prepared state for a given message (this essentially amounts to 2f + 1 all replicas, given the assumption that at most f of them are faulty). The commit phase ensures that if at least one non-faulty replica is committed-local(m, v, n, i) then committed(m, v, n) is true. In other words, if one honest replica reaches committed-local property for the message, the message can be considered set in stone.

As a result, as soon as committed-local is true for a given message, the replica is ready to execute it (apply the operation to its state). After successful execution, the replica sends the reply to the client thereby finishing the flow represented in the diagram.

The client can be certain that the operation went through if it received 2f + 1 replies from different replicas.

There are still a few gotchas to watch out for. Since replicas can commit the requests out of order (e.g. one with sequence = 5 before the one with sequence = 3) depending on message order delivery, and therefore reach committed-local(..., n+1) before committed-local(..., n). The whole point of the protocol is to agree on ordering, hence replicas cannot execute operations out of order.

The paper assumes partial synchrony - that each message from an honest replica will get delivered eventually - so with this assumption we should not get stuck on any message indefinitely as long as 2f + 1 replicas are honest (and operating).

Summary

This wraps up a very basic overview of the protocol base case. There are more details to it but if you are interested in a really deep dive the paper is your go-to. Hopefully, this primer gives some good ground to get through it.

This, however, is not the end, as in the following parts I am going to cover the high-level overview of the implementation of both pBFT as well as Key-Value store on top of it and look into the remaining parts of the protocol: checkpoints and view changes.

Link to the code repository.

Single syscall "Hello, world" - in Rust - part 2

Szymon Gibała — Wed, 01 Mar 2023 19:39:40 +0000

... or There and Back Again

Returning to Rust

As in every hero's journey, after gaining the wisdom of the gods, the hero shall bring it back to his roots. If we squint our eyes a bit, that is (kind of) what we are going to do.

Our exploration in the previous post revealed some suspects that blow up our syscalls count in a simple "Hello, world" Rust application, those are:

Rust runtime
libc

Let's try getting rid of them.

No std

As a first step of our journey back to Rust, we need to cut some fat out of Rust runtime. We can do it by removing the standard library with no_std attribute. Our minimal "Hello, world" program will look like this:

#![no_std]
#![no_main]

use libc;

const MSG: &'static str = "Hello, world!\n";

#[no_mangle]
pub extern "C" fn main(_argc: isize, _argv: *const *const u8) -> isize {
    unsafe {
        libc::printf(MSG.as_ptr() as *const _);
    }
    0
}

#[panic_handler]
fn panic(_: &core::panic::PanicInfo) -> ! {
    loop {}
}

Since there is plenty of good resources about no_std Rust, let's just get a quick overview of what is happening here:

#[no_std] attribute tells Rust to not link the standard library.
Since it is the standard library that defines panic_handler for Rust programs, without it we need to define our own using #[panic_handler] attribute on a function with a proper signature.
On the same note we cannot use Rust's default main function as an "entry point" to our program, so we need to use #![no_main] attribute and provide our own main function. #[no_mangle] attribute tells the compiler to not change the name of our function, so it can be found (and called) by libc.
And finally, we do not have access to the println macro, so we use libc::printf instead.

For our Cargo.toml we specify libc as a dependency and tell the compiler that we want our program to abort on panic. This second piece is necessary again because we do not use std and we are building a program for a target where eh_personality (eh stands for "exception handling") is defined in the standard library. Since eh_personality is necessary for stack unwinding when a panic occurs, aborting absolves us from the need to provide it:

...
[profile.dev]
panic = "abort"

[profile.release]
panic = "abort"

[dependencies]
libc = { version = "0.2.139", default-features = false }

You can try it for yourself by removing panic = "abort" lines. Rust compiler error messages would point you in the right direction.

and voila, it works:

cargo run

Hello, world!

However... In respect of your screen real estate, I am not even going to bother pasting strace output here as it is bloated like a dead whale (ok, maybe not that bad just 35 syscalls...). This is because we still use the printf function, libc, and on top of that link it dynamically...

Statically linking libc (especially musl) to no_std program turns out to be not a trivial task, and since we need to get rid of it anyway, let's not go down this rabbit hole. Let's instead get rid of it altogether.

No libc

Okay, we can remove libc from our dependencies, remove calls to printf that depend on it, and we are good. Right?

#![no_std]
#![no_main]

const MSG: &'static str = "Hello, world!\n";

#[no_mangle]
pub extern "C" fn main(_argc: isize, _argv: *const *const u8) -> isize {
    0
}

#[panic_handler]
fn panic(_: &core::panic::PanicInfo) -> ! {
    loop {}
}

Yes... but well, no...

cargo run
...
/usr/bin/ld: /usr/lib/gcc/x86_64-pc-linux-gnu/12.2.0/../../../../lib/Scrt1.o: in function `_start':
/build/glibc/src/glibc/csu/../sysdeps/x86_64/start.S:115: undefined reference to `__libc_start_main'
collect2: error: ld returned 1 exit status
...

(cut down to only relevant parts)

We have at least two problems here. The first one is what we see on the screen -- a beautiful linker error -- and the second one is what we do not see on the screen -- our "Hello, world!" message -- because we removed the printf function call.

Since tackling the first one is a prerequisite for the second, let's start with it.

As I mentioned earlier, with our #[no_std] binary we had to provide a custom main function that is called by libc. But now that we do not have libc there is nothing to call our main function...

However, as we can see in the error message above something still refers to __libc_start_main function, which reasonably so cannot be found. We can see that error originated in the Scrt1.o file in _start function. Scrt1.o is a part of the C runtime startup code so we can reason that Rust still tries to link it to our binary.

Since this is not really the problem with our code, but more with the build (linking) process, we need to tell the compiler to not link those files, and we can do it by passing -nostartfiles flag to the linker.

RUSTFLAGS="-C link-arg=-nostartfiles" ...

In gcc docs we can read:

-nostartfiles
Do not use the standard system startup files when linking. The standard system libraries are used normally, unless -nostdlib, -nolibc, or -nodefaultlibs is used.

To not specify RUSTFLAGS every time, we can move them to .cargo/config.toml:

[target.'cfg(target_os = "linux")']
rustflags = ["-C", "link-args=-nostartfiles"]

And we are good to go!

cargo run

[1]    1434 segmentation fault (core dumped)  cargo run

I will take that as a no...

Remember the _start function, right? Well, since we no longer link startup files it is no longer here (surprise!), and apparently, it is "kind of" needed.

When OS loads the program it will look for the entry point address in the ELF file header to start the execution. However, if the entry point function was not found during the linking process, the address will be set to 0x0 (NULL), which usually is a protected memory area.

readelf -h ./target/debug/hello-world | grep Entry

  Entry point address:               0x0

The _start function is a default expected by the linker. This means we could simply rename our main function to _start... Or we can convince the linker that our function is better! We can use the same trick as before -- providing a linker configuration -- this time by passing --entry flag as a link-arg:

RUSTFLAGS="-C link-arg=--entry=main" cargo run

[1]    2022 segmentation fault (core dumped)  RUSTFLAGS="-C link-arg=--entry=main" cargo run

😑 ...

Is the entry point set?

readelf -h ./target/debug/hello-world | grep Entry

  Entry point address:               0x1020

Looks like it is, but... Remember our Assembly program? On top of write we also used the exit syscall, and without it, the program would segfault too. We might be facing a simillar issue here.

We can take a closer look at this by checking the Assembly code generated by the Rust compiler. To do that new need to expand our RUSTFLAGS, this time with --emit=asm flag:

RUSTFLAGS="-C link-arg=--entry=main --emit=asm" cargo build --release

We are building in release mode this time to cut out debug symbols noise from the Assembly file. With that, we can find a concise .s file in the target/release/deps directory, and see our main function:

    .text
    .file   "hello_world.57a4ccb5-cgu.0"
    .section    .text.main,"ax",@progbits
    .globl  main
    .p2align    4, 0x90
    .type   main,@function
main:
    retq
.Lfunc_end0:
    .size   main, .Lfunc_end0-main
...

The function contains only a single retq instruction, no syscalls, and no exit codes. To get even closer to the binary we can disassemble main directly, and confirm what we have already seen:

objdump --disassemble=main ./target/release/hello-world

./target/release/hello-world:     file format elf64-x86-64


Disassembly of section .text:

0000000000001000 <main>:
    1000:   c3                      ret

Since we are yet to research how to print to the screen without libc, let's confirm the hypothesis by doing something that we can empirically detect in our code. We can do it by calling panic!, or simply adding an endless loop:

Since that is exactly the behavior of our panic_handler the result is effectively the same.

...
#[no_mangle]
pub extern "C" fn main(_argc: isize, _argv: *const *const u8) {
    loop{}
}
...

RUSTFLAGS="-C link-arg=--entry=main --emit=asm" cargo run

And now we are stuck, which is what endless loops usually do. We can update .cargo/config.toml with our new link-arg and move on:

[target.'cfg(target_os = "linux")']
rustflags = ["-C", "link-args=-nostartfiles --entry=main"]

We know the drill now from our previous adventures, we just write(2) and exit(2), and we are done! It might be time to reach out to some old "friends"...

Assembly. Again...

Remember the "wisdom of the gods" part? Yeah, that was not (entirely) a joke.

Since we already have tremendous experience with assembly after writing our "Hello, world" program, it would be a shame not to use it again... You might be asking, "Am I cheating once more?". Maybe. Or no, because it is me who made up those rules [evil laugh or something...].

Regardless, this time we are going to use Assembly from Rust (see, it is not cheating!). Fortunately, we can do that fairly easily, all we need is the asm! macro:

use core::arch::asm;
...
#[no_mangle]
pub extern "C" fn main(_argc: isize, _argv: *const *const u8) {
    unsafe {
        // Execute write syscall
        asm!(
            "syscall",
            in("rax") 1, // write syscall number
            in("rdi") 1, // stdout file descriptor
            in("rsi") MSG.as_ptr(),
            in("rdx") MSG.len(),
        );
        // Execute exit syscall
        let exit_code = 0;
        asm!(
            "syscall",
            in("rax") 60,
            in("rdi") exit_code,
            options(noreturn)
        );
    }
}
...

And run it:

cargo run

Hello, world!

Amazing! Let's just confirm with strace, and we are done...

strace -c ./target/debug/hello-world

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           mmap
  0.00    0.000000           0         1           mprotect
  0.00    0.000000           0         1           brk
  0.00    0.000000           0         1         1 access
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         2         1 arch_prctl
  0.00    0.000000           0         1           set_tid_address
  0.00    0.000000           0         1           set_robust_list
  0.00    0.000000           0         1           rseq
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0        11         2 total

Ah, yes, of course... A quick look at the output of:

strace ./target/debug/hello-world

Can refresh some memory pages in my head...

...
access("/etc/ld.so.preload", R_OK)      = -1 ENOENT (No such file or directory)
...

ld.so.preload again, isn't it? We never actually got to build our no_std binary statically, so it is still dynamically linked.

file ./target/debug/hello-world

./target/debug/hello-world: ELF 64-bit LSB pie executable, x86-64, version 1 (SYSV), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=ca87b904fa8cd9cb232c819143edd5abb16cdaa7, with debug_info, not stripped

Been there done that, we know what to do, and after quick facepalm, we can again set target-feature=+crt-static, this time in .cargo/config:

[target.'cfg(target_os = "linux")']
rustflags = ["-C", "link-args=-nostartfiles --entry=main", "-C", "target-feature=+crt-static"]

And run it again:

cargo build && strace -c ./target/debug/hello-world

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           execve
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0         2           total

There we have it, mission accomplished! "Hello, world!" program with just a single syscall, written in Rust (kind of).

It still hurts to look at tho. Perhaps we could use some library to have nice and smooth Rust functions and let someone more fluent speak Assembly for us... You know, sweep it under the rug type of thing...

Rust is not JavaScript, but there actually is a crate for that. Let's add it to our Cargo.toml:

[dependencies]
sc = "0.2.7"

Our main function parameters are useless anyway, and we no longer need pub extern "C", so we clean it up as well with our last refactor. The final program looks much nicer:

#![no_std]
#![no_main]

#[macro_use]
extern crate sc;

const MSG: &'static str = "Hello, world!\n";

#[no_mangle]
fn main() {
    unsafe {
        syscall!(WRITE, 1, MSG.as_ptr(), MSG.len());
        syscall!(EXIT, 0);
    }
}

#[panic_handler]
fn panic(_: &core::panic::PanicInfo) -> ! {
    loop {}
}

Conclusion

Ah, what a journey it was! All that hassle for writing a "Hello, world!" program...

We achieved our goal of cutting it down to a single syscall, but there are still a lot of areas we have just scratched the surface. There is also this kernel thingy that actually performs the action requested by a system call and so on, but that is a story for another day (or maybe a few years worth of stories).

In any case, I hope you enjoyed this little exploration, and maybe even learned a thing or two. If you have any questions, comments, or suggestions, feel free to reach out.

Single syscall "Hello, world" - part 1

Szymon Gibała — Wed, 01 Mar 2023 19:39:14 +0000

"Hello World" is the first program of many. Regardless of the programming language, we are learning it is a canonical example of how to create a program that simply prints "Hello, world!" to the screen.

One might then ask, how complex it really is? After all, it is just a single write(2) syscall, right?

NOTE: This post refers specifically to Linux, as I will use some Linux-only tools.

The basics

For the purpose of my "Hello, world" I want to use Rust. It is a modern language suitable for low-level programming, so it surely will have much less overhead than many others. Besides that, it is just a good language.

Let's get our "Hello, world" going:

fn main() {
    println!("Hello, world!");
}

We can now run it:

cargo run

Hello, world!

Everything works as expected! Well, that is not a big achievement, but hey we need to be happy with small things.

To see our write(2) syscall and get it over with we will use strace, a system call tracing tool for Linux:

-c option prints a summary of the system calls at the end of the trace. If you also want to see specific system calls with their arguments as they occur, use -C instead.

strace -c ./target/debug/hello-world

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ------------------
  0.00    0.000000           0         5           read
  0.00    0.000000           0         1           write
  0.00    0.000000           0         4           close
  0.00    0.000000           0         1           poll
  0.00    0.000000           0        13           mmap
  0.00    0.000000           0         5           mprotect
  0.00    0.000000           0         2           munmap
  0.00    0.000000           0         3           brk
  0.00    0.000000           0         5           rt_sigaction
  0.00    0.000000           0         2           pread64
  0.00    0.000000           0         1         1 access
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         3           sigaltstack
  0.00    0.000000           0         2         1 arch_prctl
  0.00    0.000000           0         1           sched_getaffinity
  0.00    0.000000           0         1           set_tid_address
  0.00    0.000000           0         4           openat
  0.00    0.000000           0         4           newfstatat
  0.00    0.000000           0         1           set_robust_list
  0.00    0.000000           0         2           prlimit64
  0.00    0.000000           0         1           getrandom
  0.00    0.000000           0         1           rseq
------ ----------- ----------- --------- --------- ------------------
100.00    0.000000           0        63         2 total

That is quite a bit more stuff than one might have expected... This begs the question then: Does the simple "Hello, world" need to do all of this? We should certainly do something about it.

Tracing complexity

Let's start by looking at those syscalls a bit closer and see if we can get an idea of what is going on:

strace ./target/debug/hello-world

The output is pretty verbose, so I chop it down to relevant pieces:

...
access("/etc/ld.so.preload", R_OK)      = -1 ENOENT (No such file or directory)
openat(AT_FDCWD, "/etc/ld.so.cache", O_RDONLY|O_CLOEXEC) = 3
newfstatat(3, "", {st_mode=S_IFREG|0644, st_size=44627, ...}, AT_EMPTY_PATH) = 0
mmap(NULL, 44627, PROT_READ, MAP_PRIVATE, 3, 0) = 0x7f9c5c9ae000
close(3)                                = 0
openat(AT_FDCWD, "/usr/lib/libgcc_s.so.1", O_RDONLY|O_CLOEXEC) = 3
read(3, "\177ELF\2\1\1\0\0\0\0\0\0\0\0\0\3\0>\0\1\0\0\0\0\0\0\0\0\0\0\0"..., 832) = 832
newfstatat(3, "", {st_mode=S_IFREG|0644, st_size=571848, ...}, AT_EMPTY_PATH) = 0
mmap(NULL, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7f9c5c9ac000
mmap(NULL, 127304, PROT_READ, MAP_PRIVATE|MAP_DENYWRITE, 3, 0) = 0x7f9c5c98c000
mmap(0x7f9c5c98f000, 94208, PROT_READ|PROT_EXEC, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x3000) = 0x7f9c5c98f000
mmap(0x7f9c5c9a6000, 16384, PROT_READ, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x1a000) = 0x7f9c5c9a6000
mmap(0x7f9c5c9aa000, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x1d000) = 0x7f9c5c9aa000
close(3)                                = 0
...
openat(AT_FDCWD, "/usr/lib/libc.so.6", O_RDONLY|O_CLOEXEC) = 3
read(3, "\177ELF\2\1\1\3\0\0\0\0\0\0\0\0\3\0>\0\1\0\0\0P4\2\0\0\0\0\0"..., 832) = 832
pread64(3, "\6\0\0\0\4\0\0\0@\0\0\0\0\0\0\0@\0\0\0\0\0\0\0@\0\0\0\0\0\0\0"..., 784, 64) = 784
newfstatat(3, "", {st_mode=S_IFREG|0755, st_size=1953472, ...}, AT_EMPTY_PATH) = 0
pread64(3, "\6\0\0\0\4\0\0\0@\0\0\0\0\0\0\0@\0\0\0\0\0\0\0@\0\0\0\0\0\0\0"..., 784, 64) = 784
mmap(NULL, 1994384, PROT_READ, MAP_PRIVATE|MAP_DENYWRITE, 3, 0) = 0x7f199ee8c000
mmap(0x7f199eeae000, 1421312, PROT_READ|PROT_EXEC, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x22000) = 0x7f199eeae000
mmap(0x7f199f009000, 356352, PROT_READ, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x17d000) = 0x7f199f009000
mmap(0x7f199f060000, 24576, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x1d4000) = 0x7f199f060000
mmap(0x7f199f066000, 52880, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_ANONYMOUS, -1, 0) = 0x7f199f066000
close(3)
...

We can see a lot of openats, newfstatats, mmaps, reads, and closes. And most of them refere to some dynamic shared object. In the above we can see: ld.so.cache, libgcc_s.so.1, libc.so.6.

While libgcc_s.so.1 and libc.so.6 are standard shared libraries, and ld.so.cache is basically a cache built by ldconfig. I was not really familiar with ld.so.preload, which, if we look at our system calls was not loaded successfully:
access("/etc/ld.so.preload", R_OK)      = -1 ENOENT (No such file or directory)
After a quick search, it turns out it works the same as LD_PRELOAD environment variable. It allows the user to specify ELF shared object that is loaded before all others. And indeed we can see it was accessed first, but since I do not have this file on my system, the result was ... = -1 ENOENT (No such file or directory).

We can correlate that a lot of those syscalls refer to the same files by looking at the file descriptor, which is a return value from openat(2) syscall:

openat(AT_FDCWD, "/usr/lib/libgcc_s.so.1", O_RDONLY|O_CLOEXEC) = 3

In this case, the file descriptor is 3. We can see 3 being passed to syscalls that follow, and if we consult manpages for those, we can verify that this argument is indeed expected to be a file descriptor (fd):

read(3, "\177ELF\2\1\1\0\0\0\0\0\0\0\0\0\3\0>\0\1\0\0\0\0\0\0\0\0\0\0\0"..., 832) = 832
newfstatat(3, "", {st_mode=S_IFREG|0644, st_size=571848, ...}, AT_EMPTY_PATH) = 0
...
mmap(NULL, 127304, PROT_READ, MAP_PRIVATE|MAP_DENYWRITE, 3, 0) = 0x7f9c5c98c000
...

Since those are dynamic libraries we do not explicitly touch their files in the code (only call functions etc.), as it is the job of the linker to make them available. This makes sense since most Rust targets are by default linked dynamically.
If we inspect our binary with file we can see it for ourselves:

file ./target/debug/hello-world

./target/debug/hello-world: ELF 64-bit LSB pie executable, x86-64, version 1 (SYSV), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=54d56ea3e059ced4d3b8cc088c409da6411264af, for GNU/Linux 4.4.0, with debug_info, not stripped

And in simple terms "dynamically linked" means that shared libraries are loaded into memory, and sections are mapped after the process is started.

Running the ldd on our binary shows us some of the same files we have seen in the strace output:

ldd ./target/debug/hello-world

    linux-vdso.so.1 (0x00007ffc75f26000)
    libgcc_s.so.1 => /usr/lib/libgcc_s.so.1 (0x00007fdc33996000)
    libc.so.6 => /usr/lib/libc.so.6 (0x00007fdc337af000)
    /lib64/ld-linux-x86-64.so.2 => /usr/lib64/ld-linux-x86-64.so.2 (0x00007fdc33a13000)

We have seen libgcc_s.so.1 and libc.so.6 being linked from our syscalls.
VDSO in linux-vdso.so.1 stands for virtual dynamic shared object and is used for some syscalls optimizations.
The last one remaining /usr/lib64/ld-linux-x86-64.so.2 is the linker itself. You can see it for yourself by trying to run it:

  /usr/lib64/ld-linux-x86-64.so.2 --help | head -n 4

  Usage: /usr/lib64/ld-linux-x86-64.so.2 [OPTION]... EXECUTABLE-FILE [ARGS-FOR-PROGRAM...]
  You have invoked 'ld.so', the program interpreter for dynamically-linked
  ELF programs.  Usually, the program interpreter is invoked automatically
  when a dynamically-linked executable is started.

So what all of it means for our problem is that before actually running our code that simply prints the "Hello, world!", the linker will do all this magic, open, memory map all dependencies, and so on.

While dynamic linking is great, it sounds like way too much work for a simple "Hello, world!". Let's try to cut it out...

Eliminating linker

Since we identified our first suspect that bloats output of our strace we can now eliminate it.

From the same Rust docs linked above we can read that is possible to link Rust with C runtime (crt) statically using crt-static target feature. We can pass it to the compiler using RUSTFLAGS:

RUSTFLAGS="-C target-feature=+crt-static" cargo build

Let's check our improvements in action:

strace -c ./target/debug/hello-world

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ------------------
  0.00    0.000000           0         2           read
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           close
  0.00    0.000000           0         1           poll
  0.00    0.000000           0         1           mmap
  0.00    0.000000           0         2           mprotect
  0.00    0.000000           0         1           munmap
  0.00    0.000000           0         5           brk
  0.00    0.000000           0         5           rt_sigaction
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         1           readlink
  0.00    0.000000           0         3           sigaltstack
  0.00    0.000000           0         2         1 arch_prctl
  0.00    0.000000           0         1           sched_getaffinity
  0.00    0.000000           0         1           set_tid_address
  0.00    0.000000           0         1           openat
  0.00    0.000000           0         1           newfstatat
  0.00    0.000000           0         1           set_robust_list
  0.00    0.000000           0         2           prlimit64
  0.00    0.000000           0         1           getrandom
  0.00    0.000000           0         1           rseq
------ ----------- ----------- --------- --------- ------------------
100.00    0.000000           0        35         1 total

This is significantly better as we dropped from 63 to 35 syscalls, but that is still
way more than we need. We can however confirm that our binary is now linked statically:

ldd ./target/debug/hello-world

    statically linked

An alternative way of building statically linked binary is to use musl libc instead of glibc. musl was designed with static linking in mind so it is worth giving it a shot. We can do that by specifying the x86_64-unknown-linux-musl target. We no longer need to pass RUSTFLAGS as static linking is a default behavior for musl target:

cargo build --target x86_64-unknown-linux-musl && strace -c ./target/x86_64-unknown-linux-musl/debug/hello-world

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           poll
  0.00    0.000000           0         1           mmap
  0.00    0.000000           0         1           mprotect
  0.00    0.000000           0         1           munmap
  0.00    0.000000           0         2           brk
  0.00    0.000000           0         5           rt_sigaction
  0.00    0.000000           0         3           rt_sigprocmask
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         3           sigaltstack
  0.00    0.000000           0         1           arch_prctl
  0.00    0.000000           0         1           set_tid_address
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0        21           total

Mind the different binary path in the target directory!

We dropped another few syscalls. It is pretty hard to tell "why" without diving into the actual source code of both glibc and musl. Both are completely different implementations of libc so as long as function interfaces are preserved, the implementation can handle things differently.

Coming back to our task, however, we are still quite far from the goal. Perhaps it is Rust that is at fault here? Maybe it was not a good choice after all...

Descending into C

There sometimes comes a time when you have to abandon your ideals, and just get the job done. This time is now. To verify if it is Rust runtime causing all those syscalls we can try to write the same program in good old C:

#include <stdio.h>

int main() {
    printf("Hello, world!\n");
}

Wasn't too bad... Since we already identified musl as a good candidate for static linking, we can build it with musl-gcc (a wrapper for gcc that links against musl):

musl-gcc -static main.c && ./a.out

Hello, world!

Let's see how it does:

strace -c ./a.out

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           ioctl
  0.00    0.000000           0         1           writev
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         1           arch_prctl
  0.00    0.000000           0         1           set_tid_address
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0         5           total

Now, that gets us much closer to what we want.

You may have noticed that the write syscall was replaced with writev(2). writev is simply a different version of write that allows writing multiple buffers at once (known as vectored I/O).
If we check the actual arguments passed to the syscall:

strace -e trace=writev ./a.out

-e option allows us to specify an expression that modifies events to trace and how to trace them. In our case, we want to trace only the writev syscall.

writev(1, [{iov_base="Hello, world!", iov_len=13}, {iov_base="\n", iov_len=1}], 2Hello, world!
) = 14
+++ exited with 0 +++

We can see that our string was split into two buffers, one for "Hello, world!" and another for the new line "\n".

Why? Well, it is complicated... Syscall itself comes somewhere from here. If we are adventurous enough and go up the stack we can find printf_core, which is called by vprintf, which can (indirectly) take us back to the printf itself...

There seems to be really a lot of code until we get to the actual syscall... I am sure it is all justified and so on, but for us, it sounds like a lot of unnecessary complexity.

Fortunately, we can just use the syscall directly bypassing all that magic:

#include <unistd.h>
#include <sys/syscall.h>

int main(void) {
    syscall(SYS_write, 1, "Hello, world!\n", 14);
}

We pass SYS_write as a first argument to syscall, which is nothing more than a constant that represents the syscall number (1 in the case of write). The rest of the arguments, are syscall specific, and as described in man page for write, those are:

file descriptor (1 for stdout)
buffer
number of bytes to write

Let's run it:

musl-gcc -static main.c && strace -c ./a.out

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           execve
  0.00    0.000000           0         1           arch_prctl
  0.00    0.000000           0         1           set_tid_address
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0         4           total

printf hidden another syscall (ioctl) from us, and we are back to a simple write.

This brings us down to four system calls remaining. It still sounds like more than necessary. As there are no more obvious things to chop off, it might be time to put down our axe and approach it with a bit more precision.

Last syscalls standing

Let's start with an easy one. We cannot really get rid of execve as something (in this case strace) needs to actually execute our program. So even tho we see it in strace output, it is "not really" our "Hello, world!" program that calls it.

When running strace, the process will fork, and starts tracing a child process. The child, therefore, needs to later execute the desired program that we pass as an argument (a.out binary in the case of our C program), to do that it calls the execve syscall.
We can take a peek at that by straceing the strace:
strace strace -c ./a.out
The output is a bit messy, but if we zoom in on important parts, we can see the clone syscall, which is used to create a new process, followed by ptrace with PTRACE_SEIZE argument as __ptrace_request, which attaches to the process with a pid that we got as a result of clone:
...
clone(child_stack=NULL, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7fb11c716550) = 4031
...
ptrace(PTRACE_SEIZE, 4031, NULL, PTRACE_O_TRACESYSGOOD|PTRACE_O_TRACEEXEC|PTRACE_O_TRACEEXIT) = 0
...
Only after that child process will run our program with the execve syscall, which we can see by straceing our binary:
strace -e trace=execve ./a.out
execve("./a.out", ["./a.out"], 0x7ffd96e60d30 /* 31 vars */) = 0
Hello, world!
+++ exited with 0 +++

We now know we cannot live without the execve, but what about arch_prctl and set_tid_address, then?

To the best of what I have found, those are responsible for setting up thread local storage (TLS).
As we can read in man pages for arch_prctl:

arch_prctl - set architecture-specific thread state

Digging a bit more, what this means is interfacing with FS (and GS) registers (FS in particular for TLS), which cannot be set from user space and is used to store per thread context.

Link to source code

Another syscall related to threading is set_tid_address ("set pointer to thread ID"). I did not find great sources on this one, but from reading the man page we can try to reason about it.
set_tid_address will set the clear_child_tid attribute of the given thread to the address specified by the system call. And as the name (clear_child_tid) suggests, when the thread terminates, the value at the address will be set to 0, or in other words, it will be cleared.

Why is it useful? Again a per man page, if applicable the kernel will then perform:

futex(clear_child_tid, FUTEX_WAKE, 1, NULL, NULL, 0);

which can be thought of as releasing the lock of a given memory location and waking up a single thread that is waiting on it. This does not happen for our program since we only have a single thread, so there is nothing to wake up.

If you are familiar with Go, this sound similar to sync.Cond.

Link to source code

Okay, we have a better idea of what those system calls do, and we can reasonably suspect that they come somewhere from libc (musl). At the same time, both of them are not necessary for simply printing Hello, world!. If only we could get rid of libc...

Look! There is one more door in this dark basement, and it leads to an even darker place...

Assembly

There is one language that we can "easily" reach for to write the "Hello, world!" in without all that overhead -- Assembly. We will use 64-bit x86 assembly as this is the machine I am running on.
So... brace yourself and create hello.asm:

section .text
   global _start

_start:
  mov rax, 1        ; write syscall number to rax - 1 is write
  mov rdi, 1        ; 1 is stdout file descriptor
  mov rsi, msg      ; msg is our "Hello, world!\n" defined in .rodata section
  mov rdx, msglen   ; specify message length in rdx - sizeof("Hello, world!\n")
  syscall           ; execute syscall

  mov rax, 60       ; write syscall number to rax - 60 is exit
  mov rdi, 0        ; write program exit code to rdi
  syscall           ; execute syscall

section .rodata
  msg: db "Hello, world!", 10
  msglen: equ $ - msg

Well, we got through it. The code is pretty simple, and if you do not speak assembly (do not worry me neither), comments on the right side should give you an idea of what is going on.

You may have noticed that we actually call syscall twice, which is not exactly what we wanted. However, the second call is just an exit syscall. Technically we could get rid of it and we would still get our Hello, world! printed on the screen.

The catch here is that the CPU would not know that our program is finished, and would try to run the next instruction, which is not there. So this would cause the CPU to try to read some memory that is not accessible by our program, and result in an error beloved by all C programmers:

Hello, world!
[1]    2529 segmentation fault (core dumped)  ./hello

Let's be nice to our CPU, accept the exit syscall as necessary, and do not count it for our "one syscall" goal. As we have seen before strace -c will not show it in the summary anyway.

In most cases one would likely prefer to use exit_group(2) syscall instead of exit(2), as it exits all threads in the process. That is what most (if not all) exit functions in different standard libraries do. You can see that is what our previous programs (both Rust and C) did by running strace.

For this case exit is completely sufficient.

With that in mind we can assemble the program with nasm assembler, and link it using ld:

nasm -f elf64 hello.asm
ld -static -o hello hello.o

and feed it to strace:

strace -c ./hello

Hello, world!
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
  0.00    0.000000           0         1           write
  0.00    0.000000           0         1           execve
------ ----------- ----------- --------- --------- ----------------
100.00    0.000000           0         2           total

And there we have it, "Hello, world!" stripped down to a single syscall! Doesn't victory taste sweet? If only not for this smell of Assembly everywhere, and a touch of C flashbacks... And yeah, I know, I know, it was supposed to be in Rust...

Okay, fine, let's look at the positives... At least now we have a chance to rewrite it in Rust...

We are going to embark on that journey in the next part.

Of Absence of Value and Hidden Footguns

Szymon Gibała — Mon, 12 Dec 2022 06:38:09 +0000

Of Absence of Value and Hidden Footguns

Programming languages differ in the expressiveness of their type system, ranging from very strict ones to those which just wing it.
In this post, I am not going to discuss this topic in any detail as I am not nearly qualified in this area to give it any justice. However, I would like to share my thoughts about one aspect related to the type system: representing the absence of value and one concrete example of how it impacts our programs.

Since I mostly work with Go and Rust this article touches on some specific aspects of Go and compares it to Rust, but since some other languages use similar constructs to Go and/or Rust the principles might apply to some of them as well.

Introduction

Be it a value that is set only in certain scenarios, a completely optional value, or one that is loaded lazily at some later stage of execution, in our programs we often need to represent whether something is or is not there.

To set some foundation, if you are not familiar with Go or Rust, here is a quick introduction without going into much detail.

In Go, we usually express that the value is not there with pointers, specifically with a special value nil that we can assign to any pointer type.

type SomeType struct {
    Name *string
    SomeField *OtherType
}
...
something := SomeType{
    Name: nil,
    SomeField: nil,
}

We then can check if the value is present by comparing it to nil. To access the value directly we dereference the pointer:

if something.Name != nil {
    *something.Name = "foo"
}
if something.SomeField != nil {
    // SomeField is not nil
}

Rust gives us a dedicated type to represent the value that might not be present -- Option<T>. Our fields can be set to either Some(T) or None, which represents the absence. Similar constructs exist in many modern languages:

pub struct SomeType {
    name: Option<String>,
    some_field: Option<OtherType>,
}

With Rust, we get several "options" on how to handle this wrapper type, and depending on the situation one might be more fitting than the others. The most straightforward approach would be to use a match expression, or if let Some(...) construct, but there are multiple other possibilities that can be very useful in certain situations such as using a function like map or propagating None with ? operator, or even outright checking if value is_none().

// Using match statement:
match something.some_field {
    Some(some_field) => { 
        // Value is not absent
    }
    None => {
        // Value is absent
    }
}
...

// Using `if let Some(...)`:
if let Some(some_field) = something.some_field {
    // Value is not absent
} else {
    // Value is absent
}

// Using ? operator:
fn do_something(something: SomeType) -> Option<String> {
    // If value is None, the function will return None right away
    let some_field = something.some_field?;
    ...
}

// There are few of other options...

Footguns

Our programs grow, sometimes we are tired or in a hurry, sometimes we jump into unknown codebases, or simply do other things for some time and our brain decides to discard pieces of context from the limited address space assigned to the project.

Sooner or later we forget that SomeField might not have any value and somewhere in our code we try to call a method on it or access its fields:

...
something.SomeField.DoSomething()
// or
if something.SomeField.Value == "foo" {
    ...
}
...

and then we may get an unpleasant surprise:

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x2 addr=0x0 pc=0x1029f6b64]
...

But we did not dereference the pointer, did we? Well, no, but Go did it for us as calling methods or accessing fields on a pointer type will dereference it and if the pointer value is nil, cause a runtime panic. The same holds for interfaces.

If x is of pointer type and has the value nil and x.f denotes a struct field, assigning to or evaluating x.f causes a run-time panic.

If x is of interface type and has the value nil, calling or evaluating the method x.f causes a run-time panic.

To be clear implicit dereference happens only in certain scenarios as described in the docs linked above. Simply comparing pointer type to non-pointer type will result in a compilation error:
// This does not work, we need to dereference explicitly wiht *
if something.Name == "foo" {
    ...
}
invalid operation: something.Name == "foo" (mismatched types *string and untyped string)

It is easy to forget a nil check for a field, and because of implicit dereferencing we might not even notice that the field is a pointer. Moreover, pointers are quite ubiquitous in Go, so if we are working with other people's code it might not be clear without some context if we need to perform a nil check or if the value is expected to be there always.

What happens in Rust's case, then?

...
something.some_field.do_something();
// or
if something.some_field.value == "foo" {
    ...
}
...

error[E0599]: no method named `do_something` found for enum `Option` in the current scope
  --> src/main.rs:27:26
   |
27 |     something.some_field.do_something();
   |                          ^^^^^^^^^^^^ method not found in `Option<OtherType>`
   |
note: the method `do_something` exists on the type `OtherType`
  --> src/main.rs:9:5
   |
9  |     pub fn do_something(&mut self) {
   |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
help: consider using `Option::expect` to unwrap the `OtherType` value, panicking if the value is an `Option::None`
   |
27 |     something.some_field.expect("REASON").do_something();
   |                         +++++++++++++++++

Rust will not compile the program as we are trying to call a function on a type that does not define it. We can still choose to unwrap() or expect(...) if we know value cannot be absent, if we are okay with our software crashing in this case, or if we just like living on the edge. However, Rust requires us to make this choice explicitly, eliminating a whole bunch of possible mistakes, or in other words, it politely asks us before we shoot ourselves in a foot.

Summary

Go does not have a clear way to represent the absence of value other than a nil assigned to the pointer type. And in some cases, the default behavior of the language is dereferencing the pointer, which frankly makes sense in a lot of scenarios, because pointers main purpose is not to represent that the value is not there but to reference data stored on the heap. This causes problems when we use pointers for other purposes, such as representing the absence of value. Whether the nil or NULL is a good idea in the first place is a different discussion.

Rust Option<T> on the other hand is very specific in its purpose of signaling that the value may or may not be there, so the default behavior of how it is handled can be tailored to this purpose, which in turn makes it easier to avoid mistakes.

Bookmark - command-line tool to quickly access obnoxious URLs

Szymon Gibała — Mon, 14 Sep 2020 17:39:52 +0000

Introduction

Bookmark is a simple command-line application written in Rust that allows saving, grouping, and tagging your URLs as well as accessing them quickly from the comforts of your terminal.

What does it do?

The main objective of Bookmark is to make it easier to find the URL you are looking for and open it quickly without clicking through dozens of folders with bookmarks or scrolling endlessly through the bookmarks bar.

It aims to minimize time and cognitive effort spend on switching to the browser and looking for the URL.

With the Bookmark you can:

Save URLs you are using often or just temporarily for some period of time when, for example working on some task.
Assign bookmarks into groups.
Tag them to make it easier to find.
Open URLs directly in the browser without touching your mouse.

How does it work?

The tool can work in two modes:

Standard command-line tool - you type commands and see the outputs in the terminal.
Interactive mode - the terminal user interface from which you can search, open, or modify your bookmarks.

Bookmark is most useful when working in the Interactive mode, as you can access it very quickly - just by typing bookmark in your terminal - and from there you can search for a bookmark you are looking for and simply press enter to open it in the browser saving yourself several clicks or these annoying moments trying to remember the exact URL.

Why have I written Bookmark?

I have often found myself opening a lot of the same URLs during my workday. This means switching from whatever IDE or terminal I was working with to the browser and then either clicking through my bookmarks folder or wrestling with Chrome search bar to suggest me the one URL I am currently looking for and never actually remember it (the fact that the word github is in 90% of those, probably is not very helpful for Chrome).

With Bookmark I can tag the URL with something like GitHub issue number or a specific topic and by typing it in the search bar have all the URLs at my disposal.

Adding the tag or group you came up with by yourself makes it easier to remember it as you have some connection with it as opposed to just some URL that your colleges send you and you probably did not even read it.

Bookmark may not be the most useful for websites like Twitter, Reddit, or Dev that are short and distinct. We use them all the time and our fingers already developed muscle memory for accessing them.

However, I believe it is a good tool for things like the URL for that another repository which differs just by a single word from the other fifteen internal repositories of your company as well as those ridiculously long domain names containing more dots than this entire post.

And yes, let's be honest the final obvious reason is that we developers love our terminals. ❤️

What did I use?

Bookmark is written in Rust, as this is the language I am currently learning (and really like!).

I have found it quite helpful in command-line based application, as it is not easy to test those end-to-end automatically, so the strict rules enforced by the compiler were invaluable. In the vast majority of cases when the code compiled, everything just worked, which saved me a lot of time and sanity looking for bugs or fixing regressions.

Of course, nothing is a silver bullet. I probably spend more time writing the code than I would in something like Go and a few bugs slipped in here and there anyways. Overall I am happy with my choice.

Some cool libraries used in the project are:

Clap - great library for writing command-line application.
tui-rs - library for creating awesome terminal user interfaces.

Summing up

You can find the project on GitHub.

It might not be the greatest tool on the planet it also is not particularly advanced but it still can smoothen the work just a little bit for some of us and spare ourselves some mental energy. If you are interested check it out and let me know if it is in any way useful for you!

If you have some suggestions or feedback let me know! Also, feel free to create an issue or shoot a pull request!

Building Rust Web API with Warp and Diesel

Szymon Gibała — Mon, 03 Aug 2020 17:21:37 +0000

Introduction

In this article, I would like to share with you my experience of writing a very simple Web API in Rust using Warp and Diesel.

As I am still Rust newbie, please let me know of any mistakes you have spotted, and of course, any feedback is appreciated.

Prerequisites

Basic knowledge of Rust
Basic knowledge of how web APIs work

Project overview

To not create another Todo List, we are going to create a simple book catalog (I know I know it is almost as original).

API

We will start by defining our API, it will consist of the following methods:

POST /api/v1/books - to add book to the catalog.
GET /api/v1/books - to list all our books.
PUT /api/v1/books/:id - to update the status of our book for example: ToRead, Reading, Finished, Rereading.
DELETE /api/v1/books/:id - to delete book from our collection

As mentioned before, we will use Warp as our web framework. It is based on composable Filters and I have found it quite easy to work with.

Database

To manage our database and connect it with our application, we will use Diesel, which is probably the most popular Rust ORM.

Diesel not only allows us to read and write to the database from our code but also provides a CLI tool to manage migrations.

We will use Postgres as a database but Diesel also supports other drivers like MySQL or SQLite.

Let's implement it!

Setup the project

First, we will create a new project with cargo:

cargo new rust-api-warp-and-diesel

Now let's declare dependencies for our application in Cargo.toml:

...

[dependencies]
tokio = { version = "0.2", features = ["macros"] }
warp = "0.2"
serde_derive = "1.0"
serde = "1.0"
log = "0.4"
pretty_env_logger = "0.3"
diesel = { version = "1.4.4", features = ["postgres", "r2d2"] }

To explain things quickly:

Warp is using tokio as an async runtime therefore we need it as a dependency.
We will also need serde to work with JSON.
For diesel we need postgres and r2d2 features for working with the Postgres database and creating a connection pool.
For some basic logging, we will use log and pretty_env_logger.

Setup database with Diesel

After we set up our project we can go ahead and start preparing our database. For that, we will need to install Diesel CLI. You can get a detailed guide on how to do it in Diesel getting start guide.

To setup up Diesel with our project we need to provide it with DATABASE_URL environment variable or the .env file. Let's create it now:

echo DATABASE_URL=postgres://postgres:password@localhost:5432/book_catalog > .env

To continue setup we will need a running database. You can use local Postgres or spin up an instance in Docker container:

docker run -p 5432:5432 --rm -e POSTGRES_PASSWORD=password postgres:12

And now we can run the setup:

diesel setup

This will create a book_catalog database in our Postgres and add some files to our project:

migrations directory is the place where our migrations live.
diesel.toml is a configuration file for diesel-cli for our project.

Now, let's add our first migration:

diesel migration generate book_catalog_initial_schema

Every migration is a subdirectory in the migrations and its name is a timestamp joined with the name we passed to the command. The migration consist of two SQL files:

up.sql for performing the migration.
down.sql for reverting it.

Database schema

Our database will be stupid simple with just one table representing our books. It obviously is far from perfect but it is enough for demonstration purposes.

In the up.sql we will simply create the table:

CREATE TABLE books (
    id BIGSERIAL PRIMARY KEY,
    title varchar(256) NOT NULL,
    author varchar(256) NOT NULL,
    status varchar(256) NOT NULL
);

and in the down.sql we will drop it:

DROP TABLE books;

You may ask why not use an enum for book status? Unfortunately, Diesel does not support enums out of the box so to keep it simple we will just use varchar and map it to the Rust enum in our code.
If you really need enums you can check out this create which makes it possible to use enums directly with Diesel.

The last step will be to run our migration on the database and generate schema.rs file:

diesel migration run

The file contains the table! macro which creates code based on our database schema to represent tables and columns.

If you would like to adjust the file name or its location, you can do so by modifying diesel.toml. For our case, the default is perfectly fine.

Define model

Before we start operating on the database we need to have an internal representation of our data. We will create our structs in the model.rs file.

use serde_derive::{Deserialize, Serialize};
use crate::schema::books;

#[derive(Serialize, Debug, Clone, Queryable)]
pub struct BookDTO {
    pub id: i64,
    pub title: String,
    pub author: String,
    pub status: BookStatus,
}

// Struct for creating Book
#[derive(Debug, Clone, Insertable)]
#[table_name = "books"]
pub struct CreateBookDTO {
    pub title: String,
    pub author: String,
    pub status: BookStatus,
}

This part if pretty straight forward. We declare two structs one of which - CreateBookDTO - will be used to create books, as it does not have an id field, which will be assigned by Postgres. The other one - BookDTO - will represent the whole book object. We will use it for queries.

Besides that, we specify the table_name and derive from some of the Diesel traits like Queryable for performing database queries and Insertable for performing inserts.

You may have noticed that in the case of BookDTO struct, we do not actually need to specify the table_name. That is because structs implementing Queryable are not related to a specific table. They just represent the result of a query with a specific type signature and therefore can be used with multiple tables.

We are still missing one thing which is the BookStatus enum. As I have mentioned before, enums are not supported in Diesel out of the box, and for Postgres to treat it as a text field (varchar(256) in our case), we need to implement two traits:

ToSql - to convert Rust enum value to text stored in the database.
FromSql - to match text from the database to Rust enum value.

Let's add it to our model.rs:

...
use diesel::serialize::{ToSql, Output, IsNull};
use diesel::pg::Pg;
use std::io::Write;
use diesel::{serialize, deserialize};
use diesel::deserialize::FromSql;
use diesel::sql_types::Text;

#[derive(Serialize, Deserialize, Debug, Copy, Clone, AsExpression, FromSqlRow)]
#[sql_type = "Text"]
pub enum BookStatus {
    WantToRead,
    Reading,
    Finished,
    Rereading,
}

impl ToSql<Text, Pg> for BookStatus {
    fn to_sql<W: Write>(&self, out: &mut Output<W, Pg>) -> serialize::Result {
        match *self {
            BookStatus::WantToRead => out.write_all(b"WANT_TO_READ")?,
            BookStatus::Reading => out.write_all(b"READING")?,
            BookStatus::Finished => out.write_all(b"FINISHED")?,
            BookStatus::Rereading => out.write_all(b"REREADING")?, 
        }
        Ok(IsNull::No)
    }
}

impl FromSql<Text, Pg> for BookStatus {
    fn from_sql(bytes: Option<&[u8]>) -> deserialize::Result<Self> {
        match not_none!(bytes) {
            b"WANT_TO_READ" => Ok(BookStatus::WantToRead),
            b"READING" => Ok(BookStatus::Reading),
            b"FINISHED" => Ok(BookStatus::Finished),
            b"REREADING" => Ok(BookStatus::Rereading),
            _ => Err("Unrecognized enum variant".into()),
        }
    }
}

Custom Errors

The last step for our model will be the custom error type. We will add it to a new errors.rs file. Let's define new enum - ErrorType - and new struct - AppError.

use std::fmt;
use warp::reject::Reject;

#[derive(Debug)]
pub enum ErrorType {
    NotFound,
    Internal,
    BadRequest,
}

#[derive(Debug)]
pub struct AppError {
    pub err_type: ErrorType,
    pub message: String,
}

impl AppError {
    pub fn new(message: &str, err_type: ErrorType) -> AppError {
        AppError { message: message.to_string(), err_type }
    }

    pub fn to_http_status(&self) -> warp::http::StatusCode {
        match self.err_type {
            ErrorType::NotFound => warp::http::StatusCode::NOT_FOUND,
            ErrorType::Internal => warp::http::StatusCode::INTERNAL_SERVER_ERROR,
            ErrorType::BadRequest => warp::http::StatusCode::BAD_REQUEST,
        }
    }

    pub fn from_diesel_err(err: diesel::result::Error, context: &str) -> AppError {
        AppError::new(
            format!("{}: {}", context, err.to_string()).as_str(),
            match err {
                diesel::result::Error::DatabaseError(db_err, _) => {
                    match db_err {
                        diesel::result::DatabaseErrorKind::UniqueViolation => ErrorType::BadRequest,
                        _ => ErrorType::Internal,
                    }
                }
                diesel::result::Error::NotFound => ErrorType::NotFound,
                // Here we can handle other cases if needed
                _ => {
                    ErrorType::Internal
                }
            },
        )
    }
}

impl std::error::Error for AppError {}

impl fmt::Display for AppError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "{}", self.message)
    }
}

impl Reject for AppError {}

ErrorType will help us to differentiate between different kinds of errors and map them properly to HTTP status codes in the to_http_status() method. For our application, we will only use three error types but you can add more if you need it.

We will also need to convert errors from Diesel to our AppError and for that, we have from_diesel_err(...). Note that we are mapping Diesel errors to a specific ErrorType so if we get diesel::result::Error::NotFound from the database, our API will properly respond with 404 status code.

Furthermore, the AppError implements standard traits like Display and Error but also one specific to Warp - Reject. This trait will allow us to pass AppError to the warp::reject::custom(...) function so that we can later use it while handling the rejections.

Implement data access

Now we have our database and model representing the entities. We can go ahead and write some code that will allow us to access the DB. The heavy lifting here is done by Diesel so we will just need a couple of simple methods. We will wrap them up with a DBAccessManager struct.

Let's create a new file for that and call it data_access.rs.

First, we will add the required imports and define the struct. It will contain database connection object which we will get from the connection pool - more on that later.

use diesel::prelude::*;
use diesel::r2d2::{ConnectionManager, PooledConnection};
use crate::model::{CreateBookDTO, BookDTO, BookStatus};
use crate::errors::{AppError,ErrorType};

type PooledPg = PooledConnection<ConnectionManager<PgConnection>>;

pub struct DBAccessManager {
    connection: PooledPg,
}

Now let's implement the first method.

impl DBAccessManager {
    pub fn new(connection: PooledPg) -> DBAccessManager {
        DBAccessManager {connection}
    }

    pub fn create_book(&self, dto: CreateBookDTO) -> Result<BookDTO, AppError> {
        use super::schema::books;

        diesel::insert_into(books::table) // insert into books table
            .values(&dto) // use values from CreateBookDTO
            .get_result(&self.connection) // execute query
            .map_err(|err| {
                AppError::from_diesel_err(err, "while creating book")
            }) // if error occurred map it to AppError
    }
}

For inserting data to the database we are using insert_into function, passing it the books::table generated by macro from schema.rs. Then we set values from our CreateBookDTO struct and finally we execute the query.

As a result, we are expecting to get either BookDTO or diesel::result::Error, therefore if an error occurs we use the previously prepared function AppError::from_diesel_err to map it to the AppError.

Let's add remaining methods for listing, updating, and deleting books.

impl DBAccessManager {

    ...

    pub fn list_books(&self) -> Result<Vec<BookDTO>, AppError> {
        use super::schema::books::dsl::*;

        books
            .load(&self.connection)
            .map_err(|err| {
                AppError::from_diesel_err(err, "while listing books")
            })
    }

    pub fn update_book_status(&self, book_id: i64, new_status: BookStatus) -> Result<usize, AppError> {
        use super::schema::books::dsl::*;

        let updated = diesel::update(books)
            .filter(id.eq(book_id))
            .set(status.eq(new_status))
            .execute(&self.connection)
            .map_err(|err| {
                AppError::from_diesel_err(err, "while updating book status")
            })?;

        if updated == 0 {
            return Err(AppError::new("Book not found", ErrorType::NotFound))
        }
        return Ok(updated)
    }

    pub fn delete_book(&self, book_id: i64) -> Result<usize, AppError> {
        use super::schema::books::dsl::*;

        let deleted = diesel::delete(books.filter(id.eq(book_id)))
            .execute(&self.connection)
            .map_err(|err| {
                AppError::from_diesel_err(err, "while deleting book")
            })?;

        if deleted == 0 {
            return Err(AppError::new("Book not found", ErrorType::NotFound))
        }
        return Ok(deleted)
    }
}

The code is pretty similar. We use filter([COLUMN_NAME].eq([VALUE])) as an equivalent of SQL WHERE statement and set([COLUMN_NAME].eq([NEW_VALUE]) for column updates. We use load() for querying multiple rows and execute() to run queries like update or delete.

In the case of update_book_status() and delete_book() methods we additionally check if any rows were affected and if that is not the case we return new error with type NotFound.

We can now import macros from diesel crate in our main.rs as well as declare our modules:

#[macro_use]
extern crate diesel;

mod model;
mod errors;
mod data_access;
mod schema;

Create API

Before we create our awesome Books API, let's start with something simple to get the taste of Warp.

We will start with simple HelloWorld handler so let's replace our main function with the following:

...

use std::env;
use warp::{Filter, reject};
use log::{info};

#[tokio::main]
async fn main() {
    if env::var_os("RUST_LOG").is_none() {
        env::set_var("RUST_LOG", "info");
    }
    pretty_env_logger::init();

    let routes = warp::path!("hello").map(|| "Hello World!".to_string());

    info!("Starting server on port 3030...");

    // Start up the server...
    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}

We initialize or Filter using warp::path! macro and specify the path to hello. Then we extend it with the map function which simply returns Hello World! string. By default, the response will have a 200 status code.

Then we are just starting our server on port 3030.

We can now run it with cargo:

cargo run

And verify if it is working correctly using curl:

curl localhost:3030/hello -v

We should get 200 response with:

Hello World!

Add database connection pool

To access the database we need the database connection and we will need it for handling every request. Initializing connection every time someone calls our API would be expensive so as mentioned in previous sections, we will use the connection pool.

To create a connection pool we will use the r2d2 feature from Diesel. First, we need a function to create our connection pool. Let's add it to the main.rs:

use diesel::pg::PgConnection;
use diesel::r2d2::{ConnectionManager, Pool};

type PgPool = Pool<ConnectionManager<PgConnection>>;

fn pg_pool(db_url: &str) -> PgPool {
    let manager = ConnectionManager::<PgConnection>::new(db_url);
    Pool::new(manager).expect("Postgres connection pool could not be created")
}

Instead of passing the connection object itself, we will wrap it with the DBAccessManager that we have created earlier.

use crate::data_access::DBAccessManager;
use crate::errors::{AppError, ErrorType};

fn with_db_access_manager(pool: PgPool) -> impl Filter<Extract = (DBAccessManager,), Error = warp::Rejection> + Clone {
    warp::any()
        .map(move || pool.clone())
        .and_then(|pool: PgPool| async move {  match pool.get() {
            Ok(conn) => Ok(DBAccessManager::new(conn)),
            Err(err) => Err(reject::custom(
                AppError::new(format!("Error getting connection from pool: {}", err.to_string()).as_str(), ErrorType::Internal))
            ),
        }})
}

This function will get a connection from the pool, use it to create DBAccessManager, and append it to the parameters tuple of the Filter, we will see this in action when we will be setting up our filters.

Creating handlers

Before we stitch everything together let's create structs and handlers for our endpoints. We will do it in a dedicated file api.rs.

First, we need structs that will represent the JSON objects that our API will be receiving and responding with:

use serde_derive::{Deserialize, Serialize};
use crate::model::{BookStatus, CreateBookDTO};

#[derive(Debug, Deserialize, Clone)]
pub struct AddBook {
    pub title: String,
    pub author: String,
    pub status: BookStatus,
}

#[derive(Debug, Deserialize, Clone)]
pub struct UpdateStatus {
    pub status: BookStatus,
}

#[derive(Debug, Serialize, Clone)]
pub struct IdResponse {
    pub id: i64,
}

impl IdResponse {
    pub fn new(id: i64) -> IdResponse {
        IdResponse{id}
    }
}

We can also add a method to the AddBook struct, to convert it to the CreateBookDTO that we use later:

...
impl AddBook {
    pub fn to_dto(&self) -> CreateBookDTO {
        CreateBookDTO{
            title: self.title.clone(),
            author: self.author.clone(),
            status: self.status.clone(),
        }
    }
}

Before adding handler methods, let's add one more function, that will take a Result and based on that respond either with an object serialized to JSON or an error. Here we will leverage the Reject trait implemented for our AppError:

...
use crate::AppError;

fn respond<T: Serialize>(result: Result<T, AppError>, status: warp::http::StatusCode) -> Result<impl warp::Reply, warp::Rejection> {
    match result {
        Ok(response) => {
            Ok(warp::reply::with_status(warp::reply::json(&response), status))
        }
        Err(err) => {
            log::error!("Error while trying to respond: {}", err.to_string());
            Err(warp::reject::custom(err))
        }
    }
}

To serialize the struct to JSON it needs to implement the Serialize trait, therefore T: Serialize.

Now we can use it in every handler method we create. We can add them now to api.rs:

...
use crate::data_access::DBAccessManager;
use serde::Serialize;

pub async fn add_book(
    db_manager: DBAccessManager,
    new_book: AddBook,
) -> Result<impl warp::Reply, warp::Rejection> {
    log::info!("handling add book");

        let create_book_dto = new_book.to_dto();

    let id_response = db_manager.create_book(create_book_dto).map(|book|
        { IdResponse::new(book.id) }
    );

    respond(id_response, warp::http::StatusCode::CREATED)
}

pub async fn update_status(
    book_id: i64,
    db_manager: DBAccessManager,
    status_update: UpdateStatus,
) -> Result<impl warp::Reply, warp::Rejection> {
    log::info!("handling update status");

    let id_response = db_manager.update_book_status(book_id, status_update.status).map(|_|
        { IdResponse::new(book_id) }
    );

    respond(id_response, warp::http::StatusCode::OK)
}

pub async fn delete_book(
    book_id: i64,
    db_manager: DBAccessManager,
) -> Result<impl warp::Reply, warp::Rejection> {
    log::info!("handling delete book");

    let result = db_manager.delete_book(book_id).map(|_| -> () {()});

    respond(result, warp::http::StatusCode::NO_CONTENT)
}

pub async fn list_books(
    db_manager: DBAccessManager,
) -> Result<impl warp::Reply, warp::Rejection> {
    log::info!("handling list books");

    let result = db_manager.list_books();

    respond(result, warp::http::StatusCode::OK)
}

We have four simple methods:

add_book to add a new book to our collection.
update_status to update the status of the specified book.
delete_book to delete the book.
list_books to list all of our books.

As you can see all of our handlers are async functions and their logic is quite simple:

Log that the method is called.
Call a method from DBAccessManager.
Map the result to the desired struct.
Respond with a JSON object or an error.

We could get away by not defining new methods for our handlers as their logic is quite trivial, but I find it useful to decouple it from all the Filters setup, that we will do in our main file. This would be much more apparent in the case of more complex applications.

Before we move on we need to declare a new module in our main.rs file.

...
mod api;
...

Handling rejections

We will add one more function that will help us handle rejections. Because we implemented the Reject trait for the AppError, we can now extract it from the warp::Rejection struct. We will try to do it in the handle_rejection function. Let's add it to errors.rs:

...

use std::convert::Infallible;
use warp::{Rejection, Reply};
use serde_derive::Serialize;

#[derive(Serialize)]
struct ErrorMessage {
    code: u16,
    message: String,
}

pub async fn handle_rejection(err: Rejection) -> Result<impl Reply, Infallible> {
    let code;
    let message;

    if err.is_not_found() {
        code = warp::http::StatusCode::NOT_FOUND;
        message = "Not Found";
    } else if let Some(app_err) = err.find::<AppError>() {
        code = app_err.to_http_status();
        message = app_err.message.as_str();
    } else if let Some(_) = err.find::<warp::filters::body::BodyDeserializeError>() {
        code = warp::http::StatusCode::BAD_REQUEST;
        message = "Invalid Body";
    } else if let Some(_) = err.find::<warp::reject::MethodNotAllowed>() {
        code = warp::http::StatusCode::METHOD_NOT_ALLOWED;
        message = "Method Not Allowed";
    } else {
        // In case we missed something - log and respond with 500
        eprintln!("unhandled rejection: {:?}", err);
        code = warp::http::StatusCode::INTERNAL_SERVER_ERROR;
        message = "Unhandled rejection";
    }

    let json = warp::reply::json(&ErrorMessage {
        code: code.as_u16(),
        message: message.into(),
    });

    Ok(warp::reply::with_status(json, code))
}

Here we try to extract different errors from the warp::Rejection struct and map it to proper HTTP status code.

For serializing the error response to JSON we use simple struct - ErrorMessage - and use warp::reply::with_status(...) to respond with a proper HTTP status code.

Connecting the pieces

Before we make use of our handlers we need to add one more filter, to decode the request body from JSON and append it to the parameters tuple.

...
use serde::de::DeserializeOwned;

fn with_json_body<T: DeserializeOwned + Send>(
) -> impl Filter<Extract = (T,), Error = warp::Rejection> + Clone {
    // When accepting a body, we want a JSON body
    // (and to reject huge payloads)...
    warp::body::content_length_limit(1024 * 16).and(warp::body::json())
}

We will create every route as a separate function:

...
use crate::api::{AddBook, UpdateStatus};

/// POST /books
fn add_book(
    pool: PgPool
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
    warp::path!("books")                    // Match /books path
        .and(warp::post())                  // Match POST method
        .and(with_db_access_manager(pool))  // Add DBAccessManager to params tuple
        .and(with_json_body::<AddBook>())   // Try to deserialize JSON body to AddBook
        .and_then(api::add_book)            // Pass the params touple to the handler function
}

The rest of the methods follow a similar structure:

...
/// GET /books
fn list_books(
    pool: PgPool
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
    warp::path!("books")
        .and(warp::get())
        .and(with_db_access_manager(pool))
        .and_then(api::list_books)
}

/// PUT /books/:id
fn update_status(
    pool: PgPool
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
    warp::path!("books" / i64 )
        .and(warp::put())
        .and(with_db_access_manager(pool))
        .and(with_json_body::<UpdateStatus>())
        .and_then(api::update_status)
}

/// DELETE /books/:id
fn delete_book(
    pool: PgPool
) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
    warp::path!("books" / i64 )
        .and(warp::delete())
        .and(with_db_access_manager(pool))
        .and_then(api::delete_book)
}

Now we will add the final method to combine all the previously created filters into a single one, that will be passed to warp::serve.

...
fn api_filters(
    pool: PgPool
) -> impl Filter<Extract=impl warp::Reply, Error=warp::Rejection> + Clone  {
    warp::path!("api" / "v1" / ..)   // Add path prefix /api/v1 to all our routes
        .and(
            add_book(pool.clone())
                .or(update_status(pool.clone()))
                .or(delete_book(pool.clone()))
                .or(list_books(pool))
        )
}

Finally, let's update our main function to finalize our API.

We will read the database connection string from the DATABASE_URL environment variable using env::var("DATABASE_URL"). Thanks to that we can reuse the .env file created for Diesel.

We will also use previously prepared functions to create our database connection pool and combined filter with the API endpoints.

Last but not least we will use recover function on the filter and pass it handle_rejection so that it will be called when the request will not match any filters or the error will be returned.

...

#[tokio::main]
async fn main() {
    if env::var_os("RUST_LOG").is_none() {
        env::set_var("RUST_LOG", "info");
    }
    pretty_env_logger::init();

    let database_url = env::var("DATABASE_URL").expect("DATABASE_URL env not set");

    let pg_pool = pg_pool(database_url.as_str());

    let routes = api_filters(pg_pool)
        .recover(errors::handle_rejection);

    info!("Starting server on port 3030...");

    // Start up the server...
    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
}
...

Now we should be able to successfully compile the application.

Run the application

First, let's make sure we still have our database up and running. If that is the case we need to set the DATABASE_URL environment variable, we can do it manually or leverage existing .env file that we created for Diesel:

export $(cat .env | xargs)

Now we can run the application again using cargo:

cargo run

We should see some logs indicating that the application has started:

 INFO  rust_api_warp_and_diesel > Starting server on port 3030...
 INFO  warp::server             > Server::run; addr=V4(127.0.0.1:3030)
 INFO  warp::server             > listening on http://127.0.0.1:3030

Use the API

Now that everything is up and running let's make some calls!

Add book, that we are currently reading:

curl localhost:3030/api/v1/books -X POST -d '{"title":"Game of Thrones", "author": "George R.R. Martin", "status":"Reading"}' -H "Content-Type: application/json"

{"id":1}

We should get back the id that we now can use to update the book status:

curl localhost:3030/api/v1/books/1 -X PUT -d '{"status":"Finished"}' -H "Content-type: application/json"

{"id":1}

Let's list books to see that the status was updated:

curl localhost:3030/api/v1/books

[{"id":1,"title":"Game of Thrones","author":"George R.R. Martin","status":"Finished"}]

And finally, we can also delete it:

curl localhost:3030/api/v1/books/1 -X DELETE

Summing up

Obviously our application is very simplistic and far from perfect, there are tons of things that we would have to do to make it even close to the production quality, but it is enough to get started with something and learn some fundamentals of Warp and Diesel.

If you have any suggestions or feedback please let me know!

Want to get your hands dirty?

A good way to learn new things in software space (at least for me) is to take an existing piece of code and add something to it as you exercise both your code reading and understanding skills as well as writing skills.

If you are up for the challenge and want to get your hands a little dirty, try to implement another endpoint (GET /api/v1/books/:id) that will return a book with a specified ID.

The following curl:

curl localhost:3030/api/v1/books/1

should return the book with ID 1, for example:

{"id":1,"title":"Game of Thrones","author":"George R.R. Martin","status":"Finished"}

You can clone the code from the repository.
If you will struggle with doing it by yourself, don't worry! The answer is available on the get-book-endpoint branch.