Learning by building, a Background Processing System in Ruby
Tags: ruby, learning
Published: 2019-04-10 06:52:14

---
title: Learning by building, a Background Processing System in Ruby
published: true
description: In this edition of Ruby Magic, we'll implement our very own background processing system to offload potentially slow tasks as a peek into background job
cover_image: https://blog.appsignal.com/images/blog/2019-04/background-processing.jpg
canonical_url: https://blog.appsignal.com/2019/04/02/background-processing-system-in-ruby.html
tags: ruby, learning
---

In today's post, we are going to implement a naive background processing system for fun! We might learn some things along the way as a peek into the internals of popular background processing systems like [Sidekiq](http://sidekiq.org). The product of this fun is by no means intended for production use. 
 
Let’s imagine we have a task in our application that loads one or more websites and extracts their titles. As we don’t have any influence on the performance of these websites, we’d like to perform the task outside our main thread (or the current request—if we’re building a web application), but in the background. 

## Encapsulating a Task 

Before we get into background processing, let’s build a service object to perform the task at hand. We’ll use [OpenURI](https://ruby-doc.org/stdlib-2.6.2/libdoc/open-uri/rdoc/OpenURI.html) and [Nokogiri](https://nokogiri.org) to extract the contents of the title tag.

```ruby
require 'open-uri' 
require 'nokogiri' 

class TitleExtractorService
  def call(url)
    document = Nokogiri::HTML(open(url))
    title = document.css('html > head > title').first.content
    puts title.gsub(/[[:space:]]+/, ' ').strip
  rescue
    puts "Unable to find a title for #{url}" 
  end
end
```

Calling the service prints the title of the given URL. 

```ruby
TitleExtractorService.new.call('https://appsignal.com')
# AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir
```

This works as expected, but let’s see if we can improve the syntax a little to make it look and feel a bit more like other background processing systems. By creating a `Magique::Worker` module, we can add some syntactic sugar to the service object. 

```ruby
module Magique
  module Worker
    def self.included(base)
      base.extend(ClassMethods)
    end

    module ClassMethods
      def perform_now(*args)
        new.perform(*args)
      end
    end

    def perform(*)
      raise NotImplementedError
    end
  end
end
```

The module adds a `perform` method to the worker instance and a `perform_now` method to the worker class to make the invocation a bit better. 

Let’s include the module into our service object. While we’re at it, let’s also rename it to `TitleExtractorWorker` and change the `call` method to `perform`. 

```ruby
class TitleExtractorWorker
  include Magique::Worker

  def perform(url)
    document = Nokogiri::HTML(open(url))
    title = document.css('html > head > title').first.content
    puts title.gsub(/[[:space:]]+/, ' ').strip
  rescue
    puts "Unable to find a title for #{url}"
  end
end
```

The invocation still has the same result, but it’s a bit clearer what's going on.

```ruby
TitleExtractorWorker.perform_now('https://appsignal.com')
# AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir
```

## Implementing Asynchronous Processing

Now that we have the title extraction working, we can grab all titles from past Ruby Magic articles. To do this, let’s assume we have a `RUBYMAGIC` constant with a list of all the URLs of past articles.

```ruby
RUBYMAGIC.each do |url|
  TitleExtractorWorker.perform_now(url)
end

# Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog
# Bindings and Lexical Scope in Ruby | AppSignal Blog
# Building a Ruby C Extension From Scratch | AppSignal Blog
# Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog
# ...
```

We get the titles of past articles, but it takes a while to extract them all. That’s because we wait until each request is completed before moving on to the next one. 

Let’s improve that by introducing a `perform_async` method to our worker module. To speed things up, it creates a new thread for each URL.

```ruby
module Magique
  module Worker
    module ClassMethods
      def perform_async(*args)
        Thread.new { new.perform(*args) }
      end
    end
  end
end
```

After changing the invocation to `TitleExtractorWorker.perform_async(url)`, we get all the titles almost at once. However, this also means that we’re opening more than 20 connections to the Ruby Magic blog at once. (*Sorry for messing with your blog, folks!* 😅)

*If you’re following along with your own implementation and testing this outside of a long-running process (like a web server), don’t forget to add something like `loop { sleep 1 }` to the end of your script to make sure the process doesn’t immediately terminate.*

## Queueing up Tasks

With the approach of creating a new thread for every invocation, we’ll eventually hit resource limits (both on our side and on the websites we are accessing). As we’d like to be nice citizens, let’s change the implementation to something that is asynchronous but doesn’t feel like a denial-of-service attack. 

A common way to solve this problem is to use the producer/consumer pattern. One or more producers push tasks onto a queue while one or more consumers take tasks from the queue and process them. 

A queue is basically a list of elements. In theory, a simple array would do the job. However, as we’re dealing with concurrency, we need to make sure that only one producer or consumer can access the queue at a time. If we aren’t careful about this, things will end in chaos—just like two people trying to squeeze through a door at once.

This problem is known as the [producer-consumer problem](https://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem) and there are multiple solutions to it. Luckily, it is a very common problem and Ruby ships with a proper `Queue` implementation that we can use without having to worry about thread synchronization.

To use it, let’s make sure both producers and consumers can access the queue. We do this by adding a class method to our `Magique` module and assigning an instance of `Queue` to it.

```ruby
module Magique
  def self.backend
    @backend
  end

  def self.backend=(backend)
    @backend = backend
  end
end

Magique.backend = Queue.new
```

Next, we change our `perform_async` implementation to push a task onto the queue instead of creating its own new thread. A task is represented as a hash including a reference to the worker class as well as the arguments passed to the `perform_async` method.

```ruby
module Magique
  module Worker
    module ClassMethods
      def perform_async(*args)
        Magique.backend.push(worker: self, args: args)
      end
    end
  end
end
```

With that, we’re done with the producer side of things. Next, let’s take a look at the consumer side.

Each consumer is a separate thread that takes tasks from the queue and performs them. Instead of stopping after one task, like the thread, the consumer then takes another task from the queue and performs it, and so on. Here’s a basic implementation of a consumer called `Magique::Processor`. Each processor creates a new thread that loops infinitely. For every iteration, it tries to grab a new task from the queue, creates a new instance of the worker class, and calls its `perform` method with the given arguments.  

```ruby
module Magique
  class Processor
    def self.start(concurrency = 1)
      concurrency.times { |n| new("Processor #{n}") }
    end

    def initialize(name)
      thread = Thread.new do
        loop do
          payload = Magique.backend.pop
          worker_class = payload[:worker]
          worker_class.new.perform(*payload[:args])
        end
      end

      thread.name = name
    end
  end
end
```

In addition to the processing loop, we add a convenience method called `Magique::Processor.start`. This allows us to spin up multiple processors at once. While naming the thread isn’t really necessary, it will allow us to see if things are actually working as expected.

Let’s adjust the output of our `TitleExtractorWorker` to include the name of the current thread. 

```ruby
puts "[#{Thread.current.name}] #{title.gsub(/[[:space:]]+/, ' ').strip}"
```

To test our background processing setup, we first need to spin up a set of processors before enqueueing our tasks. 

```ruby
Magique.backend = Queue.new
Magique::Processor.start(5)

RUBYMAGIC.each do |url|
  TitleExtractorWorker.perform_async(url)
end

# [Processor 3] Bindings and Lexical Scope in Ruby | AppSignal Blog
# [Processor 4] Building a Ruby C Extension From Scratch | AppSignal Blog
# [Processor 1] Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog
# [Processor 0] Ruby's Hidden Gems, StringScanner | AppSignal Blog
# [Processor 2] Fibers and Enumerators in Ruby: Turning Blocks Inside Out | AppSignal Blog
# [Processor 4] Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog
# ...
```

When this is run, we still get the titles of all articles. While it’s not as fast as just using a separate thread for every task, it’s still faster than the initial implementation that had no background processing. Thanks to the added processor names, we can also confirm that all processors are working through the queue. By tweaking the number of concurrent processors, it’s possible to find a balance between processing speed and existing resource limitations. 

## Expanding to Multiple Processes and Machines

So far, the current implementation of our background processing system works well enough. It’s still limited to the same process, though. Resource-hungry tasks will still affect the performance of the entire process. As a final step, let’s look at distributing the workload across multiple processes and maybe even multiple machines.

The queue is the only connection between producers and consumers. Right now, it’s using an in-memory implementation. Let’s take more inspiration from Sidekiq and implement a queue using [Redis](http://redis.org).

Redis has support for lists that allow us to push and fetch tasks from. Additionally, the Redis Ruby gem is thread-safe and the Redis commands to modify lists are [atomic](https://en.wikipedia.org/wiki/Linearizability). These properties make it possible to use it for our asynchronous background processing system without running into synchronization problems. 

Let’s create a Redis backed queue that implements the `push` and `shift` methods just like the `Queue` we used previously.

```ruby
require 'json'
require 'redis'

module Magique
  module Backend
    class Redis
      def initialize(connection = ::Redis.new)
        @connection = connection
      end

      def push(job)
        @connection.lpush('magique:queue', JSON.dump(job))
      end

      def shift
        _queue, job = @connection.brpop('magique:queue')
        payload = JSON.parse(job, symbolize_names: true)
        payload[:worker] = Object.const_get(payload[:worker])
        payload
      end
    end
  end
end
```

As Redis doesn’t know anything about Ruby objects, we have to serialize our tasks into JSON before storing them in the database using the `lpush` command that adds an element to the front of the list.

To fetch a task from the queue, we’re using the `brpop` command, which gets the last element from a list. If the list is empty, it’ll block until a new element is available. This is a nice way to pause our processors when no tasks are available. Finally, after getting a task out of Redis, we have to look up the real Ruby class based on the name of the worker using `Object.const_get`. 

As a final step, let’s split things up into multiple processes. On the producer side of things, the only thing we have to do is change the backend to our newly implemented Redis queue. 

```ruby
# ...

Magique.backend = Magique::Backend::Redis.new

RUBYMAGIC.each do |url|
  TitleExtractorWorker.perform_async(url)
end
```

On the consumer side of things, we can get away with a few lines like this:

```ruby
# ...

Magique.backend = Magique::Backend::Redis.new
Magique::Processor.start(5)

loop { sleep 1 }
```

When executed, the consumer process will wait for new work to arrive in the queue. Once we start the producer process that pushes tasks into the queue, we can see that they get processed immediately.

## Enjoy Responsibly and Don’t Use This in Production

While we kept it far from a real world setup you would use in production (so don't!), we took a few steps in building a background processor. We started by making a process run as a background service. Then we made it async and used `Queue` to solve the producer-consumer problem. Then we expanded the process to multiple processes or machines using Redis rather then an in-memory implementation. 

As mentioned before, this is a simplified implementation of a background processing system. There are a lot of things missing and not explicitly dealt with. These include (but are not limited to) error handling, multiple queues, scheduling, connection pooling, and signal handling.

Nonetheless, we had fun writing this and hope you enjoyed a peek under the hood of a background processing system. Perhaps you even took away a thing or two. 

_Guest writer [Benedikt Deicke](http://benediktdeicke.com) is a software engineer and CTO of [Userlist.io](http://userlist.io). On the side, he’s writing a book about building [SaaS applications in Ruby on Rails](http://saasguidebook.com). You can reach out to Benedikt via [Twitter](http://twitter.com/benediktdeicke)._