DEV Community: Alexey

Building a Multi-Connection Redis Server with Ruby's Async Library [Part 1]

Alexey — Wed, 26 Feb 2025 12:39:26 +0000

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.

Persistent Redis Connections in Sidekiq with Async::Redis: A Deep Dive.

Alexey — Thu, 18 Jul 2024 11:27:02 +0000

TL;DR: In this post I explain how you can have persistent connections to Redis using Async::Redis client inside Sidekiq and why it doesn't work by default. There are some basic explanations of how Async and Sidekiq work and some source code.

All example code can be found here. Note that we don't actually need Rails, but I used it for examples because Sidekiq is usually used with Rails.

Async is a composable asynchronous I/O framework for Ruby. It allows you to do things concurrently using Fibers. Since 3.0, Ruby has a fiber scheduler and Ruby core supports it. This means you can have non-blocking I/O without much effort, for example, when using Net::HTTP. If you perform a blocking operation, such as an HTTP call, inside a fiber, it will immediately yield so that another fiber can become active and do some useful work instead of blocking and waiting for the HTTP call to complete.

Using Ruby Async::Redis gem

Besides Ruby's support for the fiber scheduler, for some I/O operations, you might use specific gems, like Async::Redis. You can still use other Ruby gems that use native Ruby I/O with Async and they will give you non-blocking I/O as well, but there are two reasons to prefer Async::Redis:

