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   G = Rainbows::G
  10
  11   def initialize(size = 20, &block)
  12     q = Queue.new
  13     self.threads = (1..size).map do
  14       Thread.new do
  15         while job = q.shift
  16           block.call(job)
  17         end
  18       end
  19     end
  20     self.queue = q
  21   end
  22
  23   def quit!
  24     threads.each { |_| queue << nil }
  25     threads.delete_if do |t|
  26       G.tick
  27       t.alive? ? t.join(0.01) : true
  28     end until threads.empty?
  29   end
  30 end