Exploring async Ruby

During this year’s RubyConf Thailand, I realized that I never seriously played around with the async gem. This library implements an event-driven reactor (via io-event) that provides Ruby developers with structured concurrency primitives built on fibers. I generally learn programming concepts and libraries best by using them, so I decided to write a little example program which we’ll explore in this blog post.

#Code walkthrough

We want our script to be self-contained, so we use Bundler’s inline mode, which will automatically install any missing gems from everyone’s favorite new gem server and require them:

require "bundler/inline"

gemfile do
  source "https://gem.coop"

  gem "async"
  gem "async-http"
end

We then define a simple Data class to hold the result data for our HTTP requests:

Result = Data.define(:url, :status, :duration, :error) do
  def ok? = error.nil?
end

Now for the interesting part: the actual URLChecker class, where all the async code lives. It has two constructor arguments (that get their default values from constants): concurrency (the maximum number of parallel requests) and timeout (the per-request deadline in seconds).

class URLChecker
  DEFAULT_CONCURRENCY = 20
  DEFAULT_TIMEOUT = 10

  def initialize(concurrency: DEFAULT_CONCURRENCY, timeout: DEFAULT_TIMEOUT)
    @concurrency = concurrency
    @timeout = timeout
  end

Where things get interesting is the check(urls) method:

  def check(urls)
    Sync do
      results = Array.new(urls.size)
      barrier = Async::Barrier.new
      semaphore = Async::Semaphore.new(@concurrency, parent: barrier)
      internet = Async::HTTP::Internet.new

      begin
        urls.each_with_index do |url, idx|
          semaphore.async do |task|
            task.with_timeout(@timeout) do
              results[idx] = fetch_one(internet, url)
            end
          end
        end

        barrier.wait

      ensure
        barrier.cancel
        internet.close
      end

      results
    end
  end

There are several points of note here:

• While working asynchronously, the method looks synchronous from the outside. So there’s no function coloring to worry about.
• It uses an Async::Barrier, a synchronization primitive that allows us to wait for a group of tasks to complete (barrier.wait) or cancel them all at once (barrier.cancel).
• We also use an Async::Semaphore to limit the number of simultaneous requests.
• Each URL gets processed in its own asynchronous task via semaphore.async. Since tasks write to indexed slots in results, no locking is needed, and the result array preserves the order of the input.
• Cleanup is performed in the ensure block, which cancels any remaining tasks and closes the HTTP connection pool.

The fetch_one method can be blissfully ignorant of being used in an asynchronous context. It simply makes a HEAD request to the provided URL with the injected async HTTP client.

  private

  def fetch_one(internet, url)
    t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    response = internet.head(url)
    duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0

    status = response.status
    response.finish

    Result.new(url:, status:, duration:, error: nil)
  rescue StandardError => e
    Result.new(url:, status:, duration:, error: "#{e.class}: #{e.message}")
  end
end

Lastly, we can add some code to exercise the URL checker:

URLS = %w[
  http://example.com
  http://example.org
  https://httpbin.dev/status/200
  https://httpbin.dev/status/301
  https://httpbin.dev/status/404
  https://httpbin.dev/status/500
  https://this-does-not-exist.invalid
  https://httpbin.dev/delay/12
].freeze

checker = URLChecker.new(concurrency: 5, timeout: 10)

t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
results = checker.check(URLS)
elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
ok, failed = results.partition(&:ok?)

formatted = results.map do |r|
  r.ok? ?
    format("  %-52s  %3d  (%.2fs)", r.url, r.status, r.duration) :
    format("  %-52s  ERR  %s", r.url, r.error)
end

puts <<~MSG
  Check your URLs, before you wreck your URLs

  #{formatted.join("\n")}

  #{URLS.size} URLs in #{elapsed.round(2)}s — #{ok.count} ok, #{failed.count} failed
MSG

This generates the following output:

Check your URLs, before you wreck your URLs
  http://example.com                                    200  (0.19s)
  http://example.org                                    200  (0.20s)
  https://httpbin.dev/status/200                        200  (0.92s)
  https://httpbin.dev/status/301                        301  (0.92s)
  https://httpbin.dev/status/404                        404  (0.91s)
  https://httpbin.dev/status/500                        500  (0.73s)
  https://this-does-not-exist.invalid                   ERR  Socket::ResolutionError: getaddrinfo: Name or service not known
  https://httpbin.dev/delay/12                          ERR  Async::TimeoutError: execution expired
8 URLs in 10.21s — 6 ok, 2 failed

#Summary

I really enjoyed learning about the async family of gems and its approach to structured concurrency. I can remember several past occasions where I rewrote Ruby scripts in either Go or TypeScript (with Deno, see my previous post). It’s good to know that in the future I may be able to get away with a single-file Ruby script with some inline Bundler.