[PATCH 0/2] proxy_pass: fleshing out trailer support

[PATCH 0/2] proxy_pass: fleshing out trailer support

[Eric Wong]

Unfortunately, we need to support response trailers sent from upstreams
to our HTTP/1.1 clients, as trailers are a(n obscure) part of HTTP/1.1.

Then, we must also prevent trailers from being sent to HTTP/1.0 clients,
as HTTP/1.0 clients do not support reading trailers (or chunked
encoding, a prerequisite of trailers).  The only way to send trailers to
HTTP/1.0 clients is to send them as headers, but this is highly
innefficient, especially for fast clients.  So for HTTP/1.0 clients, we
shall forward HTTP/1.0 requests to the backend which should prevent
trailers from being sent at all.

Eric Wong (2):
      proxy_pass: preliminary support for passing trailers
      proxy_pass: send 1.0 requests to upstreams for 1.0 clients

 lib/yahns/proxy_http_response.rb | 102 +++++++++++++++++++++------------------
 lib/yahns/proxy_pass.rb          |   9 +++-
 test/test_proxy_pass.rb          |  84 +++++++++++++++++++++++++++++---
 3 files changed, 141 insertions(+), 54 deletions(-)

[PATCH 1/2] proxy_pass: preliminary support for passing trailers

[Eric Wong]

Rack apps may (through a round-about way) send HTTP trailers
to HTTP/1.1 clients, and we need a way to forward those responses
through without losing the trailers.
---
 lib/yahns/proxy_http_response.rb | 81 ++++++++++++++++++++++++++++++----------
 lib/yahns/proxy_pass.rb          |  1 +
 test/test_proxy_pass.rb          | 62 +++++++++++++++++++++++++++++-
 3 files changed, 123 insertions(+), 21 deletions(-)

diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index 8f4790e..fe0a37b 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -120,6 +120,7 @@ module Yahns::HttpResponse # :nodoc:
         end until len == 0
 
       else # nasty chunked body
+        req_res.proxy_trailers = nil # define to avoid warnings for now
 
         # Only HTTP/1.1 supports chunked responses, we must translate
         # otherwise.  Otherwise, we must drop the connection to signal
@@ -130,9 +131,7 @@ module Yahns::HttpResponse # :nodoc:
         when String
           kcar.filter_body(buf, tmp)
           unless buf.empty?
-            tmp = rechunk ? [ "#{buf.size.to_s(16)}\r\n", buf, "\r\n".freeze ]
-                          : buf
-            wbuf = proxy_write(wbuf, tmp, alive)
+            wbuf = proxy_write(wbuf, rechunk ? chunk_out(buf) : buf, alive)
           end
         when nil # premature EOF
           return proxy_err_response(nil, req_res, nil, wbuf)
@@ -142,8 +141,25 @@ module Yahns::HttpResponse # :nodoc:
           return :wait_readable # self remains in :ignore, wait on upstream
         end until kcar.body_eof?
 
-        # TODO: Trailer support
-        wbuf = proxy_write(wbuf, "0\r\n\r\n".freeze, alive) if rechunk
+        buf = tmp
+        req_res.proxy_trailers = [ buf, tlr = [] ]
+        rbuf = Thread.current[:yahns_rbuf] = ''
+        if rechunk
+          until kcar.trailers(tlr, buf)
+            case rv = req_res.kgio_tryread(0x2000, rbuf)
+            when String
+              buf << rv
+            when :wait_readable
+              # for ensure:
+              wbuf ||= Yahns::Wbuf.new(nil, alive, k.output_buffer_tmpdir,
+                                       false)
+              return :wait_readable
+            when nil # premature EOF
+              return proxy_err_response(nil, req_res, nil, wbuf)
+            end # no loop here
+          end
+          wbuf = proxy_write(wbuf, trailer_out(tlr), alive)
+        end
       end
     end
 
