浅谈 Thin 的事件驱动模型

module Rack
 module Handler
 class Thin
 def self.run(app, options={})
 environment = ENV['RACK_ENV'] || 'development'
 default_host = environment == 'development' ? 'localhost' : '0.0.0.0'
 host = options.delete(:Host) || default_host
 port = options.delete(:Port) || 8080
 args = [host, port, app, options]
 args.pop if ::Thin::VERSION::MAJOR < 1 && ::Thin::VERSION::MINOR < 8
 server = ::Thin::Server.new(*args)
 yield server if block_given?
 server.start
 end
 end
 end
end

From: lib/thin/server.rb @ line 100:
Owner: Thin::Server
def initialize(*args, &block)
 host, port, options = DEFAULT_HOST, DEFAULT_PORT, {}
 args.each do |arg|
 case arg
 when 0.class, /^\d+$/ then port = arg.to_i
 when String then host = arg
 when Hash then options = arg
 else
 @app = arg if arg.respond_to?(:call)
 end
 end
 @backend = select_backend(host, port, options)
 @backend.server = self
 @backend.maximum_connections = DEFAULT_MAXIMUM_CONNECTIONS
 @backend.maximum_persistent_connections = DEFAULT_MAXIMUM_PERSISTENT_CONNECTIONS
 @backend.timeout = options[:timeout] || DEFAULT_TIMEOUT
 @app = Rack::Builder.new(&block).to_app if block
end

From: lib/thin/server.rb @ line 261:
Owner: Thin::Server
def select_backend(host, port, options)
 case
 when options.has_key?(:backend)
 raise ArgumentError, ":backend must be a class" unless options[:backend].is_a?(Class)
 options[:backend].new(host, port, options)
 when options.has_key?(:swiftiply)
 Backends::SwiftiplyClient.new(host, port, options)
 when host.include?('/')
 Backends::UnixServer.new(host)
 else
 Backends::TcpServer.new(host, port)
 end
end

From: lib/thin/backends/tcp_server.rb @ line 8:
Owner: Thin::Backends::TcpServer
def initialize(host, port)
 @host = host
 @port = port
 super()
end
From: lib/thin/backends/base.rb @ line 47:
Owner: Thin::Backends::Base
def initialize
 @connections = {}
 @timeout = Server::DEFAULT_TIMEOUT
 @persistent_connection_count = 0
 @maximum_connections = Server::DEFAULT_MAXIMUM_CONNECTIONS
 @maximum_persistent_connections = Server::DEFAULT_MAXIMUM_PERSISTENT_CONNECTIONS
 @no_epoll = false
 @ssl = nil
 @threaded = nil
 @started_reactor = false
end

From: lib/thin/server.rb @ line 152:
Owner: Thin::Server
def start
 raise ArgumentError, 'app required' unless @app
 log_info "Thin web server (v#{VERSION::STRING} codename #{VERSION::CODENAME})"
 log_debug "Debugging ON"
 trace "Tracing ON"
 log_info "Maximum connections set to #{@backend.maximum_connections}"
 log_info "Listening on #{@backend}, CTRL+C to stop"
 @backend.start { setup_signals if @setup_signals }
end

From: lib/thin/backends/base.rb @ line 60:
Owner: Thin::Backends::Base
def start
 @stopping = false
 starter = proc do
 connect
 yield if block_given?
 @running = true
 end
 # Allow for early run up of eventmachine.
 if EventMachine.reactor_running?
 starter.call
 else
 @started_reactor = true
 EventMachine.run(&starter)
 end
end

From: lib/thin/backends/tcp_server.rb @ line 15:
Owner: Thin::Backends::TcpServer
def connect
 @signature = EventMachine.start_server(@host, @port, Connection, &method(:initialize_connection))
 binary_name = EventMachine.get_sockname( @signature )
 port_name = Socket.unpack_sockaddr_in( binary_name )
 @port = port_name[0]
 @host = port_name[1]
 @signature
end

From: lib/thin/backends/base.rb @ line 145:
Owner: Thin::Backends::Base
def initialize_connection(connection)
 connection.backend = self
 connection.app = @server.app
 connection.comm_inactivity_timeout = @timeout
 connection.threaded = @threaded
 connection.start_tls(@ssl_options) if @ssl
 if @persistent_connection_count < @maximum_persistent_connections
 connection.can_persist!
 @persistent_connection_count += 1
 end
 @connections[connection.__id__] = connection
end

From: lib/thin/connection.rb @ line 36:
Owner: Thin::Connection
def receive_data(data)
 @idle = false
 trace data
 process if @request.parse(data)
rescue InvalidRequest => e
 log_error("Invalid request", e)
 post_process Response::BAD_REQUEST
end

From: lib/thin/connection.rb @ line 47:
Owner: Thin::Connection
def process
 if threaded?
 @request.threaded = true
 EventMachine.defer { post_process(pre_process) }
 else
 @request.threaded = false
 post_process(pre_process)
 end
