[PATCH 1/4] wbuf: drop persistence if writing to client fails

[PATCH 1/4] wbuf: drop persistence if writing to client fails

[Eric Wong]

We cannot maintain a persistent connection to a client if
writing to the client fails.  This should fix FD leakage on
hijacked responses
---
 lib/yahns/wbuf_common.rb | 1 +
 1 file changed, 1 insertion(+)

diff --git a/lib/yahns/wbuf_common.rb b/lib/yahns/wbuf_common.rb
index c51050b..dc653ff 100644
--- a/lib/yahns/wbuf_common.rb
+++ b/lib/yahns/wbuf_common.rb
@@ -38,6 +38,7 @@ def wbuf_flush(client)
     end while @sf_count > 0
     wbuf_close(client)
   rescue
+    @wbuf_persist = false
     wbuf_close(client)
     raise
   end
-- 
EW

[PATCH 2/4] proxy_pass: ensure we do not persist connections on failure

[Eric Wong]

---
 lib/yahns/proxy_http_response.rb | 9 +++++++++
 lib/yahns/proxy_pass.rb          | 3 ++-
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index c8a2a42..7320557 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -318,4 +318,13 @@ def chunk_out(buf)
   def trailer_out(tlr)
     "0\r\n#{tlr.map! do |k,v| "#{k}: #{v}\r\n" end.join}\r\n"
   end
+
+  def proxy_shutdown
+    to_io.shutdown(Socket::SHUT_RD)
+
+    # hack to ensure @hs.next? returns false the next time it is called:
+    hs = Unicorn::HttpRequest.new
+    hs.env.merge!(@hs.env)
+    @hs = hs
+  end
 end
diff --git a/lib/yahns/proxy_pass.rb b/lib/yahns/proxy_pass.rb
index 511db02..7ef865a 100644
--- a/lib/yahns/proxy_pass.rb
+++ b/lib/yahns/proxy_pass.rb
@@ -141,9 +141,10 @@ def send_req_body(req)
         prepare_wait_readable
       end
     rescue Errno::EPIPE, Errno::ECONNRESET, Errno::ENOTCONN
+      # writing to the upstream got us shutdown by the upstream...
       # no more reading off the client socket, just prepare to forward
       # the rejection response from the upstream (if any)
-      @yahns_client.to_io.shutdown(Socket::SHUT_RD)
+      @yahns_client.proxy_shutdown
       prepare_wait_readable
     end
 
-- 
EW

[PATCH 3/4] fix blocking

[Eric Wong]

---
 lib/yahns/proxy_http_response.rb | 21 ++++++++++-----------
 1 file changed, 10 insertions(+), 11 deletions(-)

diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index 7320557..ae7bd72 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -179,7 +179,6 @@ def proxy_response_start(res, tip, kcar, req_res)
     end
 
     return proxy_busy_mod_done(alive) unless wbuf
-    req_res.resbuf = wbuf
     proxy_busy_mod_blocked(wbuf, wbuf.busy)
   rescue => e
     proxy_err_response(502, req_res, e, wbuf)
@@ -289,23 +288,23 @@ def proxy_busy_mod_done(alive)
     when :close then close
     end
 
-    nil # close the req_res, too
+    nil # signal close for ReqRes#yahns_step
   end
 
   def proxy_busy_mod_blocked(wbuf, busy)
-    q = Thread.current[:yahns_queue]
     # we are completely done reading and buffering the upstream response,
     # but have not completely written the response to the client,
     # yield control to the client socket:
     @state = wbuf
-    case busy
-    when :wait_readable then q.queue_mod(self, Yahns::Queue::QEV_RD)
-    when :wait_writable then q.queue_mod(self, Yahns::Queue::QEV_WR)
-    else
-      abort "BUG: invalid wbuf.busy: #{busy.inspect}"
-    end
-    # no touching self after queue_mod
-    :ignore
+    proxy_wait_next(case busy
+      when :wait_readable then Yahns::Queue::QEV_RD
+      when :wait_writable then Yahns::Queue::QEV_WR
+      else
+        abort "BUG: invalid wbuf.busy: #{busy.inspect}"
+      end)
+    # no touching self after proxy_wait_next, we may be running
+    # HttpClient#yahns_step in a different thread at this point
+    nil # signal close for ReqRes#yahns_step
   end
 
   # n.b.: we can use String#size for optimized dispatch under YARV instead
-- 
EW

[PATCH 4/4] proxy_http_response

[Eric Wong]

---
 lib/yahns/proxy_http_response.rb | 3 ++-
 lib/yahns/proxy_pass.rb          | 4 ++--
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index ae7bd72..2be5190 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -43,7 +43,8 @@ def proxy_err_response(code, req_res, exc, wbuf)
                      Rack::Utils::HTTP_STATUS_CODES[code]}\r\n\r\n") rescue nil
 
     shutdown rescue nil
-    req_res.shutdown rescue nil
+    @input = @input.close if @input
+    close
     nil # signal close of req_res from yahns_step in yahns/proxy_pass.rb
   ensure
     wbuf.wbuf_abort if wbuf
diff --git a/lib/yahns/proxy_pass.rb b/lib/yahns/proxy_pass.rb
index 7ef865a..2dd0a59 100644
--- a/lib/yahns/proxy_pass.rb
+++ b/lib/yahns/proxy_pass.rb
@@ -63,7 +63,7 @@ def yahns_step # yahns event loop entry point
 
         when Yahns::WbufCommon # streaming/buffering the response body
 
-          return c.proxy_response_finish(req, resbuf, self)
+          return c.proxy_response_finish(req, wbuf = resbuf, self)
 
         end while true # case @resbuf
 
@@ -79,7 +79,7 @@ def yahns_step # yahns event loop entry point
       when Errno::ECONNREFUSED, Errno::ECONNRESET, Errno::EPIPE
         e.set_backtrace([])
       end
-      c.proxy_err_response(502, self, e, nil)
+      c.proxy_err_response(502, self, e, wbuf)
     end
 
     # returns :wait_readable if complete, :wait_writable if not
-- 
EW