Its author, Samuel Williams, worked hard to ensure the interface is more consistent and easier to use. This is especially true for support of pipelines, transactions and subscriptions.
It uses Async::Pool so you don't need to make connection pool by yourself. Async::Pool has persistent connections out of box, and also it's great that connections are created in a lazy way: a new one will be created only if there are no free connections (and if specified connections limit wasn't reached). This can be extremely handy if you use Async inside Sidekiq and can't calculate how many connections you will need.

The usage is simple:

First, add gem to Gemfile:

gem 'async-redis'

Then create a client:

endpoint = Async::Redis.local_endpoint # localhost:6379
@client = Async::Redis::Client.new(endpoint, limit: 10)

Limit parameter will be passed to Async::Pool to limit maximum number of acquired connections, you also can check a short doc about another parameter you can pass, concurrency.

# source code of Async::Pool

module Async
  module Pool
    class Controller
      def self.wrap(**options, &block)
        self.new(block, **options)
      end

      def initialize(constructor, limit: nil, concurrency: (limit || 1), policy: nil)

Async::Redis.local_endpoint is just a IO::Endpoint.tcp with default local Redis host and port, we can find it in the gem's source code:

def self.local_endpoint(port: 6379)
  ::IO::Endpoint.tcp('localhost', port)
end

Now we can call the client:

Sync do
  @client.set('key', 'value')
  puts @client.get('key') # => "value"
ensure
  @client.close
end

The resulting code:

# app/lib/redis_example.rb

require 'async/redis'

endpoint = Async::Redis.local_endpoint
@client = Async::Redis::Client.new(endpoint)

Sync do
  @client.set('key', 'value')
  puts @client.get('key') # => "value"
ensure
  @client.close
end

Why do we need a Sync do block? We'll dive into this a little further.

Using Async::Redis gem with Ruby on Rails

This setup is pretty simple, however, most of the time we don't write pure Ruby programs, usually we have Rails. For Rails, it would be handy to have a persistent client that we can call anywhere. Let's create one:

# app/lib/async_redis_client.rb

class AsyncRedisClient
  def self.instance
    Thread.current.thread_variable_get(:redis_client) || 
      Thread.current.thread_variable_set(:redis_client, new)
  end

  def initialize
    @client = Async::Redis::Client.new(Async::Redis.local_endpoint, limit: 10)
  end

  def with
    yield @client
  end
end

We use Thread.current.thread_variable_get and Thread.current.thread_variable_set to make our client thread-safe. If you won't do it, you will get fiber called across threads error pretty soon.

Now we have a persistent client that should have persistent connections. We can use it in Sidekiq, right?

# app/workers/redis_worker.rb

class RedisWorker
  include Sidekiq::Job

  def perform
    AsyncRedisClient.instance.with do |redis|
      redis.set('key', 'value')
    end
  end
end

No. Remember using Sync do in the beginning? Without it, we will receive this error:

No async task available! (RuntimeError)

A quick search and we see that Async Redis client needs to be called within an existing reactor. You can achieve it by running code inside Async do ... or Sync do ... block. Let's have a look at Sync source code:

module Kernel
  # Run the given block of code synchronously, but within a reactor if not already in one.
  #
  # @yields {|task| ...} The block that will execute asynchronously.
  #     @parameter task [Async::Task] The task that is executing the given block.
  #
  # @public Since `stable-v1`.
  # @asynchronous Will block until given block completes executing.
  def Sync(&block)
    if task = ::Async::Task.current?
      yield task
    else
      # This calls Fiber.set_scheduler(self):
      reactor = Async::Reactor.new

      begin
        return reactor.run(finished: ::Async::Condition.new, &block).wait
      ensure
        Fiber.set_scheduler(nil)
      end
    end
  end
end

Pretty straightforward, creates reactor if not inside existing one. Typical pure Ruby program with Async looks like this:

  Sync do
    # blabla
    Async do
      # some async stuff
    end
    Async do
      # some more async stuff
    end
    # some more blabla
  end # will wait for the finish of all async tasks inside

But we are using Sidekiq, so let's just do what it asks, wrap in Sync block. You could also wrap it in Async, in this example it doesn't matter a lot because both will create a Reactor and wait for task to be completed before closing Reactor. But If you already have a Reactor (like we will have at the end of the article), using Async will lead for a Sidekiq job to be completed before our async task completed. But you also can do Async do ... end.wait.

# app/workers/redis_worker_with_sync.rb

class RedisWorkerWithSync
  include Sidekiq::Job

  def perform
    Sync do
      AsyncRedisClient.instance.with do |redis|
        redis.set('key', 'value')
      end
    end
  end
end

That works! You can also refactor it; for example move Sync block to Sidekiq middleware or to with block of our client.

The only thing left is to check that it works as we want: with persistent connections to Redis.

Monitoring Redis connections with Ruby and Async

We could just use redis-cli for it but we also can write a simple monitoring using Async. Redis has an info command that shows total_connections_received. We can use it to monitor newly received connections after the monitor has started:

# app/lib/redis_monitor.rb

require 'async'
require 'async/redis'
require 'tty-sparkline'
require 'tty-screen'

endpoint = Async::Redis.local_endpoint
redis = Async::Redis::Client.new(endpoint)

connection_counts = []

interval = 1
max_width = TTY::Screen.width - 20

Sync do
  initial_connections_count = redis.info[:total_connections_received].to_i

  Async do |task|
    loop do
      new_connections = redis.info[:total_connections_received].to_i - initial_connections_count

      connection_counts << new_connections

      if connection_counts.size > max_width
        connection_counts.shift
      end

      task.sleep(interval)
    end
  end

  Async do |task|
    loop do
      print "\e[H\e[2J" # Clear the screen

      puts "Redis Connection Monitoring"
      puts "Total created connections: #{connection_counts.last}"

      sparkline = TTY::Sparkline.new(connection_counts, width: max_width, height: 10)
      puts sparkline.render

      task.sleep(interval)
    end
  end
end

Here we have two tasks. The first one collects total_connections_received and writes it to connection_counts that will be used for our chart. The second task uses the tty-sparkline gem to output the chart to console.

Let's also add a Rake task to fill our Sidekiq queue:

# /lib/tasks/job_pusher.rb

namespace :jobs do
  desc "Push jobs to the queue continuously"
  task push: :environment do
    loop do
      puts 'Pushing job to the queue'
      RedisWorkerWithSync.perform_async
      sleep 1
    end
  end
end

Now let's run all we need for our experiment:

bundle exec sidekiq
bundle exec rake jobs:push
ruby app/lib/redis_monitor.rb

And lets have a look at monitoring result:

The connection count starts at 6 (Sidekiq creates and manages its own connection pool to Redis, atually 7th connection is created by Sidekiq as well when it starts processing jobs) and increases constantly, adding one new connection every second (because we push a new job every second). Why does it happen? I mentioned in the beginning that Async::Redis has persistent connections, hasn't it?

Well, it has. But only within a reactor. Because existing Reactor is definition of a "lifetime" of a Ruby program with Async. After our top Sync or Async block finishes Redis should close all connections, why does this need them anymore?

How does Sidekiq process run

Okay, back to our Rails app. In our app, the lifetime should be the whole Sidekiq process. Luckily, Sidekiq has internal documentation on how it runs. I won't copy the entire documentation here, just the part we are interested in:

Sidekiq::Manager manages the resources for a given Sidekiq::Capsule. It creates and starts N Sidekiq::Processor instances. ... Each Sidekiq::Processor instance is a separate thread.

That's exactly what we need, because we need to have a separate reactor for each thread.

Sidekiq 7.3 source code uses Processor method run looks like this:

# frozen_string_literal: true

module Sidekiq
  class Processor
     def run
      # By setting this thread-local, Sidekiq.redis will access +Sidekiq::Capsule#redis_pool+
      # instead of the global pool in +Sidekiq::Config#redis_pool+.
      Thread.current[:sidekiq_capsule] = @capsule

      process_one until @done
      @callback.call(self)
    rescue Sidekiq::Shutdown
      @callback.call(self)
    rescue Exception => ex
      @callback.call(self, ex)
    end
  end
end

It uses a loop for processing jobs process_one until @done and it's easy to wrap:

Sync do
  process_one until @done
end

We can put a file with our monkey-patch in app/lib/sidekiq/processor.rb:

# app/lib/sidekiq/processor.rb

# frozen_string_literal: true

module Sidekiq
  class Processor
    def run
      # By setting this thread-local, Sidekiq.redis will access +Sidekiq::Capsule#redis_pool+
      # instead of the global pool in +Sidekiq::Config#redis_pool+.
      Thread.current[:sidekiq_capsule] = @capsule

      Sync do
        process_one until @done
      end
      @callback.call(self)
    rescue Sidekiq::Shutdown
      @callback.call(self)
    rescue Exception => ex
      @callback.call(self, ex)
    end
  end
end

And we need to add it to initializer to load our code instead of original:

# config/initializers/sidekiq.rb

require_relative '../../app/lib/sidekiq/processor'

That's it! It will run in Sync block until we shut down Sidekiq. Now we can remove Sync block from our worker and everything works perfectly with persistent connections.

class RedisWorker
  include Sidekiq::Job

  def perform
    AsyncRedisClient.instance.with do |redis|
      redis.set('key', 'value')
    end
  end
end

Now when our Sidekiq is wrapped, we can change job pusher to use RedisWorker instead of RedisWorkerWithSinc and run it. Let's have a look at the monitor:

We see that connections size raises slowly until it reaches 12. That's because we have Sidekiq with default concurrency of 5, this means that we have 5 separate threads and when our job gets into thread that wasn't used before it has to create it's first connection. Let's change our rake task to push more jobs:

namespace :jobs do
  desc "Push jobs to the queue continuously"
  task push: :environment do
    loop do
      puts 'Pushing job to the queue'
      20.times do
        RedisWorker.perform_async
      end
      sleep 1
    end
  end
end

It reaches 12 instantly and stops at this number as we wanted.

Now, let's use Async inside Sidekiq. Currently, each woker performs only one Redis operation, that's why we use one same connection from the pool each time. Imagine that each worker needs to do 100 different Redis operations, with Async we could do them concurrently and more effective:

class RedisWorker
  include Sidekiq::Job

  def perform
    100.times do
      Async do
        rand = SecureRandom.hex
        AsyncRedisClient.instance.with do |redis|
          redis.set(rand, rand)
        end
      end
    end
  end
end

The mathematics is simple: Sidekiq utilizes 7 connection by itself and we have 5 Sidekiq threads, each thread has a connection pool to Redis with limit equal 10. Eventually we reach 57 connections.

Note that code above will finish Sidekiq job as soon as all tasks will be scheduled, not completed. We can change code to wait for completion, but we'll need much more tasks to utilize all connections in this way:

class RedisWorker
  include Sidekiq::Job

  def perform
    tasks = 10000.times.map do
      Async do
        rand = SecureRandom.hex
        AsyncRedisClient.instance.with do |redis|
          redis.set(rand.to_s, rand.to_s)
        end
      end
    end
    tasks.map(&:wait)
  end
end

And this is what would happen if we rolled back our monkey-patch:

# config/initializers/sidekiq.rb
# require_relative '../../app/lib/sidekiq/processor'

class RedisWorker
  include Sidekiq::Job

  def perform
    100.times do
      Async do
        rand = SecureRandom.hex
        AsyncRedisClient.instance.with do |redis|
          redis.set(rand, rand)
        end
      end
    end
  end
end

Finally, we can wrap whole worker in Sync as before to see what will happen:

class RedisWorker
  include Sidekiq::Job

  def perform
    Sync do
      100.times do
        Async do
          rand = SecureRandom.hex
          AsyncRedisClient.instance.with do |redis|
            redis.set(rand, rand)
          end
        end
      end
    end
  end
end

A little better because previously each Async created it's Reactor with new connection pool and now we have Sync that's owns a Reactor with pool for all Async tasks inside. But still, much worse than with patched Sidekiq.

This example is built around Redis, but same would be true for other Async I/O gems that need a Reactor, for example Async::HTTP.

Using Async could be tricky, but I encourage you to try it. If you are not familiar to it (and for some reason read this article till the end) I can recommend you to watch this video from the latest RailsConf 2024.