Recently I've started Codecrafters' Build your own Redis (which I highly recommend!). One of the first steps is to make a TCP server that can accept more than one connection. I noticed that all the provided solutions use threads or IO#Select, and none of them use a fiber-based solution with Async. In this really short article, I'll show how one could achieve the result using Async.
A brief intro: Ruby Fibers are cooperative concurrency primitives that enable context switching without threads by maintaining their own execution stack. The Async library implements a non-blocking reactor pattern using these Fibers to schedule I/O operations across an event loop. With Ruby 3's introduction of the Fiber scheduler API, Async can now intercept potentially blocking operations and transform them into non-blocking equivalents through event-driven callbacks. The result is natural, synchronous-looking code that executes asynchronously with lower memory overhead compared to thread-based solutions, providing developers with an elegant way to handle concurrent operations without sacrificing readability or performance.
A one-connection Redis server
Now, let's move on to the task. In the beginning, we have something like this:
class YourRedisServer
def initialize(port)
@port = port
end
def start
server = TCPServer.new(@port)
loop do
client = server.accept # blocking
while client.gets do # blocking as well
process_input($_)
end
end
end
def process_input(input); end
end
YourRedisServer.new(6379).start
What happens here? We seem to have an infinite loop; however, it's not actually looping. That's because we have blocking operations. First is server.accept: the code won't move on unless a connection is established. Next, after a connection is established, we have while client.gets that will read all input line by line and will wait for new input forever. Thus, we will never reach the start of the loop again, being unable to receive a new connection in parallel with an existing one.
Adding Async
First of all, one important thing is that all Async operations should be called within an existing reactor, like this:
Sync do # This wraps everything with a Reactor
# some code
Async do # creates a Fiber and immediately moves on
# some async stuff
end
Async do # creates a Fiber and immediately moves on
# some more async stuff
end
# some more code
end # will wait for the finish of all async tasks inside
If you look at the source code of Sync, you'll see that it calls Fiber.set_scheduler(self) so that all Async tasks inside will use the same Async scheduler. Let's wrap our code:
def start
server = TCPServer.new(@port)
Sync do
loop do
client = server.accept # blocking
while client.gets do # blocking as well
process_input($_)
end
end
end
end
The only thing left to do is to allow our loop to move on after accepting a connection to asynchronously process messages from the received connection and wait for another one:
def start
server = TCPServer.new(@port)
Sync do
loop do
client = server.accept
Async do
process_input($_) while client.gets
end
end
end
end
What happens here? After accepting a connection, we create a new Fiber that will process messages from this connection in a non-blocking way, allowing our code to move on in the loop and go back to client = server.accept to wait for new connections. Thus, each connection will have its own Fiber. When receiving a message, the Fiber scheduler will pass control to the Fiber in which a message is received from the connection, allowing it to process the message.
When we call blocking operations like server.accept within an Async block, something interesting happens behind the scenes. The Fiber.scheduler intercepts these blocking I/O calls and transforms them into non-blocking equivalents. Instead of halting execution of the entire program while waiting for a connection, the Fiber.scheduler registers interest in the socket with the event loop and yields control back to the scheduler. When a connection is ready, the event loop notifies the scheduler, which then resumes the appropriate fiber. This means that while one fiber is waiting for a connection on server.accept, other fibers can continue executing, allowing our server to handle multiple connections concurrently without using threads.
One should be careful here: if you write, for example, like this:
loop do
Async do
client = server.accept
process_input(client, $LAST_READ_LINE) while client.gets
end
end
Without having a blocking operation in the loop, you end up with an infinite number of fibers each waiting for a connection, which is obviously not good.
Further improvements
You've probably already noticed the absence of resource cleanup here. This was done intentionally: I will write part 2 of this article to explain this topic further, because although we could simply add ensure blocks, that's not the best approach. We can use Async::IO::Endpoint.tcp instead, and I will explain the differences in the follow-up article.
Conclusion: The Benefit of Async
With just a few modifications to our code, we've transformed a basic single-connection server into one that can efficiently handle multiple connections simultaneously.
The beauty of this approach lies in its simplicity. Instead of managing complex thread pools or wrestling with IO#Select, Async leverages Ruby's Fibers to provide a clean, readable solution to concurrency challenges. Each connection gets its own Fiber, making the code both efficient and easier to reason about.
If you've been relying on threads for similar tasks, I'd strongly encourage giving Async a try. The library handles the complex event loop mechanics behind the scenes, allowing you to focus on your application logic rather than concurrency implementation details.
Next time you need to build something that handles multiple connections, consider reaching for Async - it might just make your concurrent programming life considerably easier.
Top comments (0)