end

From: lib/thin/connection.rb @ line 63:
Owner: Thin::Connection
def pre_process
 @request.remote_address = remote_address
 @request.async_callback = method(:post_process)
 response = AsyncResponse
 catch(:async) do
 response = @app.call(@request.env)
 end
 response
rescue Exception => e
 unexpected_error(e)
 can_persist? && @request.persistent? ? Response::PERSISTENT_ERROR : Response::ERROR
end

From: lib/thin/connection.rb @ line 95:
Owner: Thin::Connection
def post_process(result)
 return unless result
 result = result.to_a
 return if result.first == AsyncResponse.first
 @response.status, @response.headers, @response.body = *result
 @response.each do |chunk|
 send_data chunk
 end
rescue Exception => e
 unexpected_error(e)
 close_connection
ensure
 if @response.body.respond_to?(:callback) && @response.body.respond_to?(:errback)
 @response.body.callback { terminate_request }
 @response.body.errback { terminate_request }
 else
 terminate_request unless result && result.first == AsyncResponse.first
 end
end

From: lib/eventmachine.rb @ line 149:
Owner: #<Class:EventMachine>
def self.run blk=nil, tail=nil, &block
 # ...
 begin
 @reactor_pid = Process.pid
 @reactor_running = true
 initialize_event_machine
 (b = blk || block) and add_timer(0, b)
 if @next_tick_queue && !@next_tick_queue.empty?
 add_timer(0) { signal_loopbreak }
 end
 @reactor_thread = Thread.current
 run_machine
 ensure
 until @tails.empty?
 @tails.pop.call
 end
 release_machine
 cleanup_machine
 @reactor_running = false
 @reactor_thread = nil
 end
end

From: lib/em/pure_ruby.rb @ line 522:
Owner: EventMachine::Reactor
def initialize_for_run
 @running = false
 @stop_scheduled = false
 @selectables ||= {}; @selectables.clear
 @timers = SortedSet.new # []
 set_timer_quantum(0.1)
 @current_loop_time = Time.now
 @next_heartbeat = @current_loop_time + HeartbeatInterval
end

From: lib/em/pure_ruby.rb @ line 540:
Owner: EventMachine::Reactor
def run
 raise Error.new( "already running" ) if @running
 @running = true
 begin
 open_loopbreaker
 loop {
 @current_loop_time = Time.now
 break if @stop_scheduled
 run_timers
 break if @stop_scheduled
 crank_selectables
 break if @stop_scheduled
 run_heartbeats
 }
 ensure
 close_loopbreaker
 @selectables.each {|k, io| io.close}
 @selectables.clear
 @running = false
 end
end

From: lib/em/pure_ruby.rb @ line 540:
Owner: EventMachine::Reactor
def crank_selectables
 readers = @selectables.values.select { |io| io.select_for_reading? }
 writers = @selectables.values.select { |io| io.select_for_writing? }
 s = select(readers, writers, nil, @timer_quantum)
 s and s[1] and s[1].each { |w| w.eventable_write }
 s and s[0] and s[0].each { |r| r.eventable_read }
 @selectables.delete_if {|k,io|
 if io.close_scheduled?
 io.close
 begin
 EventMachine::event_callback io.uuid, ConnectionUnbound, nil
 rescue ConnectionNotBound; end
 true
 end
 }
end

From: lib/eventmachine.rb @ line 516:
Owner: #<Class:EventMachine>
def self.start_server server, port=nil, handler=nil, *args, &block
 port = Integer(port)
 klass = klass_from_handler(Connection, handler, *args)
 s = if port
 start_tcp_server server, port
 else
 start_unix_server server
 end
 @acceptors[s] = [klass, args, block]
 s
end

From: lib/em/pure_ruby.rb @ line 184:
Owner: #<Class:EventMachine>
def self.start_tcp_server host, port
 (s = EvmaTCPServer.start_server host, port) or raise "no acceptor"
 s.uuid
end

From: lib/em/pure_ruby.rb @ line 1108:
Owner: #<Class:EventMachine::EvmaTCPServer>
def self.start_server host, port
 sd = Socket.new( Socket::AF_LOCAL, Socket::SOCK_STREAM, 0 )
 sd.setsockopt( Socket::SOL_SOCKET, Socket::SO_REUSEADDR, true )
 sd.bind( Socket.pack_sockaddr_in( port, host ))
 sd.listen( 50 ) # 5 is what you see in all the books. Ain't enough.
 EvmaTCPServer.new sd
end

From: lib/em/pure_ruby.rb @ line 687:
Owner: EventMachine::Selectable
def initialize io
 @io = io
 @uuid = UuidGenerator.generate
 @is_server = false
 @last_activity = Reactor.instance.current_loop_time
 m = @io.fcntl(Fcntl::F_GETFL, 0)
 @io.fcntl(Fcntl::F_SETFL, Fcntl::O_NONBLOCK | m)
 @close_scheduled = false
 @close_requested = false
 se = self; @io.instance_eval { @my_selectable = se }
 Reactor.instance.add_selectable @io
