rack: refactor

6 files changed, 148 insertions, 116 deletions

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