An implementation of parallel workers in Ruby Much in the spirit of this question, I have implemented a simple class for parallel workers and wanted to get some feedback on it. Can I make the code even more concise or readable? Are there any hidden problems that I should be aware of?

Worker class

class Worker

    def initialize queue
        @queue = queue

        # Set the "idle state" (returned by the idle? method)
        # and a mutex for accessing it
        @idle_state = false
        @idle_mutex = Mutex.new

        # Set the "exit state" (returned by the done? method)
        # and a mutex for accessing it
        @exit_state = false
        @exit_mutex = Mutex.new

        poll
    end

    # Poll a queue for Proc objects to process
    def poll
        @thread = Thread.new do
            loop do
                while @queue.empty?
                    set_idle true
                    exit 0 if done?
                    sleep 1
                end
                set_idle false
                job = @queue.pop
                job.call
            end
        end
    end

    def done?
        @exit_mutex.synchronize { @exit_state }
    end

    def shut_down
        @exit_mutex.synchronize { @exit_state = true }
        @thread.join
    end

    def idle?
        @idle_mutex.synchronize { @idle_state }
    end

    def set_idle state
        @idle_mutex.synchronize { @idle_state = state }
    end

end


Usage

# Set up a job queue
queue = Queue.new

# Spin up a few workers
workers = []
(1..5).each do
    workers &lt;&lt; Worker.new(queue)
end

# Add some jobs to the queue
(1..50).each do |i|
    queue &lt;&lt; Proc.new { $stdout.print "Job ##{i}\n" }
end

# Shut down each of the workers once the queue is empty
sleep 0.1 until queue.empty?
workers.each { |worker| worker.shut_down }


References:


Correctness of a minimal parallel workers queue in Ruby
Where not to lock
I'm not sure why you decided to lock on @exit_state and @idle_state, as they don't seem to have any potential to cause any race condition (except for maybe when the manager calls thread.shut_down, and even then - all that would happen is that you will 'lose' a round of one second - hardly catastrophic).

Where to lock
When you want to concurrently work on a queue - you should synchronize your work with the queue (in this case @queue.empty? and @queue.pop), which you don't do...

How to lock
You should also note the @queue.pop is implicitly @queue.pop(false) which means that if the queue is empty, it will block until a new message arrives, which is probably not what you want, since you can't be sure that it is not empty (between while @queue.empty? to job = @queue.pop another worker might have 'hijacked' your message).

Since Queue is thread-safe, you could write your worker like this:

class Worker

  def initialize(queue)
    @thread = Thread.new { poll(queue) }
  end

  def poll(queue)
    until done?
      begin
        queue.pop(true).call
      rescue ThreadError
        # queue was empty
        sleep 1
      end
    end
    exit 0
  end

  def done?
    @done
  end

  def shut_down
    @done = true
  end
end