lib/rainbows/thread_pool.rb

   1 # -*- encoding: binary -*-
   2
   3 module Rainbows
   4
   5   # Implements a worker thread pool model.  This is suited for platforms
   6   # where the cost of dynamically spawning a new thread for every new
   7   # client connection is too high.
   8   #
   9   # Applications using this model are required to be thread-safe.
  10   # Threads are never spawned dynamically under this model.  If you're
  11   # connecting to external services and need to perform DNS lookups,
  12   # consider using the "resolv-replace" library which replaces parts of
  13   # the core Socket package with concurrent DNS lookup capabilities.
  14   #
  15   # This model is less suited for many slow clients than the others and
  16   # thus a lower +worker_connections+ setting is recommended.
  17   module ThreadPool
  18
  19     include Base
  20
  21     def worker_loop(worker)
  22       init_worker_process(worker)
  23       RACK_DEFAULTS["rack.multithread"] = true
  24       pool = (1..worker_connections).map { new_worker_thread }
  25       m = 0
  26
  27       while LISTENERS.first && master_pid == Process.ppid
  28         pool.each do |thr|
  29           worker.tmp.chmod(m = 0 == m ? 1 : 0)
  30           # if any worker dies, something is serious wrong, bail
  31           thr.join(timeout) and break
  32         end
  33       end
  34       join_threads(pool, worker)
  35     end
  36
  37     def new_worker_thread
  38       Thread.new {
  39         begin
  40           begin
  41             ret = IO.select(LISTENERS, nil, nil, timeout) or next
  42             ret.first.each do |sock|
  43               begin
  44                 process_client(sock.accept_nonblock)
  45               rescue Errno::EAGAIN, Errno::ECONNABORTED
  46               end
  47             end
  48           rescue Errno::EINTR
  49             next
  50           rescue Errno::EBADF, TypeError
  51             return
  52           end
  53         rescue Object => e
  54           listen_loop_error(e) if LISTENERS.first
  55         end while ! Thread.current[:quit] && LISTENERS.first
  56       }
  57     end
  58
  59   end
  60 end