Thread and Queue

Question

I am interested in knowing what would be the best way to implement a thread based queue.

For example:

I have 10 actions which I want to execute with only 4 threads. I would like to create a queue with all the 10 actions placed linearly and start the first 4 action with 4 threads, once one of the thread is done executing, the next one will start etc - So at a time, the number of thread is either 4 or less than 4.

Answer 1

There is a Queue class in thread in the standard library. Using that you can do something like this:

require 'thread'  queue = Queue.new threads = []  # add work to the queue queue << work_unit  4.times do   threads << Thread.new do     # loop until there are no more things to do     until queue.empty?       # pop with the non-blocking flag set, this raises       # an exception if the queue is empty, in which case       # work_unit will be set to nil       work_unit = queue.pop(true) rescue nil       if work_unit         # do work       end     end     # when there is no more work, the thread will stop   end end  # wait until all threads have completed processing threads.each { |t| t.join } 

The reason I pop with the non-blocking flag is that between the until queue.empty? and the pop another thread may have pop'ed the queue, so unless the non-blocking flag is set we could get stuck at that line forever.

If you're using MRI, the default Ruby interpreter, bear in mind that threads will not be absolutely concurrent. If your work is CPU bound you may just as well run single threaded. If you have some operation that blocks on IO you may get some parallelism, but YMMV. Alternatively, you can use an interpreter that allows full concurrency, such as jRuby or Rubinius.

Answer 2

There area a few gems that implement this pattern for you; parallel, peach,and mine is called threach (or jruby_threach under jruby). It's a drop-in replacement for #each but allows you to specify how many threads to run with, using a SizedQueue underneath to keep things from spiraling out of control.

So...

(1..10).threach(4) {|i| do_my_work(i) } 

Not pushing my own stuff; there are plenty of good implementations out there to make things easier.

If you're using JRuby, jruby_threach is a much better implementation -- Java just offers a much richer set of threading primatives and data structures to use.