[PATCH] preliminary HTTP/2 support for Rack 1.x apps

[PATCH] preliminary HTTP/2 support for Rack 1.x apps

[Eric Wong]

Add support for serving Rack 1.x apps over HTTP/2
using the pure-Ruby  "http-2" gem by Ilya Grigorik.

This will allow us to act as (an in-process) a proxy layer to
transparently support Rack 1.x apps while taking advantage of
any (and if any) advantages HTTP/2 provides.

Currently, upgrading from an HTTP/1.x connection is not
supported, so it is unlikely to work with real-world clients.

TLS is also untested (and ":scheme" is not handled correctly)
at the moment.

In the future, a "raw" HTTP/2 layer may be supported to allow
usage via rack.hijack or even completely without Rack at all.
---
 examples/yahns_http2_rack1.conf.rb |  14 +++
 lib/yahns/config.rb                |  12 +++
 lib/yahns/http2_rack1.rb           | 184 +++++++++++++++++++++++++++++++++++++
 lib/yahns/http_context.rb          |   9 ++
 test/test_http2_rack1.rb           |  72 +++++++++++++++
 5 files changed, 291 insertions(+)
 create mode 100644 examples/yahns_http2_rack1.conf.rb
 create mode 100644 lib/yahns/http2_rack1.rb
 create mode 100644 test/test_http2_rack1.rb

diff --git a/examples/yahns_http2_rack1.conf.rb b/examples/yahns_http2_rack1.conf.rb
new file mode 100644
index 0000000..ced9db3
--- /dev/null
+++ b/examples/yahns_http2_rack1.conf.rb
@@ -0,0 +1,14 @@
+# To the extent possible under law, Eric Wong has waived all copyright and
+# related or neighboring rights to this examples
+# A typical Rack example for hosting a single Rack application with yahns
+#
+# See yahns_config(5) manpage for more information
+app(:rack, "config.ru", preload: false) do
+  protocols "http", "http2"
+  listen 8080
+  client_max_body_size 1024 * 1024
+  input_buffering true
+  output_buffering true # this lazy by default
+  client_timeout 5
+  persistent_connections true
+end
diff --git a/lib/yahns/config.rb b/lib/yahns/config.rb
index e880d92..6ae5a36 100644
--- a/lib/yahns/config.rb
+++ b/lib/yahns/config.rb
@@ -395,6 +395,18 @@ class Yahns::Config # :nodoc:
       @block.ctx.input_buffer_tmpdir = _check_tmpdir(var, tmpdir)
   end
 
+  def protocols(*opts)
+    _check_in_block(:app, :protocols)
+    allowed = %w(http http2)
+    opts = Array(opts).flatten
+    opts.empty? and raise ArgumentError, 'no protocols specified'
+    bad = opts - allowed
+    bad.any? and raise ArgumentError,
+                        "unrecognized protocols: #{bad.inspect}, "\
+                        "only: #{allowed.inspect} are supported"
+    @block.ctx.protocols = opts
+  end
+
   # used to configure rack.errors destination
   def errors(val)
     var = _check_in_block(:app, :errors)