end

From: lib/em/pure_ruby.rb @ line 532:
Owner: EventMachine::Reactor
def add_selectable io
 @selectables[io.uuid] = io
end

From: lib/em/pure_ruby.rb @ line 1130:
Owner: EventMachine::EvmaTCPServer
def eventable_read
 begin
 10.times {
 descriptor, peername = io.accept_nonblock
 sd = EvmaTCPClient.new descriptor
 sd.is_server = true
 EventMachine::event_callback uuid, ConnectionAccepted, sd.uuid
 }
 rescue Errno::EWOULDBLOCK, Errno::EAGAIN
 end
end

From: lib/eventmachine.rb @ line 1462:
Owner: #<Class:EventMachine>
def self.event_callback conn_binding, opcode, data
 if opcode == # ...
 # ...
 elsif opcode == ConnectionAccepted
 accep, args, blk = @acceptors[conn_binding]
 raise NoHandlerForAcceptedConnection unless accep
 c = accep.new data, *args
 @conns[data] = c
 blk and blk.call(c)
 c
 else
 # ...
 end
end

From: lib/em/pure_ruby.rb @ line 1130:
Owner: EventMachine::StreamObject
def eventable_read
 @last_activity = Reactor.instance.current_loop_time
 begin
 if io.respond_to?(:read_nonblock)
 10.times {
 data = io.read_nonblock(4096)
 EventMachine::event_callback uuid, ConnectionData, data
 }
 else
 data = io.sysread(4096)
 EventMachine::event_callback uuid, ConnectionData, data
 end
 rescue Errno::EAGAIN, Errno::EWOULDBLOCK, SSLConnectionWaitReadable
 rescue Errno::ECONNRESET, Errno::ECONNREFUSED, EOFError, Errno::EPIPE, OpenSSL::SSL::SSLError
 @close_scheduled = true
 EventMachine::event_callback uuid, ConnectionUnbound, nil
 end
end

From: lib/eventmachine.rb @ line 1462:
Owner: #<Class:EventMachine>
def self.event_callback conn_binding, opcode, data
 if opcode == # ...
 # ...
 elsif opcode == ConnectionData
 c = @conns[conn_binding] or raise ConnectionNotBound, "received data #{data} for unknown signature: #{conn_binding}"
 c.receive_data data
 else
 # ...
 end
end

From: lib/em/connection.rb @ line 324:
Owner: EventMachine::Connection
def send_data data
 data = data.to_s
 size = data.bytesize if data.respond_to?(:bytesize)
 size ||= data.size
 EventMachine::send_data @signature, data, size
end
From: lib/em/pure_ruby.rb @ line 172:
Owner: #<Class:EventMachine>
def self.send_data target, data, datalength
 selectable = Reactor.instance.get_selectable( target ) or raise "unknown send_data target"
 selectable.send_data data
end
From: lib/em/pure_ruby.rb @ line 851:
Owner: EventMachine::StreamObject
def send_data data
 unless @close_scheduled or @close_requested or !data or data.length <= 0
 @outbound_q << data.to_s
 end
end

From: lib/em/pure_ruby.rb @ line 823:
Owner: EventMachine::StreamObject
def eventable_write
 @last_activity = Reactor.instance.current_loop_time
 while data = @outbound_q.shift do
 begin
 data = data.to_s
 w = if io.respond_to?(:write_nonblock)
 io.write_nonblock data
 else
 io.syswrite data
 end
 if w < data.length
 @outbound_q.unshift data[w..-1]
 break
 end
 rescue Errno::EAGAIN, SSLConnectionWaitReadable, SSLConnectionWaitWritable
 @outbound_q.unshift data
 break
 rescue EOFError, Errno::ECONNRESET, Errno::ECONNREFUSED, Errno::EPIPE, OpenSSL::SSL::SSLError
 @close_scheduled = true
 @outbound_q.clear
 end
 end
end

From: lib/em/pure_ruby.rb @ line 540:
Owner: EventMachine::Reactor
def crank_selectables
 # ...
 @selectables.delete_if {|k,io|
 if io.close_scheduled?
 io.close
 begin
 EventMachine::event_callback io.uuid, ConnectionUnbound, nil
 rescue ConnectionNotBound; end
 true
 end
 }
end

def self.event_callback conn_binding, opcode, data
 if opcode == ConnectionUnbound
 if c = @conns.delete( conn_binding )
 c.unbind
 io = c.instance_variable_get(:@io)
 begin
 io.close
 rescue Errno::EBADF, IOError
 end
 elsif c = @acceptors.delete( conn_binding )
 else
 raise ConnectionNotBound, "received ConnectionUnbound for an unknown signature: #{conn_binding}"
 end
 elsif opcode = 1
 #...
 end
end