From b8dc2b931a6525d102e657ec9cb951548cbe7925 Mon Sep 17 00:00:00 2001 From: Kazuki Yamaguchi Date: Thu, 14 Jan 2016 12:44:28 +0900 Subject: rack: refactor --- lib/plum/rack.rb | 1 + lib/plum/rack/legacy_session.rb | 36 +++++++++++++++++ lib/plum/rack/listener.rb | 85 +++++++++++++++++++++++++++++++++++---- lib/plum/rack/server.rb | 89 +++++++++-------------------------------- lib/plum/rack/session.rb | 28 ++++++------- lib/plum/rack/thread_pool.rb | 25 +----------- 6 files changed, 148 insertions(+), 116 deletions(-) create mode 100644 lib/plum/rack/legacy_session.rb diff --git a/lib/plum/rack.rb b/lib/plum/rack.rb index 923389b..a9d2c66 100644 --- a/lib/plum/rack.rb +++ b/lib/plum/rack.rb @@ -8,4 +8,5 @@ require "plum/rack/dsl" require "plum/rack/listener" require "plum/rack/server" require "plum/rack/session" +require "plum/rack/legacy_session" require "plum/rack/thread_pool" diff --git a/lib/plum/rack/legacy_session.rb b/lib/plum/rack/legacy_session.rb new file mode 100644 index 0000000..6712933 --- /dev/null +++ b/lib/plum/rack/legacy_session.rb @@ -0,0 +1,36 @@ +# -*- frozen-string-literal: true -*- +using Plum::BinaryString + +module Plum + module Rack + class LegacySession + def initialize(svc, e, sock) + @svc = svc + @e = e + @sock = sock + @config = svc.config + end + + def run + if @config[:fallback_legacy_host] + @logger.info "legacy HTTP: fallbacking to: #{@config[:fallback_legacy_host]}:#{@config[:fallback_legacy_port]}" + upstream = TCPSocket.open(@config[:fallback_legacy_host], @config[:fallback_legacy_port]) + upstream.write(@e.buf) if @e.buf + loop do + ret = IO.select([@sock, upstream]) + ret[0].each { |s| + a = s.readpartial(65536) + if s == upstream + @sock.write(a) + else + upstream.write(a) + end + } + end + end + ensure + upstream.close if upstream + end + end + end +end diff --git a/lib/plum/rack/listener.rb b/lib/plum/rack/listener.rb index 3ff6c4a..a221ade 100644 --- a/lib/plum/rack/listener.rb +++ b/lib/plum/rack/listener.rb @@ -10,6 +10,10 @@ module Plum raise "not implemented" end + def accept(svc) + raise "not implemented" + end + def method_missing(name, *args) @server.__send__(name, *args) end @@ -24,8 +28,24 @@ module Plum @server.to_io end - def plum(sock) - ::Plum::HTTPServerConnection.new(sock.method(:write)) + def accept(svc) + sock = @server.accept + Thread.start { + begin + plum = ::Plum::HTTPServerConnection.new(sock.method(:write)) + sess = Session.new(svc, sock, plum) + sess.run + rescue Errno::ECONNRESET, EOFError # closed + rescue ::Plum::LegacyHTTPError => e + @logger.info "legacy HTTP client: #{e}" + sess = LegacySession.new(svc, e, sock) + sess.run + rescue => e + svc.log_exception(e) + ensure + sock.close + end + } end end @@ -57,7 +77,7 @@ module Plum } tcp_server = ::TCPServer.new(lc[:hostname], lc[:port]) @server = OpenSSL::SSL::SSLServer.new(tcp_server, ctx) - @server.start_immediately = false + @server.start_immediately = false # call socket#accept twice: [tcp, tls] end def parse_chained_cert(str) @@ -68,12 +88,49 @@ module Plum @server.to_io end - def plum(sock) - raise ::Plum::LegacyHTTPError.new("client didn't offer h2 with ALPN", nil) unless sock.alpn_protocol == "h2" - ::Plum::ServerConnection.new(sock.method(:write)) + def accept(svc) + sock = @server.accept + Thread.start { + begin + sock = sock.accept + raise ::Plum::LegacyHTTPError.new("client didn't offer h2 with ALPN", nil) unless sock.alpn_protocol == "h2" + plum = ::Plum::ServerConnection.new(sock.method(:write)) + sess = Session.new(svc, sock, plum) + sess.run + rescue Errno::ECONNRESET, EOFError # closed + rescue ::Plum::LegacyHTTPError => e + @logger.info "legacy HTTP client: #{e}" + sess = LegacySession.new(svc, e, sock) + sess.run + rescue => e + svc.log_exception(e) + ensure + sock.close if sock + end + } end private + def handle_legacy(svc, e, sock) + if svc.config[:fallback_legacy_host] + svc.logger.info "legacy HTTP: fallbacking to: #{svc.config[:fallback_legacy_host]}:#{svc.config[:fallback_legacy_port]}" + upstream = TCPSocket.open(svc.config[:fallback_legacy_host], svc.config[:fallback_legacy_port]) + upstream.write(e.buf) if e.buf + loop do + ret = IO.select([sock, upstream]) + ret[0].each { |s| + a = s.readpartial(65536) + if s == upstream + sock.write(a) + else + upstream.write(a) + end + } + end + end + ensure + upstream.close if upstream + end # returns: [cert, key] def dummy_key STDERR.puts "WARNING: Generating new dummy certificate..." @@ -126,8 +183,20 @@ module Plum @server.to_io end - def plum(sock) - ::Plum::ServerConnection.new(sock.method(:write)) + def accept(svc) + sock = @server.accept + Thread.start { + begin + plum = ::Plum::ServerConnection.new(sock.method(:write)) + sess = Session.new(svc, sock, plum) + sess.run + rescue Errno::ECONNRESET, EOFError # closed + rescue => e + svc.log_exception(e) + ensure + sock.close if sock + end + } end end end diff --git a/lib/plum/rack/server.rb b/lib/plum/rack/server.rb index 3bd48ff..9aedc20 100644 --- a/lib/plum/rack/server.rb +++ b/lib/plum/rack/server.rb @@ -2,21 +2,15 @@ module Plum module Rack class Server - attr_reader :config + attr_reader :config, :app, :logger, :threadpool def initialize(app, config) @config = config @state = :null @app = config[:debug] ? ::Rack::CommonLogger.new(app) : app - @logger = Logger.new(config[:log] || $stdout).tap { |l| - l.level = config[:debug] ? Logger::DEBUG : Logger::INFO - } - @listeners = config[:listeners].map { |lc| - lc[:listener].new(lc) - } - if @config[:threaded] - @threadpool = ThreadPool.new(@config[:threadpool_size]) - end + @logger = Logger.new(config[:log] || $stdout).tap { |l| l.level = config[:debug] ? Logger::DEBUG : Logger::INFO } + @listeners = config[:listeners].map { |lc| lc[:listener].new(lc) } + @threadpool = ThreadPool.new(@config[:threadpool_size]) if @config[:threaded] @logger.info("Plum #{::Plum::VERSION}") @logger.info("Config: #{config}") @@ -27,88 +21,41 @@ module Plum end def start + #trap(:INT) { @state = :ee } + #require "lineprof" + #Lineprof.profile(//){ @state = :running - while @state == :running - break if @listeners.empty? + while @state == :running && !@listeners.empty? begin if ss = IO.select(@listeners, nil, nil, 2.0) ss[0].each { |svr| - new_con(svr) + begin + svr.accept(self) + rescue Errno::ECONNRESET, Errno::ECONNABORTED # closed + rescue => e + log_exception(e) + end } end - rescue Errno::EBADF, Errno::ENOTSOCK, IOError => e # closed - rescue StandardError => e + rescue Errno::EBADF # closed + rescue => e log_exception(e) end end + #} end def stop @state = :stop @listeners.map(&:stop) - # TODO: gracefully shutdown connections + # TODO: gracefully shutdown connections (wait threadpool?) end private - def new_con(svr) - sock = svr.accept - Thread.new { - begin - begin - sock = sock.accept if sock.respond_to?(:accept) - plum = svr.plum(sock) - - con = Session.new(app: @app, - plum: plum, - sock: sock, - logger: @logger, - config: @config, - remote_addr: sock.peeraddr.last, - threadpool: @threadpool) - con.run - rescue ::Plum::LegacyHTTPError => e - @logger.info "legacy HTTP client: #{e}" - handle_legacy(e, sock) - end - rescue Errno::ECONNRESET, Errno::EPROTO, Errno::EINVAL, EOFError => e # closed - rescue StandardError => e - log_exception(e) - ensure - sock.close if sock - end - } - rescue Errno::ECONNRESET, Errno::ECONNABORTED, Errno::EPROTO, Errno::EINVAL => e # closed - sock.close if sock - rescue StandardError => e - log_exception(e) - sock.close if sock - end - def log_exception(e) @logger.error("#{e.class}: #{e.message}\n#{e.backtrace.map { |b| "\t#{b}" }.join("\n")}") end - def handle_legacy(e, sock) - if @config[:fallback_legacy_host] - @logger.info "legacy HTTP: fallbacking to: #{@config[:fallback_legacy_host]}:#{@config[:fallback_legacy_port]}" - upstream = TCPSocket.open(@config[:fallback_legacy_host], @config[:fallback_legacy_port]) - upstream.write(e.buf) if e.buf - loop do - ret = IO.select([sock, upstream]) - ret[0].each { |s| - a = s.readpartial(65536) - if s == upstream - sock.write(a) - else - upstream.write(a) - end - } - end - end - ensure - upstream.close if upstream - end - def drop_privileges begin user = @config[:user] diff --git a/lib/plum/rack/session.rb b/lib/plum/rack/session.rb index c40f609..3a27248 100644 --- a/lib/plum/rack/session.rb +++ b/lib/plum/rack/session.rb @@ -8,15 +8,15 @@ module Plum class Session attr_reader :app, :plum - def initialize(app:, plum:, sock:, logger:, config:, remote_addr:, threadpool:) - @app = app - @plum = plum + def initialize(svc, sock, plum) + @svc = svc + @app = svc.app @sock = sock - @logger = logger - @config = config - @remote_addr = remote_addr - @threadpool = threadpool - @request_tokens = Set.new + @plum = plum + @logger = svc.logger + @config = svc.config + @remote_addr = sock.peeraddr.last + @threadpool = svc.threadpool setup_plum end @@ -25,12 +25,15 @@ module Plum @plum.close end + def to_io + @sock.to_io + end + def run while !@sock.closed? && !@sock.eof? @plum << @sock.readpartial(1024) end ensure - @request_tokens.each { |token| @threadpool.cancel(token) } if @threadpool stop end @@ -64,9 +67,8 @@ module Plum @logger.error(err) } if @threadpool - headers = req[:headers] - @request_tokens << @threadpool.acquire(headers, err) { - handle_request(stream, headers, req[:data]) + @threadpool.acquire(err) { + handle_request(stream, req[:headers], req[:data]) } else begin @@ -149,8 +151,6 @@ module Plum send_body(st, p_body) unless pno_body } end - - @request_tokens.delete(headers) # headers is tag end def new_env(h, data) diff --git a/lib/plum/rack/thread_pool.rb b/lib/plum/rack/thread_pool.rb index 7c3fc48..9e07943 100644 --- a/lib/plum/rack/thread_pool.rb +++ b/lib/plum/rack/thread_pool.rb @@ -5,9 +5,6 @@ module Plum def initialize(size = 20) @workers = Set.new @jobs = Queue.new - @tags = {} - @cancels = Set.new - @mutex = Mutex.new size.times { |i| spawn_worker @@ -16,37 +13,19 @@ module Plum # returns cancel token def acquire(tag = nil, err = nil, &blk) - @jobs << [blk, err, tag] - tag - end - - def cancel(tag) - worker = @mutex.synchronize { @tags.delete?(tag) || (@cancels << tag; return) } - @workers.delete(worker) - worker.kill - spawn_worker + @jobs << [blk, err] end private def spawn_worker t = Thread.new { while true - job, err, tag = @jobs.pop - if tag - next if @mutex.synchronize { - c = @cancels.delete?(tag) - @tags[tag] = self unless c - c - } - end - + job, err = @jobs.pop begin job.call rescue => e err << e if err end - - @mutex.synchronize { @tags.delete(tag) } if tag end } @workers << t -- cgit v1.2.3