Asynchronously drain Resque queue with concurrent workers in Ruby

!resque_async_drainer.rb

require_relative './logging'
require_relative './async_consumer'

def drain(data_store, queue_name, output_filename)
  json_store = File.open(output_filename, 'a')

  executor = lambda do |class_name, args|
    Logging.logger.info "class_name: #{class_name}, args: #{args}"

    json_store.write("#{ { class: class_name, args: args }.to_json }\n")
  end

  consumer = AsyncConsumer.new(data_store, queue_name, executor)
  consumer.run
  consumer.print_performance
ensure
  json_store.close
end

Resque.redis = Redis.new
queue_name = 'foofoo'

drain(Resque.data_store, queue_name, "resque_queue_#{queue_name}.jsonl")

async_consumer.rb

require 'concurrent'
require 'benchmark'
require_relative './logging'
require_relative './resque_queue_drainer'
class AsyncConsumer
  include Logging

  NUM_WORKERS = 3
  NUM_JOBS = 50

  attr_accessor :stats

  def initialize(data_store, queue_name, executor, options: { n_workers: NUM_WORKERS, n_jobs: NUM_JOBS })
    @drainer = ResqueQueueDrainer.new(data_store, queue_name)
    @jobs = SizedQueue.new(options[:n_jobs])
    @executor = executor
    @stats = Concurrent::Hash.new
  end

  def run
    Concurrent::Future.execute do
      puts "Draining to queue"
      @drainer.drain_all do |class_name, args|
        @jobs.push([class_name, args])
      end
      @jobs.push(nil)
      @jobs.close
    end

    logger.info 'Starting workers'

    t = Benchmark.realtime do
      promises = Array.new(NUM_WORKERS) do |i|
        Concurrent::Promises.future { run_worker(i) }
      end
      promises.each(&:value!) # resolve promises
    end

    logger.info "Elapsed %.3f s" % t
  rescue => e
    logger.error "Terminated due to error: #{e.message}\n#{e.backtrace[0]}"
  ensure
    logger.info 'Finished workers'
  end

  def print_performance
    puts "Performance: #{stats.inspect}"
  end

  private

  def run_worker(worker)
    logger.info "Starting worker #{worker}"

    while !@jobs.empty? && (x = @jobs.pop)
      logger.info "Worker #{worker} picked job: #{x.inspect}"
      incr_counter(worker)

      class_name, args = x
      begin
        @executor.call(class_name, args)
      rescue => e
        logger.error "Bad luck! Job #{job_name} has failed due to error: #{e.message}\n#{e.backtrace[0]}"
      end

      logger.info "Worker #{worker} finished job: #{x.inspect}"
    end

    logger.info "Stopping worker #{worker}"
  end

  def incr_counter(worker)
    @stats[worker] ||= 0
    @stats[worker] += 1
  end
end

LICENSE.txt

Copyright 2022 Roman Miro

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

logging.rb

# Credits to Jacob: https://stackoverflow.com/a/6768164/10118318 . Licensed CC BY-SA 3.0
require 'logger'
module Logging
  # This is the magical bit that gets mixed into your classes
  def logger
    Logging.logger
  end

  # Global, memoized, lazy initialized instance of a logger
  def self.logger
    @logger ||= Logger.new($stdout)
  end
end

resque_queue_drainer.rb

require 'resque'
class ResqueQueueDrainer
  def initialize(data_store, queue_name)
    @data_store = data_store
    @queue_name = queue_name
  end

  def drain_all
    while (payload = pop_job_data)
      begin
        job_data = JSON.parse(payload)
        class_name = job_class_name(job_data)
        args = job_args(job_data)
        yield class_name, args
      rescue JSON::ParserError
        puts "Error parsing job data: #{payload[0, 1000]}"
      end
    end
  end

  private

  def pop_job_data
    @data_store.pop_from_queue(@queue_name)
  end

  def job_class_name(job_data)
    job_data['class']
  end

  def job_args(job_data)
    job_data.dig('args', 0)
  end
end