lib/rainbows/event_machine.rb

   1 # -*- encoding: binary -*-
   2 require 'eventmachine'
   3 EM::VERSION >= '0.12.10' or abort 'eventmachine 0.12.10 is required'
   4
   5 # Implements a basic single-threaded event model with
   6 # {EventMachine}[http://rubyeventmachine.com/].  It is capable of
   7 # handling thousands of simultaneous client connections, but with only
   8 # a single-threaded app dispatch.  It is suited for slow clients,
   9 # and can work with slow applications via asynchronous libraries such as
  10 # {async_sinatra}[http://github.com/raggi/async_sinatra],
  11 # {Cramp}[http://cramp.in/],
  12 # and {rack-fiber_pool}[http://github.com/mperham/rack-fiber_pool].
  13 #
  14 # It does not require your Rack application to be thread-safe,
  15 # reentrancy is only required for the DevFdResponse body
  16 # generator.
  17 #
  18 # Compatibility: Whatever \EventMachine ~> 0.12.10 and Unicorn both
  19 # support, currently Ruby 1.8/1.9.
  20 #
  21 # This model is compatible with users of "async.callback" in the Rack
  22 # environment such as
  23 # {async_sinatra}[http://github.com/raggi/async_sinatra].
  24 #
  25 # For a complete asynchronous framework,
  26 # {Cramp}[http://cramp.in/] is fully
  27 # supported when using this concurrency model.
  28 #
  29 # This model is fully-compatible with
  30 # {rack-fiber_pool}[http://github.com/mperham/rack-fiber_pool]
  31 # which allows each request to run inside its own \Fiber after
  32 # all request processing is complete.
  33 #
  34 # Merb (and other frameworks/apps) supporting +deferred?+ execution as
  35 # documented at Rainbows::EventMachine::TryDefer
  36 #
  37 # This model does not implement as streaming "rack.input" which allows
  38 # the Rack application to process data as it arrives.  This means
  39 # "rack.input" will be fully buffered in memory or to a temporary file
  40 # before the application is entered.
  41 #
  42 # === RubyGem Requirements
  43 #
  44 # * event_machine 0.12.10
  45 module Rainbows::EventMachine
  46   autoload :ResponsePipe, 'rainbows/event_machine/response_pipe'
  47   autoload :ResponseChunkPipe, 'rainbows/event_machine/response_chunk_pipe'
  48   autoload :TryDefer, 'rainbows/event_machine/try_defer'
  49   autoload :Client, 'rainbows/event_machine/client'
  50
  51   include Rainbows::Base
  52
  53   # Cramp (and possibly others) can subclass Rainbows::EventMachine::Client
  54   # and provide the :em_client_class option.  We /don't/ want to load
  55   # Rainbows::EventMachine::Client in the master process since we need
  56   # reloadability.
  57   def em_client_class
  58     case klass = Rainbows::O[:em_client_class]
  59     when Proc
  60       klass.call # e.g.: proc { Cramp::WebSocket::Rainbows }
  61     when Symbol, String
  62       eval(klass.to_s) # Object.const_get won't resolve multi-level paths
  63     else # @use should be either :EventMachine or :NeverBlock
  64       Rainbows.const_get(@use).const_get(:Client)
  65     end
  66   end
  67
  68   def defers_finished?
  69     # EventMachine 1.0.0+ has defers_finished?
  70     EM.respond_to?(:defers_finished?) ? EM.defers_finished? : true
  71   end
  72
  73   # runs inside each forked worker, this sits around and waits
  74   # for connections and doesn't die until the parent dies (or is
  75   # given a INT, QUIT, or TERM signal)
  76   def worker_loop(worker) # :nodoc:
  77     init_worker_process(worker)
  78     server = Rainbows.server
  79     server.app.respond_to?(:deferred?) and
  80       server.app = TryDefer.new(server.app)
  81
  82     # enable them both, should be non-fatal if not supported
  83     EM.epoll
  84     EM.kqueue
  85     logger.info "#@use: epoll=#{EM.epoll?} kqueue=#{EM.kqueue?}"
  86     client_class = em_client_class
  87     max = worker_connections + LISTENERS.size
  88     Rainbows::EventMachine::Server.const_set(:MAX, max)
  89     Rainbows::EventMachine::Server.const_set(:CL, client_class)
  90     Rainbows::EventMachine::Client.const_set(:APP, Rainbows.server.app)
  91     EM.run {
  92       conns = EM.instance_variable_get(:@conns) or
  93         raise RuntimeError, "EM @conns instance variable not accessible!"
  94       Rainbows::EventMachine::Server.const_set(:CUR, conns)
  95       Rainbows.at_quit do
  96         EM.next_tick do
  97           LISTENERS.clear
  98           conns.each_value do |c|
  99             case c
 100             when client_class
 101               c.quit
 102             when Rainbows::EventMachine::Server
 103               c.detach
 104             end
 105           end
 106         end
 107       end
 108       EM.add_periodic_timer(1) do
 109         if ! Rainbows.tick && conns.empty? && defers_finished? &&
 110             EM.reactor_running?
 111           EM.stop
 112         end
 113       end
 114       LISTENERS.map! do |s|
 115         EM.watch(s, Rainbows::EventMachine::Server) do |c|
 116           c.notify_readable = true
 117         end
 118       end
 119     }
 120     EM.reactor_thread.join if EM.reactor_running?
 121   end
 122 end
 123 # :enddoc:
 124 require 'rainbows/event_machine/server'