lib/rainbows/revactor.rb

   1 # -*- encoding: binary -*-
   2 require 'revactor'
   3 require 'fcntl'
   4 Revactor::VERSION >= '0.1.5' or abort 'revactor 0.1.5 is required'
   5
   6 # Enables use of the Actor model through
   7 # {Revactor}[http://revactor.org] under Ruby 1.9.  It spawns one
   8 # long-lived Actor for every listen socket in the process and spawns a
   9 # new Actor for every client connection accept()-ed.
  10 # +worker_connections+ will limit the number of client Actors we have
  11 # running at any one time.
  12 #
  13 # Applications using this model are required to be reentrant, but do
  14 # not have to worry about race conditions unless they use threads
  15 # internally.  \Rainbows! does not spawn threads under this model.
  16 # Multiple instances of the same app may run in the same address space
  17 # sequentially (but at interleaved points).  Any network dependencies
  18 # in the application using this model should be implemented using the
  19 # \Revactor library as well, to take advantage of the networking
  20 # concurrency features this model provides.
  21 module Rainbows::Revactor
  22
  23   # :stopdoc:
  24   RD_ARGS = {}
  25
  26   autoload :Proxy, 'rainbows/revactor/proxy'
  27
  28   include Rainbows::Base
  29   LOCALHOST = Kgio::LOCALHOST
  30   TCP = ::Revactor::TCP::Socket
  31
  32   # once a client is accepted, it is processed in its entirety here
  33   # in 3 easy steps: read request, call app, write app response
  34   def process_client(client) # :nodoc:
  35     io = client.instance_variable_get(:@_io)
  36     io.fcntl(Fcntl::F_SETFD, Fcntl::FD_CLOEXEC)
  37     rd_args = [ nil ]
  38     remote_addr = if TCP === client
  39       rd_args << RD_ARGS
  40       client.remote_addr
  41     else
  42       LOCALHOST
  43     end
  44     hp = Unicorn::HttpParser.new
  45     buf = hp.buf
  46
  47     begin
  48       until env = hp.parse
  49         buf << client.read(*rd_args)
  50       end
  51
  52       env[CLIENT_IO] = client
  53       env[RACK_INPUT] = 0 == hp.content_length ?
  54                NULL_IO : TeeInput.new(TeeSocket.new(client), hp)
  55       env[REMOTE_ADDR] = remote_addr
  56       status, headers, body = app.call(env.update(RACK_DEFAULTS))
  57
  58       if 100 == status.to_i
  59         client.write(EXPECT_100_RESPONSE)
  60         env.delete(HTTP_EXPECT)
  61         status, headers, body = app.call(env)
  62       end
  63
  64       if hp.headers?
  65         headers = HH.new(headers)
  66         range = make_range!(env, status, headers) and status = range.shift
  67         env = hp.keepalive? && G.alive && G.kato > 0
  68         headers[CONNECTION] = env ? KEEP_ALIVE : CLOSE
  69         client.write(response_header(status, headers))
  70       end
  71       write_body(client, body, range)
  72     end while env && hp.reset.nil?
  73   rescue ::Revactor::TCP::ReadError
  74   rescue => e
  75     Rainbows::Error.write(io, e)
  76   ensure
  77     client.close
  78   end
  79
  80   # runs inside each forked worker, this sits around and waits
  81   # for connections and doesn't die until the parent dies (or is
  82   # given a INT, QUIT, or TERM signal)
  83   def worker_loop(worker) #:nodoc:
  84     init_worker_process(worker)
  85     require 'rainbows/revactor/body'
  86     self.class.__send__(:include, Rainbows::Revactor::Body)
  87     RD_ARGS[:timeout] = G.kato if G.kato > 0
  88     nr = 0
  89     limit = worker_connections
  90     actor_exit = Case[:exit, Actor, Object]
  91
  92     revactorize_listeners.each do |l, close, accept|
  93       Actor.spawn(l, close, accept) do |l, close, accept|
  94         Actor.current.trap_exit = true
  95         l.controller = l.instance_variable_set(:@receiver, Actor.current)
  96         begin
  97           while nr >= limit
  98             l.disable if l.enabled?
  99             logger.info "busy: clients=#{nr} >= limit=#{limit}"
 100             Actor.receive do |f|
 101               f.when(close) {}
 102               f.when(actor_exit) { nr -= 1 }
 103               f.after(0.01) {} # another listener could've gotten an exit
 104             end
 105           end
 106
 107           l.enable unless l.enabled?
 108           Actor.receive do |f|
 109             f.when(close) {}
 110             f.when(actor_exit) { nr -= 1 }
 111             f.when(accept) do |_, _, s|
 112               nr += 1
 113               Actor.spawn_link(s) { |c| process_client(c) }
 114             end
 115           end
 116         rescue => e
 117           Rainbows::Error.listen_loop(e)
 118         end while G.alive
 119         Actor.receive do |f|
 120           f.when(close) {}
 121           f.when(actor_exit) { nr -= 1 }
 122         end while nr > 0
 123       end
 124     end
 125
 126     Actor.sleep 1 while G.tick || nr > 0
 127     rescue Errno::EMFILE
 128       # ignore, let another worker process take it
 129   end
 130
 131   def revactorize_listeners
 132     LISTENERS.map do |s|
 133       case s
 134       when TCPServer
 135         l = ::Revactor::TCP.listen(s, nil)
 136         [ l, T[:tcp_closed, ::Revactor::TCP::Socket],
 137           T[:tcp_connection, l, ::Revactor::TCP::Socket] ]
 138       when UNIXServer
 139         l = ::Revactor::UNIX.listen(s)
 140         [ l, T[:unix_closed, ::Revactor::UNIX::Socket ],
 141           T[:unix_connection, l, ::Revactor::UNIX::Socket] ]
 142       end
 143     end
 144   end
 145
 146   # Revactor Sockets do not implement readpartial, so we emulate just
 147   # enough to avoid mucking with TeeInput internals.  Fortunately
 148   # this code is not heavily used so we can usually avoid the overhead
 149   # of adding a userspace buffer.
 150   class TeeSocket
 151     def initialize(socket)
 152       # IO::Buffer is used internally by Rev which Revactor is based on
 153       # so we'll always have it available
 154       @socket, @rbuf = socket, IO::Buffer.new
 155     end
 156
 157     # Revactor socket reads always return an unspecified amount,
 158     # sometimes too much
 159     def kgio_read(length, dst = "")
 160       return dst.replace("") if length == 0
 161
 162       # always check and return from the userspace buffer first
 163       @rbuf.size > 0 and return dst.replace(@rbuf.read(length))
 164
 165       # read off the socket since there was nothing in rbuf
 166       tmp = @socket.read
 167
 168       # we didn't read too much, good, just return it straight back
 169       # to avoid needlessly wasting memory bandwidth
 170       tmp.size <= length and return dst.replace(tmp)
 171
 172       # ugh, read returned too much
 173       @rbuf << tmp[length, tmp.size]
 174       dst.replace(tmp[0, length])
 175       rescue EOFError
 176     end
 177
 178     # just proxy any remaining methods TeeInput may use
 179     def close
 180       @socket.close
 181     end
 182   end
 183
 184   # :startdoc:
 185 end