Making a task multi-threaded using Queue in Ruby

Often times we will have a task that requires running a simple operation or method over all of the items in a list. Usually, we will use #each or #map to perform these operations one-at-a-time, but there are times when having multiple operations running at once is more efficient. This is especially true with any operation that requires a network call. Even adding just one extra thread will speed up the task dramatically. However, this idea will work on any portion of code that calls #each or #map to call a method or block on multiple objects.

Using an Enumerable or Array in a multi-threaded environment can be done easily with Queue, which is thread-safe and similar to Array in its function. Unlike Array, Queue lacks many helper methods and feels surprisingly "low-level" for a native Ruby class. For instance, the only methods to access the data are #push, #pop, and #shift (#pop and #shift have the cumbersome side affect of removing the object as you access it). Despite its limitations, Queue is perfect for a simple stack or FIFO. Although there is no method to convert between an Array and a Queue, there is a simple trick to populate one using any Enumerable object:

(1..100_000).inject(Queue.new, :push)

Conveniently, Queue#push returns the Queue instance after performing its operation. This allows us to chain together #push calls with each object passed as parameter to a call. This is what .inject(Queue.new, :push) does in the code above:

# Essentially the same
Queue.new.push(1).push(2).push(3) #... .push(100_000)

Now that our Queue is populated, we need to start a fixed number of threads to process it:

NUM_THREADS = 4
Thread.abort_on_exception = true

def do_stuff(object)
  # ...
end

@queue = (1..100_000).inject(Queue.new, :push)

@threads = Array.new(NUM_THREADS) do
  Thread.new do
    until @queue.empty?
      # This will remove the first object from @queue
      next_object = @queue.shift

      do_stuff(next_object)
    end
  end
end

@threads.each(&:join)

Essentially, the above code creates 4 threads which will continuously pull an object out of @queue and call #do_stuff on it until @queue is empty. Note that Thread.abort_on_exception = true will allow any errors thrown within a thread to be caught during the #join call:

begin
  @threads.each(&:join)
ensure
  cleanup()
end