@@ -178,22 +194,36 @@ module Yahns::HttpResponse # :nodoc:
       # Rack/Rails can always send streaming responses.
       rechunk = @hs.env['HTTP_VERSION'] == 'HTTP/1.1'.freeze
       buf = ''
-      case tmp = req_res.kgio_tryread(0x2000, rbuf)
-      when String
-        kcar.filter_body(buf, tmp)
-        unless buf.empty?
-          tmp = rechunk ? [ "#{buf.size.to_s(16)}\r\n", buf, "\r\n".freeze ]
-                        : buf
-          wbuf.wbuf_write(self, tmp)
-        end
-      when nil # premature EOF
-        return proxy_err_response(nil, req_res, nil, wbuf)
-      when :wait_readable
-        return :wait_readable # self remains in :ignore, wait on upstream
-      end until kcar.body_eof?
 
-      # TODO: Trailer support
-      wbuf.wbuf_write(self, "0\r\n\r\n".freeze) if rechunk
+      unless req_res.proxy_trailers
+        # are we done dechunking the main body, yet?
+        case tmp = req_res.kgio_tryread(0x2000, rbuf)
+        when String
+          kcar.filter_body(buf, tmp)
+          buf.empty? or wbuf.wbuf_write(self, rechunk ? chunk_out(buf) : buf)
+        when nil # premature EOF
+          return proxy_err_response(nil, req_res, nil, wbuf)
+        when :wait_readable
+          return :wait_readable # self remains in :ignore, wait on upstream
+        end until kcar.body_eof?
+        req_res.proxy_trailers = [ tmp, [] ] # onto trailers!
+        rbuf = Thread.current[:yahns_rbuf] = ''
+      end
+
+      buf, tlr = *req_res.proxy_trailers
+      if rechunk
+        until kcar.trailers(tlr, buf)
+          case rv = req_res.kgio_tryread(0x2000, rbuf)
+          when String
+            buf << rv
+          when :wait_readable
+            return :wait_readable
+          when nil # premature EOF
+            return proxy_err_response(nil, req_res, nil, wbuf)
+          end # no loop here
+        end
+        wbuf.wbuf_write(self, trailer_out(tlr))
+      end
     end
 
     busy = wbuf.busy and return proxy_busy_mod_blocked(wbuf, busy)
@@ -226,4 +256,15 @@ module Yahns::HttpResponse # :nodoc:
     # no touching self after queue_mod
     :ignore
   end
+
+  # n.b.: we can use String#size for optimized dispatch under YARV instead
+  # of String#bytesize because all the IO read methods return a binary
+  # string when given a maximum read length
+  def chunk_out(buf)
+    [ "#{buf.size.to_s(16)}\r\n", buf, "\r\n".freeze ]
+  end
+
+  def trailer_out(tlr)
+    "0\r\n#{tlr.map! do |k,v| "#{k}: #{v}\r\n" end.join}\r\n"
+  end
 end
diff --git a/lib/yahns/proxy_pass.rb b/lib/yahns/proxy_pass.rb
index ec1fbcf..3812dbf 100644
--- a/lib/yahns/proxy_pass.rb
+++ b/lib/yahns/proxy_pass.rb
@@ -12,6 +12,7 @@ require_relative 'proxy_http_response'
 class Yahns::ProxyPass # :nodoc:
   class ReqRes < Kgio::Socket
     attr_writer :resbuf
+    attr_accessor :proxy_trailers
 
     def req_start(c, req, input, chunked)
       @hdr = @resbuf = nil
diff --git a/test/test_proxy_pass.rb b/test/test_proxy_pass.rb
index 47fc231..943fb35 100644
--- a/test/test_proxy_pass.rb
+++ b/test/test_proxy_pass.rb
@@ -53,6 +53,33 @@ class TestProxyPass < Testcase
           io = env['rack.hijack'].call
           io.write(TRUNCATE_HEAD)
           io.close
