lib/rainbows/writer_thread_pool.rb

   1 # -*- encoding: binary -*-
   2
   3 # This concurrency model implements a single-threaded app dispatch
   4 # with a separate thread pool for writing responses.
   5 #
   6 # Unlike most \Rainbows! concurrency models, WriterThreadPool is
   7 # designed to run behind nginx just like Unicorn is.  This concurrency
   8 # model may be useful for existing Unicorn users looking for more
   9 # output concurrency than socket buffers can provide while still
  10 # maintaining a single-threaded application dispatch (though if the
  11 # response body is dynamically generated, it must be thread safe).
  12 #
  13 # For serving large or streaming responses, using more threads (via
  14 # the +worker_connections+ setting) and setting "proxy_buffering off"
  15 # in nginx is recommended.  If your application does not handle
  16 # uploads, then using any HTTP-aware proxy like haproxy is fine.
  17 # Using a non-HTTP-aware proxy will leave you vulnerable to
  18 # slow client denial-of-service attacks.
  19 module Rainbows::WriterThreadPool
  20   # :stopdoc:
  21   include Rainbows::Base
  22
  23   @@nr = 0
  24   @@q = nil
  25
  26   def async_write_body(qclient, body, range)
  27     if body.respond_to?(:close)
  28       Rainbows::SyncClose.new(body) do |body|
  29         qclient.q << [ qclient.to_io, :body, body, range ]
  30       end
  31     else
  32       qclient.q << [ qclient.to_io, :body, body, range ]
  33     end
  34   end
  35
  36   def process_client(client) # :nodoc:
  37     @@nr += 1
  38     super(Client.new(client, @@q[@@nr %= @@q.size]))
  39   end
  40
  41   def init_worker_process(worker)
  42     super
  43     self.class.__send__(:alias_method, :sync_write_body, :write_body)
  44     Rainbows::WriterThreadPool.__send__(
  45                         :alias_method, :write_body, :async_write_body)
  46   end
  47
  48   def worker_loop(worker) # :nodoc:
  49     # we have multiple, single-thread queues since we don't want to
  50     # interleave writes from the same client
  51     qp = (1..worker_connections).map do |n|
  52       Rainbows::QueuePool.new(1) do |response|
  53         begin
  54           io, arg1, arg2, arg3 = response
  55           case arg1
  56           when :body then sync_write_body(io, arg2, arg3)
  57           when :close then io.close unless io.closed?
  58           else
  59             io.write(arg1)
  60           end
  61         rescue => err
  62           Rainbows::Error.write(io, err)
  63         end
  64       end
  65     end
  66
  67     @@q = qp.map { |q| q.queue }
  68     super(worker) # accept loop from Unicorn
  69     qp.map { |q| q.quit! }
  70   end
  71   # :startdoc:
  72 end
  73 # :enddoc:
  74 require 'rainbows/writer_thread_pool/client'