lib/rainbows/base.rb

   1 # -*- encoding: binary -*-
   2 require 'rainbows/tee_input'
   3
   4 # base class for Rainbows concurrency models, this is currently used by
   5 # ThreadSpawn and ThreadPool models.  Base is also its own
   6 # (non-)concurrency model which is basically Unicorn-with-keepalive, and
   7 # not intended for production use, as keepalive with a pure prefork
   8 # concurrency model is extremely expensive.
   9 module Rainbows::Base
  10
  11   # :stopdoc:
  12   include Rainbows::Const
  13
  14   # shortcuts...
  15   G = Rainbows::G
  16   NULL_IO = Unicorn::HttpRequest::NULL_IO
  17   TeeInput = Rainbows::TeeInput
  18   HttpResponse = Rainbows::HttpResponse
  19   HttpParser = Unicorn::HttpParser
  20
  21   # this method is called by all current concurrency models
  22   def init_worker_process(worker)
  23     super(worker)
  24     Rainbows::MaxBody.setup
  25     G.tmp = worker.tmp
  26
  27     # avoid spurious wakeups and blocking-accept() with 1.8 green threads
  28     if ! defined?(RUBY_ENGINE) && RUBY_VERSION.to_f < 1.9
  29       require "io/nonblock"
  30       Rainbows::HttpServer::LISTENERS.each { |l| l.nonblock = true }
  31     end
  32
  33     # we're don't use the self-pipe mechanism in the Rainbows! worker
  34     # since we don't defer reopening logs
  35     Rainbows::HttpServer::SELF_PIPE.each { |x| x.close }.clear
  36     trap(:USR1) { reopen_worker_logs(worker.nr) }
  37     trap(:QUIT) { G.quit! }
  38     [:TERM, :INT].each { |sig| trap(sig) { exit!(0) } } # instant shutdown
  39     logger.info "Rainbows! #@use worker_connections=#@worker_connections"
  40   end
  41
  42   if IO.respond_to?(:copy_stream)
  43     def write_body(client, body)
  44       if body.respond_to?(:to_path)
  45         IO.copy_stream(Rainbows.body_to_io(body), client)
  46       else
  47         body.each { |chunk| client.write(chunk) }
  48       end
  49       ensure
  50         body.respond_to?(:close) and body.close
  51     end
  52   else
  53     def write_body(client, body)
  54       body.each { |chunk| client.write(chunk) }
  55       ensure
  56         body.respond_to?(:close) and body.close
  57     end
  58   end
  59
  60   module_function :write_body
  61
  62   def wait_headers_readable(client)
  63     IO.select([client], nil, nil, G.kato)
  64   end
  65
  66   # once a client is accepted, it is processed in its entirety here
  67   # in 3 easy steps: read request, call app, write app response
  68   # this is used by synchronous concurrency models
  69   #   Base, ThreadSpawn, ThreadPool
  70   def process_client(client)
  71     buf = client.readpartial(CHUNK_SIZE) # accept filters protect us here
  72     hp = HttpParser.new
  73     env = {}
  74     alive = true
  75     remote_addr = Rainbows.addr(client)
  76
  77     begin # loop
  78       until hp.headers(env, buf)
  79         wait_headers_readable(client) or return
  80         buf << client.readpartial(CHUNK_SIZE)
  81       end
  82
  83       env[CLIENT_IO] = client
  84       env[RACK_INPUT] = 0 == hp.content_length ?
  85                         NULL_IO : TeeInput.new(client, env, hp, buf)
  86       env[REMOTE_ADDR] = remote_addr
  87       status, headers, body = app.call(env.update(RACK_DEFAULTS))
  88
  89       if 100 == status.to_i
  90         client.write(EXPECT_100_RESPONSE)
  91         env.delete(HTTP_EXPECT)
  92         status, headers, body = app.call(env)
  93       end
  94
  95       alive = hp.keepalive? && G.alive
  96       if hp.headers?
  97         out = [ alive ? CONN_ALIVE : CONN_CLOSE ]
  98         client.write(HttpResponse.header_string(status, headers, out))
  99       end
 100       write_body(client, body)
 101     end while alive and hp.reset.nil? and env.clear
 102   # if we get any error, try to write something back to the client
 103   # assuming we haven't closed the socket, but don't get hung up
 104   # if the socket is already closed or broken.  We'll always ensure
 105   # the socket is closed at the end of this function
 106   rescue => e
 107     Rainbows::Error.write(client, e)
 108   ensure
 109     client.close unless client.closed?
 110   end
 111
 112   def self.included(klass)
 113     klass.const_set :LISTENERS, Rainbows::HttpServer::LISTENERS
 114     klass.const_set :G, Rainbows::G
 115   end
 116
 117   # :startdoc:
 118 end