diff --git a/lib/yahns/http2_rack1.rb b/lib/yahns/http2_rack1.rb
new file mode 100644
index 0000000..62f4def
--- /dev/null
+++ b/lib/yahns/http2_rack1.rb
@@ -0,0 +1,184 @@
+# -*- encoding: binary -*-
+# Copyright (C) 2014, all contributors <yahns-public@yhbt.net>
+# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
+
+require 'http/2' # gem install http-2
+
+# This implements a best-effort HTTP2 -> Rack 1.x translation layer
+# to allow Rack 1.x applications to use HTTP2.
+#
+# note: this is written with Ruby 2.2 optimization in mind via
+# opt_str_freeze, opt_aref_with, opt_aset_with instructions
+# However, this module should work under Ruby 2.1, too.
+module Yahns::Http2Rack1 # :nodoc:
+
+  # Default connection "fast-fail" preamble string as defined by the spec.
+  CONNECTION_PREFACE_MAGIC = HTTP2::CONNECTION_PREFACE_MAGIC
+  NULL_IO = Yahns::HttpClient
+
+  def yahns_init
+    @h2 = nil # false, nil (unknown), or HTTPServer
+    super # HttpClient#yahns_init
+  end
+
+  def yahns_step
+    n = self.class.client_header_buffer_size
+    rbuf = Thread.current[:yahns_rbuf]
+
+    case @h2
+    when HTTP2::Server # already HTTP2
+      case rv = kgio_tryread(n, rbuf)
+      when String
+        @h2.receive(rv)
+      when :wait_readable, :wait_writable, nil
+        return rv
+      end while true
+    when nil # detect if HTTP2
+      case rv = kgio_tryread(n, rbuf)
+      when String
+        # likely, we try this to avoid raising an exception in the common case
+        if rv.start_with?(CONNECTION_PREFACE_MAGIC)
+          h2_init.receive(rv)
+          break # to outer loop where we'll see @h2 is an HTTP2::Server
+        elsif rv.size < CONNECTION_PREFACE_MAGIC.bytesize # unlikely
+          begin
+            h2_init.receive(rv) # TODO: deal with client trickling
+          rescue HTTP2::Error::HandshakeError
+            return nil
+          end
+        else
+          # TODO: deal with upgrade
+          @h2 = false # fall back to HTTP/1.x
+          if @hs.add_parse(rv) # fast path dispatch for short requests
+            case input = input_ready
+            when :wait_readable, :wait_writable, :close
+              return input
+            when false
+              break # to outer loop to reevaluate @state == :body in HTTP/1.x
+            else
+              return app_call(input)
+            end
+          end
+          # break outer loop where we'll see @h2 == false
+        end
+      when :wait_readable, :wait_writable, nil
+        return rv
+      end while true
+    when false
+      super # Yahns::HttpClient#yahns_step for HTTP/1.x
+    end while true
+  rescue => e
+    handle_error(e)
+  end
+
+  class H2Logger
+    def initialize(id)
+      @id = id
+    end
+
+    def info(msg)
+      # warn "[Stream #{@id}]: #{msg}"
+    end
+  end
+
+  def h2_init
+    @hs = nil
+    @h2 = HTTP2::Server.new
+    @h2.on(:frame) { |b| kgio_write(b) } # TODO: output buffer for slow clients
+    # @h2.on(:frame_sent) { |frame| warn "> #{frame.inspect}" }
+    # @h2.on(:frame_received) { |frame| warn "< #{frame.inspect}" }
+
+    @h2.on(:stream) do |stream|
+      log = H2Logger.new(stream.id)
+      # these are stream-local variables, not connection-local:
+      req = input = nil
+
+      stream.on(:active) { log.info('stream++') }
+      stream.on(:close) { log.info('stream--') }
+
+      stream.on(:headers) do |headers|
+        log.info "request headers: #{headers.inspect}"
+        req = headers
+      end
+
+      # TODO: support input_buffering :lazy
+      stream.on(:data) do |d|
+        log.info "payload chunk: <<#{d.inspect}>>"
+        (input ||= tmpio_new).write(d)
+      end
+
+      stream.on(:half_close) do
+        log.info "dispatch: #{req.inspect}"
+        req = rack_env_for(req, input)
+        h2_res_emit(stream, *self.class.app.call(req))
+      end
+    end
+    @h2
+  end
+
+  def tmpio_new
+    k = self.class
+    mbs = k.client_max_body_size
+    tmpdir = k.input_buffer_tmpdir
+    mbs ? Yahns::CapInput.new(mbs, tmpdir) : Yahns::TmpIO.new(tmpdir)
+  end
+
+  def h2_res_emit(stream, status, headers, body)
+    headers = Rack::Utils::HeaderHash.new(headers)
+    goaway = headers.delete("connection") =~ /\bclose\b/i
+    headers = headers.to_hash
+    headers[':status'] = status.to_s
+
+    prev = false
+    body.each do |chunk|
+      if headers
+        stream.headers(headers, end_stream: false)
+        headers = nil
+      end
+
+      # only write the previous chunk with end_stream: false if
+      # we get a new one
+      stream.data(prev, end_stream: false) if prev
+      prev = chunk.dup
+    end
+
+    stream.headers(headers, end_stream: !prev) if headers
+    stream.data(prev) if prev # end_stream on the final chunk
+    @h2.goaway if goaway
+  ensure
+    body.close if body.respond_to?(:close)
+  end
+
+  def rack_env_for(req, input)
+    env = self.class.app_defaults.merge(
+      'REMOTE_ADDR' => @kgio_addr,
+      # no hijack support in 2.0 for now...
+      'rack.input' => input ? (input.rewind; input) : NULL_IO,
+      'rack.hijack?' => false
+    )
+
+    req.each do |k, v|
+      case k
+      when ':scheme'
+        env['rack.url_scheme'] = v
+      when ':method'
+        env['REQUEST_METHOD'] = v
+      when ':authority'
+        env['HTTP_HOST'] = v
+      when ':path'
+        env['REQUEST_URI'] = v
+        pi, qs = v.split('?', 2)
+        env['PATH_INFO'] = pi
+        env['QUERY_STRING'] = qs ? "?#{qs}" : ''
+      when /\Acontent-(?:length|type)\z/i
+        # uppercase + underscore, but no "HTTP_" prefix as required by Rack
+        env[k.tr('a-z-', 'A-Z_')] = v
+      else
+        env["HTTP_#{k.tr('-', '_')}"] = v
+      end
+    end
+
+    env['HTTP_VERSION'] = 'HTTP/2.0'.freeze
+    env
+  end
+end
diff --git a/lib/yahns/http_context.rb b/lib/yahns/http_context.rb
index 73bb49a..d61a6d6 100644
--- a/lib/yahns/http_context.rb
+++ b/lib/yahns/http_context.rb
@@ -17,6 +17,7 @@ module Yahns::HttpContext # :nodoc:
   attr_accessor :queue # set right before spawning acceptors
   attr_reader :app
   attr_reader :app_defaults