+        when '/response-trailer'
+          h = [
+            %w(Content-Type text/pain),
+            %w(Transfer-Encoding chunked),
+            %w(Trailer Foo)
+          ]
+          b = [ "3\r\n", "hi\n", "\r\n", "0\r\n", "Foo: bar", "\r\n", "\r\n" ]
+          case env['HTTP_X_TRAILER']
+          when 'fast'
+            b = [ b.join ]
+          when 'allslow'
+            def b.each
+              size.times do |i|
+                sleep 0.1
+                yield self[i]
+              end
+            end
+          when /\Atlrslow(\d+)/
+            b.instance_variable_set(:@yahns_sleep_thresh, $1.to_i)
+            def b.each
+              size.times do |i|
+                sleep(0.1) if i > @yahns_sleep_thresh
+                yield self[i]
+              end
+            end
+          end
+          [ 200, h, b ]
         when '/immediate-EOF'
           env['rack.hijack'].call.close
         when %r{\A/chunky-slow-(\d+(?:\.\d+)?)\z}
@@ -184,7 +211,7 @@ class TestProxyPass < Testcase
       @srv.close
       cfg.instance_eval do
         app(:rack, ProxiedApp.new) do
-          listen "#{host2}:#{port2}"
+          listen "#{host2}:#{port2}", rcvbuf: 4096
           client_max_body_size nil
         end
         stderr_path err.path
@@ -192,6 +219,7 @@ class TestProxyPass < Testcase
     end
 
     check_pipelining(host, port)
+    check_response_trailer(host, port)
 
     gplv3 = File.open('COPYING')
 
@@ -434,4 +462,36 @@ class TestProxyPass < Testcase
   ensure
     s.close if s
   end
+
+  def check_response_trailer(host, port)
+    thrs = [
+      "X-Trailer: fast\r\n",
+      "X-Trailer: allslow\r\n",
+      "X-Trailer: tlrslow1\r\n",
+      "X-Trailer: tlrslow2\r\n",
+      "X-Trailer: tlrslow3\r\n",
+      "X-Trailer: tlrslow4\r\n",
+      ''
+    ].map do |x|
+      Thread.new do
+        s = TCPSocket.new(host, port)
+        s.write "GET /response-trailer HTTP/1.1\r\n#{x}" \
+                "Host: example.com\r\n\r\n"
+        res = ''
+        buf = ''
+        Timeout.timeout(60) do
+          until res =~ /Foo: bar\r\n\r\n\z/
+            res << s.readpartial(16384, buf)
+          end
+        end
+        assert_match(%r{\r\n0\r\nFoo: bar\r\n\r\n\z}, res)
+        assert_match(%r{^Trailer: Foo\r\n}, res)
+        assert_match(%r{^Transfer-Encoding: chunked\r\n}, res)
+        assert_match(%r{\AHTTP/1\.1 200 OK\r\n}, res)
+        s.close
+        :OK
+      end
+    end
+    thrs.each { |t| assert_equal(:OK, t.value) }
+  end
 end
-- 
EW

[PATCH 2/2] proxy_pass: send 1.0 requests to upstreams for 1.0 clients

[Eric Wong]

We cannot pass trailers from upstreams to HTTP/1.0 clients without
fully-buffering the response body AND trailers before forwarding the
response header.

Of course, one of the reasons yahns exists is to support lazy
buffering, so fully-buffering up front is wasteful and hurts
latency.  So instead, degrade to 1.0 requests to upstreams for
HTTP/1.0 clients, this should prevent upstreams from sending
trailers in the first place.

HTTP/1.0 clients on Rails apps may suffer, but there probably
are not too many HTTP/1.0 clients out there.
---
 lib/yahns/proxy_http_response.rb | 79 ++++++++++++----------------------------
 lib/yahns/proxy_pass.rb          |  8 +++-
 test/test_proxy_pass.rb          | 22 ++++++++---
 3 files changed, 47 insertions(+), 62 deletions(-)

diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index fe0a37b..cbfc17e 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -58,7 +58,6 @@ module Yahns::HttpResponse # :nodoc:
     have_body = !Rack::Utils::STATUS_WITH_NO_ENTITY_BODY.include?(si) &&
                 env[REQUEST_METHOD] != HEAD
     flags = MSG_DONTWAIT
-    rechunk = false
     k = self.class
     alive = @hs.next? && k.persistent_connections
 
@@ -69,18 +68,6 @@ module Yahns::HttpResponse # :nodoc:
         next # do not let some upstream headers leak through
       when %r{\AContent-Length\z}i
         flags |= MSG_MORE if have_body && value.to_i > 0
-      when %r{\ATransfer-Encoding\z}i
-        if value =~ /\bchunked\b/i
-          case env['HTTP_VERSION'] # this is the original client request
-          when 'HTTP/1.1'
-            rechunk = true
-          else
-            # too expensive to calculate Content-Length for HTTP/1.0
-            # persistent clients, so we will drop persistence instead
-            alive = false # this should already be implied
-            next
-          end
-        end
       end
 
       res << "#{key}: #{value}\r\n"
@@ -121,18 +108,11 @@ module Yahns::HttpResponse # :nodoc:
 
       else # nasty chunked body
         req_res.proxy_trailers = nil # define to avoid warnings for now
-
-        # Only HTTP/1.1 supports chunked responses, we must translate
-        # otherwise.  Otherwise, we must drop the connection to signal
-        # the end.  We only send HTTP/1.1 requests to the upstream so
-        # Rack/Rails can always send streaming responses.
         buf = ''
         case tmp = tip.shift || req_res.kgio_tryread(0x2000, rbuf)
         when String
           kcar.filter_body(buf, tmp)
-          unless buf.empty?
-            wbuf = proxy_write(wbuf, rechunk ? chunk_out(buf) : buf, alive)
-          end
+          wbuf = proxy_write(wbuf, chunk_out(buf), alive) unless buf.empty?
         when nil # premature EOF
           return proxy_err_response(nil, req_res, nil, wbuf)
         when :wait_readable
@@ -144,22 +124,19 @@ module Yahns::HttpResponse # :nodoc:
         buf = tmp
         req_res.proxy_trailers = [ buf, tlr = [] ]
         rbuf = Thread.current[:yahns_rbuf] = ''
-        if rechunk
-          until kcar.trailers(tlr, buf)
-            case rv = req_res.kgio_tryread(0x2000, rbuf)
-            when String
-              buf << rv
-            when :wait_readable
-              # for ensure:
-              wbuf ||= Yahns::Wbuf.new(nil, alive, k.output_buffer_tmpdir,
-                                       false)
-              return :wait_readable
-            when nil # premature EOF
-              return proxy_err_response(nil, req_res, nil, wbuf)
-            end # no loop here
-          end
-          wbuf = proxy_write(wbuf, trailer_out(tlr), alive)
+        until kcar.trailers(tlr, buf)
+          case rv = req_res.kgio_tryread(0x2000, rbuf)
+          when String
+            buf << rv
+          when :wait_readable
+            # for ensure:
+            wbuf ||= Yahns::Wbuf.new(nil, alive, k.output_buffer_tmpdir, false)
+            return :wait_readable
+          when nil # premature EOF
+            return proxy_err_response(nil, req_res, nil, wbuf)
+          end # no loop here
         end
+        wbuf = proxy_write(wbuf, trailer_out(tlr), alive)
       end
     end
 
@@ -187,12 +164,6 @@ module Yahns::HttpResponse # :nodoc:
       end while len != 0
 
     else # nasty chunked body
-
-      # Only HTTP/1.1 supports chunked responses, we must translate
-      # otherwise.  Otherwise, we must drop the connection to signal
-      # the end.  We only send HTTP/1.1 requests to the upstream so
-      # Rack/Rails can always send streaming responses.
-      rechunk = @hs.env['HTTP_VERSION'] == 'HTTP/1.1'.freeze
       buf = ''
 
       unless req_res.proxy_trailers
