thread-pooled-job-server.rb

require 'concurrent'

MAX_THREADS = 20

class Job
  def step
    sleep 3               # simulates performing the next task in the jobs workflow
  end
end

class App
  def run(runner)
    pool = Concurrent::ThreadPoolExecutor.new(pool_opts)
    until runner.exit?
      job = poll          # only a single thread ever polls for work
      pool.post do
        job.step          # jobs are performed in parallel
      end
    end
    pool.shutdown
    pool.wait_for_termination
  end

  private

  def poll
    Job.new               # simulate polling the broker for a job that's ready to progress
  end

  def pool_opts
    {
      min_threads: MAX_THREADS,
      max_threads: MAX_THREADS,
      max_queue: 1,
      fallback_policy: :caller_runs
    }
  end
end

module Runner
  class << self
    def run(app)
      @exit = false
      thread = Thread.new do
        app.run(self)
      end
      thread.join
    rescue Interrupt
      @exit = true
      thread.join
    end

    def exit?
      @exit
    end
  end
end

Signal.trap("TERM") { raise Interrupt }

Runner.run(App.new)