Correctness of a minimal parallel workers queue in Ruby

Question

I've programmed a minimal parallel workers queue in Ruby, and I wanted to know if it's correct, and if there are simpler ways of implementing it.

I'm aware of the existence of MonitorMixin, but it won't save any code, I think.

The design is kept intentionally trivial (e.g. it uses Thread.abort_on_exception).

# Usage:
#
#   queue = ParallelWorkersQueue.new( <threads> )
#   queue.push { <operation_1> }
#   queue.push { <operation_2> }
#   queue.join
#
class ParallelWorkersQueue

  def initialize( slots )
    @max_slots  = slots
    @free_slots = slots

    @mutex              = Mutex.new
    @condition_variable = ConditionVariable.new

    Thread.abort_on_exception = true
  end

  def push( &task )
    @mutex.synchronize do
      while @free_slots == 0
        @condition_variable.wait( @mutex )
      end

      @free_slots -= 1
    end

    Thread.new do
      yield

      @mutex.synchronize do
        @free_slots += 1

        @condition_variable.signal
      end
    end
  end

  def join
    @mutex.synchronize do
      while @free_slots < @max_slots
        @condition_variable.wait( @mutex )
      end
    end
  end

end

Answer 1

Like @Victor Moroz, I'm also a fan of Queue. I also prefer to keep the threads around: it's a bit more efficient, but more important, the code is simpler:

require 'thread'

class ParallelWorkersQueue

  def initialize(slots)
    @work = Queue.new
    @threads = slots.times.map do
      Thread.new do
        while (work = @work.deq) != :stop
          work[]
        end
      end
    end
  end

  def join
    @threads.each do
      @work.enq(:stop)
    end
    @threads.each(&:join)
  end

  def push(&task)
    @work.enq(task)
  end

end

Example of its use:

pwq = ParallelWorkersQueue.new(2)
5.times do |i|
  pwq.push do
    puts i
  end
end
pwq.join

# => 0
# => 1
# => 2
# => 3
# => 4

Replace Queue with SizedQueue if you wish to limit the amount of work that can be awaiting a worker thread.

Answer 2

I would prefer to use Queue in Erlang style (message passing), it looks cleaner. You can even pass something more useful than :done back, e.g. creating parallel map.

require "thread"

POOL_SIZE = 5

tasks = (0..99).map { |i| lambda { puts "Processing #{i}"; sleep 1 } }

message_queue = Queue.new

start_thread = 
  lambda do
    Thread.new(tasks.shift) do |task|
      task[]
      message_queue.push(:done)
    end
  end

tasks_left = tasks.length

[tasks_left, POOL_SIZE].min.times do
  start_thread[]
end

while tasks_left > 0
  message_queue.pop
  tasks_left -= 1
  start_thread[] unless tasks_left < POOL_SIZE
end