@@ -200,7 +171,7 @@ module Yahns::HttpResponse # :nodoc:
         case tmp = req_res.kgio_tryread(0x2000, rbuf)
         when String
           kcar.filter_body(buf, tmp)
-          buf.empty? or wbuf.wbuf_write(self, rechunk ? chunk_out(buf) : buf)
+          buf.empty? or wbuf.wbuf_write(self, chunk_out(buf))
         when nil # premature EOF
           return proxy_err_response(nil, req_res, nil, wbuf)
         when :wait_readable
@@ -211,19 +182,17 @@ module Yahns::HttpResponse # :nodoc:
       end
 
       buf, tlr = *req_res.proxy_trailers
-      if rechunk
-        until kcar.trailers(tlr, buf)
-          case rv = req_res.kgio_tryread(0x2000, rbuf)
-          when String
-            buf << rv
-          when :wait_readable
-            return :wait_readable
-          when nil # premature EOF
-            return proxy_err_response(nil, req_res, nil, wbuf)
-          end # no loop here
-        end
-        wbuf.wbuf_write(self, trailer_out(tlr))
+      until kcar.trailers(tlr, buf)
+        case rv = req_res.kgio_tryread(0x2000, rbuf)
+        when String
+          buf << rv
+        when :wait_readable
+          return :wait_readable
+        when nil # premature EOF
+          return proxy_err_response(nil, req_res, nil, wbuf)
+        end # no loop here
       end
+      wbuf.wbuf_write(self, trailer_out(tlr))
     end
 
     busy = wbuf.busy and return proxy_busy_mod_blocked(wbuf, busy)
diff --git a/lib/yahns/proxy_pass.rb b/lib/yahns/proxy_pass.rb
index 3812dbf..e3ba7f0 100644
--- a/lib/yahns/proxy_pass.rb
+++ b/lib/yahns/proxy_pass.rb
@@ -205,7 +205,13 @@ class Yahns::ProxyPass # :nodoc:
     req = @path.gsub(/\$(\w+)/) { req.__send__($1) }
 
     # start the connection asynchronously and early so TCP can do a
-    req = "#{env['REQUEST_METHOD']} #{req} HTTP/1.1\r\n" \
+    case ver = env['HTTP_VERSION']
+    when 'HTTP/1.1' # leave alone, response may be chunked
+    else # no chunking for HTTP/1.0 and HTTP/0.9
+      ver = 'HTTP/1.0'.freeze
+    end
+
+    req = "#{env['REQUEST_METHOD']} #{req} #{ver}\r\n" \
           "X-Forwarded-For: #{env["REMOTE_ADDR"]}\r\n"
 
     # pass most HTTP_* headers through as-is
diff --git a/test/test_proxy_pass.rb b/test/test_proxy_pass.rb
index 943fb35..c1539f8 100644
--- a/test/test_proxy_pass.rb
+++ b/test/test_proxy_pass.rb
@@ -86,13 +86,23 @@ class TestProxyPass < Testcase
           delay = $1.to_f
           chunky = Object.new
           chunky.instance_variable_set(:@delay, delay)
-          def chunky.each
-            sleep @delay
-            yield "3\r\nHI!\r\n"
-            sleep @delay
-            yield "0\r\n\r\n"
+          if env['HTTP_VERSION'] == 'HTTP/1.0'
+            h = [ %w(Content-Type text/pain), %w(Content-Length 3) ]
+            def chunky.each
+              %w(H I !).each do |x|
+                sleep @delay
+                yield x
+              end
+            end
+          else
+            h = [ %w(Content-Type text/pain), %w(Transfer-Encoding chunked) ]
+            def chunky.each
+              sleep @delay
+              yield "3\r\nHI!\r\n"
+              sleep @delay
+              yield "0\r\n\r\n"
+            end
           end
-          h = [ %w(Content-Type text/pain), %w(Transfer-Encoding chunked) ]
           [ 200, h, chunky ]
         else
           [ 200, h, [ "hi\n"] ]
-- 
EW