lib/rainbows/queue_pool.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 require 'thread'
   4
   5 # Thread pool class based on pulling off a single Ruby Queue.
   6 # This is NOT used for the ThreadPool class, since that class does not
   7 # need a userspace Queue.
   8 class Rainbows::QueuePool < Struct.new(:queue, :threads)
   9   def initialize(size = 20, &block)
  10     q = Queue.new
  11     self.threads = (1..size).map do
  12       Thread.new do
  13         while job = q.shift
  14           block.call(job)
  15         end
  16       end
  17     end
  18     self.queue = q
  19   end
  20
  21   def quit!
  22     threads.each { |_| queue << nil }
  23     threads.delete_if do |t|
  24       Rainbows.tick
  25       t.alive? ? t.join(0.01) : true
  26     end until threads.empty?
  27   end
  28 end