lib/rainbows/thread_pool.rb

   1 # -*- encoding: binary -*-
   2
   3 # Implements a worker thread pool model.  This is suited for platforms
   4 # like Ruby 1.9, where the cost of dynamically spawning a new thread for
   5 # every new client connection is higher than with the ThreadSpawn model,
   6 # but the cost of an idle thread is low (e.g. NPTL under Linux).
   7 #
   8 # This model should provide a high level of compatibility with all Ruby
   9 # implementations, and most libraries and applications.  Applications
  10 # running under this model should be thread-safe but not necessarily
  11 # reentrant.
  12 #
  13 # Applications using this model are required to be thread-safe.  Threads
  14 # are never spawned dynamically under this model.
  15 #
  16 # If you're using green threads (MRI 1.8) and need to perform DNS lookups,
  17 # consider using the "resolv-replace" library which replaces parts of the
  18 # core Socket package with concurrent DNS lookup capabilities.
  19 module Rainbows::ThreadPool
  20   include Rainbows::Base
  21
  22   def worker_loop(worker) # :nodoc:
  23     init_worker_process(worker)
  24     pool = (1..worker_connections).map do
  25       Thread.new { LISTENERS.size == 1 ? sync_worker : async_worker }
  26     end
  27
  28     while Rainbows.alive
  29       # if any worker dies, something is serious wrong, bail
  30       pool.each do |thr|
  31         Rainbows.tick or break
  32         thr.join(1) and Rainbows.quit!
  33       end
  34     end
  35     Rainbows::JoinThreads.acceptors(pool)
  36   end
  37
  38   def sync_worker # :nodoc:
  39     s = LISTENERS[0]
  40     begin
  41       c = s.kgio_accept and c.process_loop
  42     rescue => e
  43       Rainbows::Error.listen_loop(e)
  44     end while Rainbows.alive
  45   end
  46
  47   def async_worker # :nodoc:
  48     begin
  49       # TODO: check if select() or accept() is a problem on large
  50       # SMP systems under Ruby 1.9.  Hundreds of native threads
  51       # all working off the same socket could be a thundering herd
  52       # problem.  On the other hand, a thundering herd may not
  53       # even incur as much overhead as an extra Mutex#synchronize
  54       ret = select(LISTENERS) and ret[0].each do |s|
  55         s = s.kgio_tryaccept and s.process_loop
  56       end
  57     rescue Errno::EINTR
  58     rescue => e
  59       Rainbows::Error.listen_loop(e)
  60     end while Rainbows.alive
  61   end
  62 end