Adding support for queues to RubyJob
Tags: ruby, tdd, design, testing
Published: 2020-01-13 00:15:49

Over the holidays, with a bit of spare time between family gatherings, I decided to scratch an itch and create a new open source job processing framework for Ruby, called [RubyJob](https://github.com/mimperatore/ruby_job).

I'd used various other such frameworks in the past, such as Sidekiq, Resque, Delayed::Job and SuckerPunch, but none of them behaved and performed to my liking.

I'm not going to go over the pros & cons of all of these frameworks here, but feel free to check out the nice article the __Scout APM Team__ put together [here](https://scoutapm.com/blog/which-ruby-background-job-framework-is-right-for-you) if you're interested in that.

To my surprise, in the past two days, after I released my gem on __RubyGems.org__, it has been downloaded over _200 times_!  I have no idea who might be trying it out, but I'm glad they are.  The interest shown further increases my desire to write the next feature I had been planning: _Queue Support_.

Remaining in the spirit of openness, I've decided to capture the thought process and work I'm going to go through in order to take the current version of _RubyJob_, which currently has no support for multiple queues, and enhance it to allow the developer to specify both a queue to use, as well as its priority relative to other queues.

## TDD: Test Driven [Design|Development]

I generally like to work using my own flavour of _TDD_, one that isn't exactly like the often discussed [red/green/refactor](https://www.codecademy.com/articles/tdd-red-green-refactor) method, but rather a less constraining form that sees me sometimes writing tests first, and sometimes writing tests second (but in all cases before any change is deemed complete.)

I like to use [guard](https://github.com/guard/guard), and its [RSpec-plugin](https://github.com/guard/guard-rspec), to watch the filesystem for any changes to my source code and automatically re-run tests.  This not only makes it easy for me to remember to think about the tests, but also helps steer me in a direction that allows me to make incremental changes that _always_ keep the tests green while refactoring, and _almost always_ when in the process of adding a breaking change.

You may have noticed the title of this section having `Design|Development` as the last `D` in _TDD_.  That's because I firmly believe that tests are a great tool to use to actually flesh out emergent software design.  Often, and as is the case for the feature I'm going to be adding, we don't know exactly what we want the API to look like in advance, and writing tests actually helps us better visualize what the end result may end up looking like.  That's because test code tends to look similar, or have similar properties, to production code that uses the software we're developing.

Let's dig into thinking about the design of the queue/priority feature...

## Typical Queue Support In Other Gems
Most gems similar to __RubyJob__ have some way of partitioning jobs across one of more queues, and to assign to each queue a relative priority, such that jobs get processed in an order that can be controlled by the developer.  For example, you may put urgent jobs on a queue called `critical`, and regular jobs on the `default` queue.

In Sidekiq, this option can be specified with the `:queue` _option_:
```ruby
class MyCriticalWorker
  include Sidekiq::Worker

  sidekiq_options queue: 'critical'

  def perform
    #job code goes here
  end
end
```

Adding to that a bit of configuration in _sidekiq.yml_, you also establish the queue priorities.  In the case below, queue `critical` is 10x more important than `default`.

```ruby
:queues:
  - ["critical", 10]
  - ["default", 1]
```

## Specifying Queues In RubyJob
`RubyJob` is a bit different from the other job processing frameworks in that it has the notion of a _Job Store_, which is an _abstraction_ that enables us to choose how we want to keep track of our jobs.

For example, if our job worker looked like this:
```ruby
class MyWorker
  include RubyJob::Worker

  def perform
    #job code goes here
  end
end
```

then we tell the framework to track `MyWorker` jobs like this:
```ruby
MyWorker.jobstore = RubyJob::InMemoryJobStore.new

```

In this case, the jobs are being tracked (stored) in an `InMemoryJobStore`, a data structure backed by a [_Fibonacci Heap_](https://en.wikipedia.org/wiki/Fibonacci_heap) which is very efficient for this use case.

The `server` to then process the jobs is started like this:
```ruby
server = RubyJob::ThreadedServer.new(num_threads: 10, jobstore: MyWorker.jobstore)
server_thread = server.start
```

The _Job Store_ is explicitly passed to the constructor of `ThreadedServer`.  The _server_ spawns the specified number of threads, each of which repeatedly fetches the next ready job from the job store and processes it.  (For additional information on the full Job Store API, check out the docs [here](https://mimperatore.github.io/ruby_job/)).

So, how could we envision job _queues_ meshing into this design?

## Job Stores As Queues
In some ways, the job stores themselves are already queues; they already keep track of jobs to process, and the order in which to process them.

The issue with the example above is that all threads read the same, single, job store.  But what if we could specify a different number of threads for each different job stores, like this:

```ruby
server = RubyJob::ThreadedServer.new(
  config: [
    { num_threads: 10, jobstore: MyCriticalWorker.jobstore },
    { num_threads: 1, jobstore: MyRegularWorker.jobstore }
  ])
server.start
```

On the surface, this seems like a solution that could work, and it would be pretty trivial to implement.  The code in `ThreadedServer#start` which is currently:
```ruby
    def start
      Thread.new do
        @num_threads.times.map do
          Thread.new do
            JobProcessor.new(@jobstore).run(**@options)
          end
        end.each(&:join)
      end
    end
```

would need to be only slightly tweaked so we iterate over the specified `config` elements, and create the right sets of threads.

## Different Server Per Queue
But what about a different approach -- one that doesn't require us to change anything at all?

We know we can create a multi-threaded server object with a specified number of threads to process a single job store.  And the server itself is merely a top-level parent thread of the worker threads. What if we simply created multiple servers?

```ruby
critical_server = RubyJob::ThreadedServer.new(
  num_threads: 10, jobstore: MyCriticalWorker.jobstore
)
regular_server = RubyJob::ThreadedServer.new(
  num_threads: 1, jobstore: MyRegularWorker.jobstore
)
critical_server.start
regular_server.start
```

This option to me feels more robust and extensible than the first, because it doesn't couple the behaviour of the entire set of queues into a single `ThreadedServer` running instance.  Granted, you could pass different `config`'s using the first approach as well, but that seems counterintuitive and forced, whereas this approach fits in naturally with the current API design.

## So, RubyJob Already Supports Queues & Priorities?
If we want to go with the approach above, yes!  And we do want to go with the approach above, for the reasons already stated.

Pretty cool!  We started off thinking we didn't have the queueing and priority features, but thinking through it, we actually discovered that how the system had been designed already afforded us that capability!  Nice, pleasant surprise!  And this is not to say that things may not change in the future, but for now there is no real reason to make things more complicated.

## We're Not Done Though
Even though we do have the intrinsic ability to partition jobs into various queues and ensure that they run with a priority that we control, doing so requires a bit more boilerplate code than desirable.

Ideally, we wouldn't need to write all this code:
```ruby
MyCriticalWorker.jobstore = RubyJob::InMemoryJobStore.new
MyRegularWorker.jobstore = RubyJob::InMemoryJobStore.new

critical_server = RubyJob::ThreadedServer.new(
  num_threads: 10, jobstore: MyCriticalWorker.jobstore
)
regular_server = RubyJob::ThreadedServer.new(
  num_threads: 1, jobstore: MyRegularWorker.jobstore
)

critical_server.start
regular_server.start
```

### Keeping Queue and Worker Decoupled
One of the problems with Sidekiq's and Resque's way of specifying the queue is that it is coupled to the code that defines the worker.  The decision of which queue and priority a particular job should have should rest with whoever is orchestrating and managing the overall system, not the worker class itself.

Likewise, and for similar reasons, the number of threads should not be specified in the worker either, like Sucker Punch does.

For RubyJob, it would be therefore desirable to keep the configuration completely external (e.g. in a _YAML_ file), such that it can be read by a coordinator which configures and starts the necessary `ThreadedServer`s.  I'm envisioning a config file like this:

```ruby
MyCriticalWorker:
  num_threads: 10
  jobstore: RubyJob::InMemoryJobStore
MyRegularWorker:
  num_threads: 1
  jobstore: RubyJob::InMemoryJobStore
```

and a way of starting the servers like this:
```bash
$ ruby_job -w MyCriticalWorker start
$ ruby_job -w MyRegularWorker start
```

This, however, is beyond the scope of what I had intended to cover in today's post, and will be subject of a future day's work and probable post.  It's a different enough topic that deserves its own set of design considerations.  I'll be busting out [thor](https://github.com/erikhuda/thor) and some interesting _RSpec_ techniques for testing command-line interfaces!

I had hoped that I'd actually be writing some more code that I could share with you today, but hopefully sharing my thoughts about what was next on my list of things to do on `RubyJob` was at least a bit interesting or informative.  Feel free to share any comments or questions below... thanks for reading!