+  attr_reader :protocols
   attr_writer :input_buffer_tmpdir
   attr_writer :output_buffer_tmpdir
 
@@ -65,6 +66,14 @@ module Yahns::HttpContext # :nodoc:
     @app_defaults["rack.logger"]
   end
 
+  def protocols=(protocols)
+    @protocols = protocols
+    if protocols.include?('http2')
+      require_relative 'http2_rack1'
+      include Yahns::Http2Rack1
+    end
+  end
+
   def mkinput(client, hs)
     (@input_buffering ? Yahns::TeeInput : Yahns::StreamInput).new(client, hs)
   end
diff --git a/test/test_http2_rack1.rb b/test/test_http2_rack1.rb
new file mode 100644
index 0000000..7cf5b4a
--- /dev/null
+++ b/test/test_http2_rack1.rb
@@ -0,0 +1,72 @@
+# Copyright (C) 2014, all contributors <yahns-public@yhbt.net>
+# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
+require_relative 'server_helper'
+begin
+  require 'http/2'
+rescue LoadError
+end
+require 'rack/lobster'
+require 'yahns/rack'
+
+class TestHttp2Rack < Testcase
+  ENV["N"].to_i > 1 and parallelize_me!
+  include ServerHelper
+  alias setup server_helper_setup
+  alias teardown server_helper_teardown
+
+  def test_http2_rack1
+    err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
+    pid = mkserver(cfg) do
+      cfg.instance_eval do
+        ru = lambda { |env|
+          [ 666,
+            [ %w(Content-Length 3), %w(Content-Type text/PLAIN) ],
+            [ "HI\n" ] ]
+        }
+        app(:rack, ru) do
+          protocols %w(http http2)
+          listen "#{host}:#{port}"
+        end
+        stderr_path err.path
+      end
+    end
+    client = TCPSocket.new(host, port)
+    conn = HTTP2::Client.new
+    conn.on(:frame) { |bytes| client.write(bytes) }
+    stream = conn.new_stream
+    response = nil
+    body = ""
+    stream_alive = true
+    stream.on(:headers) { |h| response = h }
+    stream.on(:data) { |d| body << d }
+    stream.on(:close) { stream_alive = false }
+
+    head = {
+      ":scheme" => 'http',
+      ":method" => 'GET',
+      ":authority" => "#{host}:#{port}",
+      ":path" => '/',
+      "accept" => "*/*"
+    }
+    stream.headers(head, end_stream: true)
+    n = 0
+    begin
+      client.wait(60)
+      n += 1
+      conn << client.readpartial(16384)
+    end while stream_alive
+    refute client.closed?
+    res = Hash[*response.flatten]
+    exp = {
+      'content-length' => '3',
+      'content-type' => 'text/PLAIN',
+      ':status' => '666',
+    }
+    assert_equal exp, res
+    assert_equal "HI\n", body
+    assert_equal false, stream_alive
+  ensure
+    client.close if client && !client.closed?
+    quit_wait(pid)
+  end
+end if defined?(::HTTP2)
-- 
EW

Re: [PATCH] preliminary HTTP/2 support for Rack 1.x apps

[Eric Wong]

Eric Wong <e@80x24.org> wrote:
> +++ b/examples/yahns_http2_rack1.conf.rb

> +app(:rack, "config.ru", preload: false) do
> +  protocols "http", "http2"

Maybe the new config option should actually be something like:

     queue(protocol: 'http2') do
       backlog 128
       worker_threads 8
     end

This means we'd have a third type of thread pool...

> +++ b/lib/yahns/http2_rack1.rb

> +      stream.on(:half_close) do
> +        log.info "dispatch: #{req.inspect}"
> +        req = rack_env_for(req, input)
> +        h2_res_emit(stream, *self.class.app.call(req))
> +      end

And maybe enqueue work here:

        stream.on(:half_close) do
          @http2_queue.push([stream, req, input])
        end

And making h2_res_emit thread-safe when called from another thread.

This would also make lazy, synchronous reading of rack.input for
TeeInput/StreamInput easier.

And yeah, the whole SPDY/HTTP-2 thing of reinventing a multi-stream
socket layer inside of TCP connections smells bad to me.