System design
What Concurrency Actually Means, Explained with Rust
Concurrency is grossly misunderstood and often conflated into parallelism. Parallelism means things run in simultaneously**. Concurrently** means to work on more than one thing at once but not simultaneously. While it’s hard to find a dictionary that truly differentiates between the two words in a way that we do in the computer science world I was able to find somewhere via the third person that did (as in the source is dead) differentiate between the two.
According to Lexico, the word simultaneously means “At the same time”.
And the word concurrently means “Existing, happening, or done at the same time”. — Stack Exchange
The key difference here is “existing, happening or done” at the same time. If it’s done at the same time then it must be done by one thing. If it simply exists at the same time then that implies parallelism because there’s no condition on the actor(s).
The Waiter Analogy
In the often given restaurant and waiter analogy a restaurant that can serve patrons in parallel has two or more waiters serving customers. A restaurant that is concurrent would have one server working many tables at once.
Obviously we wouldn’t have a waiter serve only one table at a time waiting from the time one patron has started eating until they left to serve a new table. This would cause a huge line out the door and many people would leave. This is how by default a TCP Socket works in Unix.
A restaurant with concurrency has waiters that serve more than one table at once going to tables as they need something from him. A restaurant without concurrency would be more of a private dining experience not a restaurant.
A restaurant that operates with only parallelism has multiple waiters working at once to serve their clients but those waiters can only serve on table at a time.
A restaurant that works with concurrency and with parallelism is any restaurant with more than one waiter and waiters are required to work more than one table at a time.
In our analogy waiters can be threads and concurrency is the methodology, which includes the underlying data structures threads use to manage incoming and outgoing I/O. Remember I/O could be files, connections, disk reads, device communication etc.
**Writing a Simple TCP Server
Taking directly from the Rust Book and modifying the basic example to return a value we get this:
use std::{net::{TcpListener, TcpStream}, io::{Write, Read, BufReader, BufRead}, iter::StepBy, fs};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let mut stream = stream.unwrap();
handle_connection(stream);
}
}
fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&mut stream);
let http_request: Vec<_> = buf_reader
.lines()
.map(|result| result.unwrap())
.take_while(|line| !line.is_empty())
.collect();
let status_line = "HTTP/1.1 200 OK";
let contents = "<html><head></head><body><h1>Hello, World!</h1></body></html>";
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}All we’re doing here is creating a TCP listener, bind it to the port on localhost we want the server to listen on and then loop over listener.incoming() which gives us a stream to write a response to. Once we write all of our response we flush the buffer telling the server to send everything now and do everything all over again.
The problem with this example is that it’s not concurrent nor is it doing anything in parallel. The server is waiting for one request to completely finish before accepting another. This is akin to a restaurant having one waiter, and standing at one table from smart to finish while other diners line up out the door.
Let’s run an autocannon test.
Autocannon with our simple server
That load test is abysmal.
What we’re seeing with latency, which is a measure of pure speed is bad. The 99th percentile of latency goes to 36 seconds. That’s horrid!
Our requests per second which is our measure of concurrency is somewhat respectable if this were a slower framework in Node.JS but this is Rust. We should see over 100k requests per second on this machine (M1 Mac).
So what’s going on? Well when Rust implements a TCP listener, the API abstracts a lower level API that’s offered by the kernel in the form of system calls commonly called syscalls . Syscalls can be used for everything from memory and file management to audit log management. A list of Mac syscalls for example can be found here.
The syscalls that are called on a linux machine (syscalls differ by OS) to create a TCP server are the following:
tcp_listen_fd = socket(TCP)— Create a new TCP socket and assign to a file descriptor which is a number that represents a file, a socket, connection, stream or other system resource.bind(tcp_listen_fd, 3000)— Bind the file descriptor representing the new socket to port 3000.listen(tcp_listen_fd)— Listen on the newly created TCP socket for new connectionstcp_conn_fd = accept(tcp_listen_fd)— Sleep until new connection; When there is a new connection accept the incoming connection and assign them to another file descriptor for later reference.bytes = recv(tcp_conn_fd)— Sleep until they send me something. When they send me something receive those bytes and assign it to a variable for processing.send(tcp_conn_fd, bytes)— Send bytes back to the connection and onto the client connecting to us.close(tcp_conn_fd)— Close the connection via it’s file descriptor because we’re done with itclose(tcp_listen_fd)— Close the TCP socket via it’s file descriptor because we’re done with our server and we’re shutting down
Technically speaking the problem with our server is that recv is what we call blocking IO. What blocking IO means is that it’s a waiter that’s not concurrent nor parallel. It’s going to server a request, one-at-a-time from start until finish while not responding to any other requests because by default no queuing mechanism is implemented.
When Rust implements the for stream in listener.incoming() { line, it’s really implementing recv hence our server is blocking.
Concurrency Starts In the Kernel
Before we start to make our server concurrent we need to understand where concurrency starts. It starts in the kernel as I’ve alluded to above. In every operating system the kernel is responsible for managing incoming and outgoing connections and any other I/O such as files, devices, or disks. This means no matter the language you’re using to implement a server, be it Python, JavaScript, Rust, or Java, the language ultimately communicates with the kernel via syscalls.
The base on which the kernel does this is via “file descriptors”. In a Unix system every device, file, directory, connection etc. is assigned a file descriptor which is a positive integer. When a connection is opened for example there’s a syscall to the kernel to create the connection which returns a file descriptor that points to that connection. That file descriptor is put into a table in the kernel that tracks the descriptor along with any properties it may have such as if it’s ready or not.
So because by default, without any further implementation a TCP socket isn’t concurrent we need to lean on the kernel to help us make it concurrent. Enter select , poll and epoll .
These are ways of multiplexing I/O. Technically, I/O is blocking no matter what when we’re doing anything meaningful with the file descriptor that’s because the system call to recv or write on the file descriptor is blocking no matter what so claims like “Node uses non-blocking I/O” are technically incorrect but they’re accurate enough to be in common usage. That disclaimer aside, the most modern syscall implementation for frameworks to build a server off of is epoll. Redis, Node, Tokio and Go are some of the technologies that are known to use epoll.
Why Not Select?
Select is the earliest addition to the Linux kernel to allow multiplexing and it’s also the least efficient. It uses a level triggered mechanism meaning that it keeps notifying us as data comes in.
Select’s API works by passing select a set of file descriptors you’d like to monitor and returns the ready count. Then you must loop through the ready count to find to do I/O. Sounds good so far right? Nope. The implementation is incredibly inefficient. If you have only one file descriptor you’d like to monitor and say your file descriptor is 777, then it will loop through 777 iterations to find your ready file descriptor. File descriptors need to be re-initialized on each “event-loop” iteration and finally, these things can cause issues with your call stack causing crashes and useless debugger call traces.
What about Poll?
Poll is quite a bit more efficient. It’s what we call “level-triggered” multiplexing which means it’s triggered to a “ready” state each time more data comes in (and subsequently not-ready once that data has been read) in an endless cycle. However, this is inefficient because when you give poll a list of file descriptors to use the “event-loop” loops over every file descriptor which is ready or not for an O(n) time. Poll exists in current OSX versions. Poll’s O(n) performance exists in Windows as the select system call oddly enough.
We can see our Rust Poll implementation and results below.
Enter E-Poll
Unix’s e-poll is what most modern day, highly concurrent system are built upon. E-poll is a kernel based API and data structure that allows us to loop over file descriptors that are ready and only ready. While it means that it’s worst case is O(n), it’s normally O(ready file descriptors). This is a huge benefit since we’re not looping through file descriptors that are not ready as in poll and we’re not looping all file descriptors and then copying as in select. In 2024 E-Poll is the best we’ve got at the kernel level for ensuring we’re as concurrent as possible.
Now, e-poll is Linux specific and OSX (the machine I’m writing on) is not exactly Linux. It uses the Free BSD derived kernel and not a true Linux derived kernel. While this means there’s a lot of crossover between linux and the kernel API and the OSX kernel API unfortunately E-Poll isn’t on OSX’s menu.
OSX does have an equivalent however with a similar API and the same big-o results the equivalent methods. OSX’s top concurrency API and equivalent to e-poll is called KQueue. To help us write our example on OSX we’re going to use a library called Mio which abstracts out the Windows, Linux and Mac concurrency APIs into one library. This is a standard practice when writing cross-platform highly concurrent apps. Node.JS for example uses Libuv to accomplish this.
Let’s implement e-poll via Mio in Rust and throw it to autocannon.
use std::collections::HashMap;
use std::io;
use std::io::{Read, Write};
use std::net::{SocketAddr};
use mio::{Events, Interest, Poll, Token};
use mio::event::Source;
use mio::net::{TcpListener, TcpStream};
/*
Checks for /r/n/r/n in a window of 4 bytes in the request header
r/n/r/n Signifies end of header and that the connection is ready
*/
fn is_double_crnl(window: &[u8]) -> bool {
window.len() >= 4 &&
(window[0] == b'\r') &&
(window[1] == b'\n') &&
(window[2] == b'\r') &&
(window[3] == b'\n')
}
fn main() {
let address: SocketAddr = "0.0.0.0:7878".parse().unwrap();
// assign server token id 0
const SERVER: Token = Token(0);
// counter will be used for associating socket with an event
let mut counter: usize = 0;
// create new poll instance
let mut poll = Poll::new().unwrap();
// Use TCP Listener from Mio because it implements a trait used in poll.register
let mut listener = TcpListener::bind(address).unwrap();
// Register server socket in poll registry; epoll will wake up and notify us of new signals
// from our server after it's registered
poll.registry().register(&mut listener, SERVER, Interest::READABLE | Interest::WRITABLE).unwrap();
// Create Epoll Events Holder
let mut events = Events::with_capacity(1024);
// HashMap to store our sockets; token AKA socket id is the key
let mut sockets: HashMap<Token, TcpStream> = HashMap::new();
// HashMap to store requests; token AKA socket ID is the key
let mut requests: HashMap<Token, Vec<u8>> = HashMap::new();
// Create buffer for reading writing to socket
let mut buffer = [0 as u8; 1024];
// Infinite loop to keep server running
loop {
// Poll for events
poll.poll(&mut events, None).unwrap();
// loop through socket events
for event in events.iter() {
match event.token() {
Token(0) => loop {
match listener.accept() {
Ok((mut socket, _)) => {
// Increment our counter; remember counter is unique id for temporary socket created from incoming request
counter += 1;
// Tokenize counter for mio
let token = Token(counter);
// Reregister socket and flag as readable now that we've accepted the connection
socket.register(poll.registry(), token, Interest::READABLE).unwrap();
sockets.insert(token, socket);
requests.insert(token, Vec::with_capacity(192));
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock =>
// Break loop because we're doing blocking IO; accept no more tokens
// Loop will initiate when we're done blocking
break,
Err(e) =>
panic!("Unexpected error: {}", e)
}
},
token if event.is_readable() => {
// loop to continuously read into buffer
loop {
// read request into buffer; returns number of bytes read
let bytes_read = sockets.get_mut(&token).unwrap().read(&mut buffer);
match bytes_read {
Ok(0) => {
// Successful read of zero bytes means connection is closed
sockets.remove(&token);
break;
},
Ok(n) => {
// Push buffer onto request
let req = requests.get_mut(&token).unwrap();
for b in &buffer[0..n] {
req.push(*b);
}
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(e) => panic!("Unexpected error: {}", e)
}
}
// Check request for /r/n/r/b which signifies end of request headers
// End of request headers signifies that the request
let ready = requests.get(&token).unwrap()
.windows(4)
.any(is_double_crnl);
if ready {
let socket = sockets.get_mut(&token).unwrap();
// If the socket is ready, flag as writable
socket.reregister(poll.registry(), token, Interest::WRITABLE).unwrap();
}
},
token if event.is_writable() => {
let status_line = "HTTP/1.1 200 OK";
let contents = "<html><head></head><body><h1>Hello, World!</h1></body></html>";
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
// clear request data for token
requests.get_mut(&token).unwrap().clear();
// write all bytes to socket
sockets.get_mut(&token).unwrap().write_all(response.as_bytes()).unwrap();
let socket = sockets.get_mut(&token).unwrap();
// re-register socket as readable
socket.reregister(poll.registry(), token, Interest::READABLE).unwrap();
}
_ => unreachable!()
}
}
}
}Code Rundown
A quick look at the code shows our process. We start by creating a TCP Listener and registering it with our E-Poll instance. We create an infinite loop to keep our server running. One in loop we call poll which adds events as they come in through E-Poll in the kernel to a list-like data structure. From there we iterate over the events as they come into our server. If the token / file descriptor matches our TCP Listeners file descriptor we know it’s an incoming connection and we accept it, register it with our E-Poll registry as readable and add it to memory for use later.
Since its readable our event.is_readable() match clause picks up and reads the payload into the buffer. Once the incoming data is fully read in, we check it for\r\n\r\n which would signify the end of the headers in an HTTP request indicating that it’s fully read in and ready for a response. We mark the socket as writable now to prepare it for the response.
Now the our event.is_writeable() arm is picked up and we’re ready to write our response and drop the memory we allocated for our response.
As you can see E-Poll is event driven non-blocking I/O. The basis of what we’re doing is having E-Poll stream data to us as it comes in. Once the data comes in, we look parse it, store it and re-register it with E-Poll so E-Poll can schedule it back to our events list again when its ready. This is the equivalent of a waiter servicing multiple tables and being flagged down by their customers when needed which is what we call concurrency.
**Results
Autocannon with E-Poll (single thread)
As you can see, it’s a huge improvement. To be exact, it’s an improvement by over a order of magnitude; 1462% to be exact! Even though we’re back into the 100k realm which is where we should be on the M1 Macbook Pro, we’re not done. Remember our waiter analogy? Well, right now we’re concurrent but not parallel. We can implement threads.
Parallelism & Concurrency
We’re already concurrent. We can further improve our code by implementing threads to be both concurrent and parallel.
Threads are interesting. They can exist both on the kernel (aptly known as kernel threads) and there can be user-space abstractions that exist within the programming language we’re using (also very aptly called user-space threads). In kernel threads the OS is responsible for scheduling, terminating, monitoring, executing, saving and restoring threads via a syscall API. With user-space threads these responsibilities are implemented at the language level.
use std::collections::HashMap;
use std::io::{self, Read, Write};
use std::net::SocketAddr;
use mio::net::{TcpListener, TcpStream};
use mio::{Events, Interest, Poll, Token};
use crossbeam_channel::{unbounded, Sender, Receiver};
use std::thread;
use std::time::Duration;
const SERVER: Token = Token(0);
const WORKER_COUNT: usize = 10;
enum Message {
NewConnection(TcpStream, SocketAddr),
Shutdown,
}
struct Server {
listener: TcpListener,
sender: Sender<Message>,
}
struct Worker {
id: usize,
poll: Poll,
connections: HashMap<Token, Connection>,
receiver: Receiver<Message>,
next_token: usize,
}
struct Connection {
stream: TcpStream,
buffer: Vec<u8>,
}
impl Server {
fn new(addr: &str) -> io::Result<Self> {
let addr = addr.parse().unwrap();
let listener = TcpListener::bind(addr)?;
let (sender, receiver) = unbounded();
// Spawn threads and channel for communication
for id in 0..WORKER_COUNT {
let worker_receiver = receiver.clone();
thread::spawn(move || {
if let Err(e) = Worker::new(id, worker_receiver).run() {
eprintln!("Worker {} error: {}", id, e);
}
});
}
Ok(Self { listener, sender })
}
fn run(&mut self) -> io::Result<()> {
let mut poll = Poll::new()?;
poll.registry().register(&mut self.listener, SERVER, Interest::READABLE)?;
let mut events = Events::with_capacity(1024);
println!("Server listening on {}", self.listener.local_addr()?);
// Start infinite loop to accept connections; this keeps server running and is blocking wait on CPU
loop {
// Start epoll on kernel; use events struct for storing events
poll.poll(&mut events, Some(Duration::from_millis(100)))?;
// Once an event is added to the struct by the kernel which acts as a queue; the queued events are accepted in iter loop;
for event in events.iter() {
if event.token() == SERVER {
// accept connection
self.accept_connections()?;
}
}
}
}
fn accept_connections(&mut self) -> io::Result<()> {
// in accept connections we have a wait loop for accepting connections.
loop {
match self.listener.accept() {
Ok((stream, addr)) => {
println!("Accepted connection from: {}", addr);
self.sender.send(Message::NewConnection(stream, addr)).unwrap();
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
Err(e) => return Err(e),
}
}
Ok(())
}
}
impl Worker {
fn new(id: usize, receiver: Receiver<Message>) -> Self {
Self {
id,
poll: Poll::new().unwrap(),
connections: HashMap::new(),
receiver,
next_token: 1,
}
}
fn run(&mut self) -> io::Result<()> {
let mut events = Events::with_capacity(1024);
loop {
if let Ok(message) = self.receiver.try_recv() {
match message {
Message::NewConnection(stream, addr) => self.handle_new_connection(stream, addr)?,
Message::Shutdown => break,
}
}
self.poll.poll(&mut events, Some(Duration::from_millis(100)))?;
for event in events.iter() {
let token = event.token();
if event.is_readable() {
self.handle_readable(token)?;
}
if event.is_writable() {
self.handle_writable(token)?;
}
}
}
Ok(())
}
fn handle_new_connection(&mut self, mut stream: TcpStream, addr: SocketAddr) -> io::Result<()> {
let token = Token(self.next_token);
self.next_token += 1;
self.poll.registry().register(&mut stream, token, Interest::READABLE | Interest::WRITABLE)?;
self.connections.insert(token, Connection { stream, buffer: Vec::new() });
println!("Worker {} handling new connection from: {}", self.id, addr);
Ok(())
}
fn handle_readable(&mut self, token: Token) -> io::Result<()> {
let mut buffer = [0; 1024];
let mut closed = false;
if let Some(connection) = self.connections.get_mut(&token) {
match connection.stream.read(&mut buffer) {
Ok(0) => {
closed = true;
}
Ok(n) => {
connection.buffer.extend_from_slice(&buffer[..n]);
if connection.buffer.windows(4).any(|w| w == b"\r\n\r\n") {
self.poll.registry().reregister(&mut connection.stream, token, Interest::WRITABLE)?;
}
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(e) => return Err(e),
}
}
if closed {
if let Some(mut connection) = self.connections.remove(&token) {
self.poll.registry().deregister(&mut connection.stream)?;
println!("Worker {} closed connection: {:?}", self.id, token);
}
}
Ok(())
}
fn handle_writable(&mut self, token: Token) -> io::Result<()> {
let response = b"HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, World!";
let mut remove_connection = false;
if let Some(connection) = self.connections.get_mut(&token) {
match connection.stream.write(response) {
Ok(_) => {
println!("Worker {} sent response: {:?}", self.id, token);
connection.buffer.clear();
self.poll.registry().reregister(&mut connection.stream, token, Interest::READABLE)?;
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(_) => {
remove_connection = true;
}
}
}
if remove_connection {
// Deregister upon errored connection
if let Some(mut connection) = self.connections.remove(&token) {
self.poll.registry().deregister(&mut connection.stream)?;
println!("Worker {} removed connection due to write error: {:?}", self.id, token);
}
}
Ok(())
}
}
fn main() -> io::Result<()> {
Server::new("127.0.0.1:7878")?.run()
}So here you can see we refactored a bit. Large functions and excessive self referential data structures in Rust cause ownership and lifetime issues. Here we made our functions specific, e.g. accept, write, read so it became easier to avoid fighting the borrow checker. You can also notice we’re using channels for cross thread communication. You can also see we have three wait loops going on, each connection, worker and the server have their own loop. This means the server waits for incoming connections, each connection is awaited for updates, and each worker is awaited for updates as well.
Autocannon with E-Poll (multi-threaded)
Wow! That was unexpected. We see a -%40.23 change in performance. We see a higher max latency as well. This surely is a shock right? Well, not nessecarily. Let’s start by saying single-threaded event loops often outperform multi-threaded event loops for I/O bound problems. Why is this though? The reality is that managing threads introduces a lot of locking (locking == higher latency) and more steps for the OS to manage and synchronize resources for the threads (more steps == higher latency). When we have higher latency, generally our capacity for performant concurrent threads goes down.
Beyond introducing quite a bit of kernel level complexity, we get better cache locality with the single threaded version meaning that the code in the single threaded version is more likely to fit into one of the CPU caches and be closer to the processor than the larger multi-threaded version. We also get one epoll event loop which lowers the amount of syscalls the our code uses which helps because syscalls are not known to be the fastest.
While we covered a lot of in-the-weeds computer science concepts in the prior paragraphs, if we go back to our waiter analogy, we’d have to redefine the analogy a bit to to get it right since generally in a restaurant our parallelism and concurrency would generally lead to a more efficient restaurant. Our analogy would have to change to something along the lines of: even though we have multiple employees, servicing more than one table at a time, since our employees need instructions from the manager after every interaction the manager is the bottleneck due to his extra load. This works for us since threads are not intelligent autonomous individuals, they require instruction management from a manager (the OS) before continuing work.
Honorable Mention: Tokio
Throughout the article, I’ve used Mio because it operates at a very low level, close to the system’s kernel. Mio is designed to provide non-blocking I/O with minimal abstractions, which makes it a great option for developers who need fine-grained control. Mio is also cross-platform, allowing it to leverage OS-specific mechanisms like epoll (Linux), kqueue (macOS/BSD), or IOCP (Windows) for efficient event handling.
However, the more popular Rust asynchronous runtime is Tokio. Tokio builds on the foundation of Mio but adds higher-level abstractions, including an asynchronous paradigm called futures (similar to “promises” in other languages). This enables you to write non-blocking, concurrent code in a more ergonomic way.
In our previous single-threaded example, there is a subtle performance bottleneck. The Events iterator from Mio is blocking in nature, meaning that each iteration over incoming connections is handled synchronously. This introduces waiting time whenever a connection is accepted, as the thread waits for the syscall to complete before moving on to the next task.
Tokio allows us to mitigate this issue by leveraging asynchronous futures. Instead of waiting for the syscall to complete, you can “fire and forget,” continuing with other tasks while Tokio handles the connection asynchronously in the background. By combining epoll (or the OS-specific equivalent) for kernel-level event notification with futures for asynchronous task management in the code, Tokio makes it easier to achieve complete concurrency, even in a single-threaded environment.
For those wondering, yes, this implies even better benchmarks exist when using Tokio. Comment and share this article on LinkedIn and I’ll write a follow up explaining futures in Rust!