[YUCK 0/2] wbuf_lite: fix write retries for OpenSSL

[YUCK 0/2] wbuf_lite: fix write retries for OpenSSL

[Eric Wong]

Removing the tmpdir parameter for all Wbufs is a nice change,
but the actual bug fix is nasty and gross :<  Oh well.

I suppose this increases my impetus for removing kgio from this
project entirely and improving the visibility of buffering.

And all the sendfile stuff is worthless for HTTPS users; but
it's still useful on private LANs and within VPNs (or hosting
Tor hidden services).

Eric Wong (2):
      wbuf: remove tmpdir parameter
      wbuf_lite: fix write retries for OpenSSL sockets

 lib/yahns/http_response.rb           |  4 ++--
 lib/yahns/proxy_http_response.rb     |  2 +-
 lib/yahns/wbuf.rb                    |  5 ++--
 lib/yahns/wbuf_lite.rb               | 46 ++++++++++++++++--------------------
 test/test_proxy_pass_no_buffering.rb |  7 +++++-
 test/test_wbuf.rb                    |  9 ++++---
 6 files changed, 37 insertions(+), 36 deletions(-)

[PATCH 1/2] wbuf: remove tmpdir parameter

[Eric Wong]

We can retrieve it when we actually need to create the
temporary file.  This saves an ivar slot and method dispatch
parameters.

This patch is nice, unfortunately the patch which follows is
not :P
---
 lib/yahns/http_response.rb       | 4 ++--
 lib/yahns/proxy_http_response.rb | 2 +-
 lib/yahns/wbuf.rb                | 5 ++---
 test/test_wbuf.rb                | 9 ++++++---
 4 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/lib/yahns/http_response.rb b/lib/yahns/http_response.rb
index 531194f..d957df6 100644
--- a/lib/yahns/http_response.rb
+++ b/lib/yahns/http_response.rb
@@ -62,7 +62,7 @@ def response_header_blocked(header, body, alive, offset, count)
       alive = Yahns::StreamFile.new(body, alive, offset, count)
       body = nil
     end
-    wbuf = Yahns::Wbuf.new(body, alive, self.class.output_buffer_tmpdir)
+    wbuf = Yahns::Wbuf.new(body, alive)
     rv = wbuf.wbuf_write(self, header)
     if body && ! alive.respond_to?(:call) # skip body.each if hijacked
       body.each { |chunk| rv = wbuf.wbuf_write(self, chunk) }
@@ -199,7 +199,7 @@ def http_response_write(status, headers, body)
           chunk = rv # hope the skb grows when we loop into the trywrite
         when :wait_writable, :wait_readable
           if k.output_buffering
-            wbuf = Yahns::Wbuf.new(body, alive, k.output_buffer_tmpdir)
+            wbuf = Yahns::Wbuf.new(body, alive)
             rv = wbuf.wbuf_write(self, chunk)
             break
           else
diff --git a/lib/yahns/proxy_http_response.rb b/lib/yahns/proxy_http_response.rb
index 61f1539..8de5b4f 100644
--- a/lib/yahns/proxy_http_response.rb
+++ b/lib/yahns/proxy_http_response.rb
@@ -21,7 +21,7 @@ def proxy_unbuffer(wbuf)
 
   def wbuf_alloc(req_res)
     if req_res.proxy_pass.proxy_buffering
-      Yahns::Wbuf.new(nil, req_res.alive, self.class.output_buffer_tmpdir)
+      Yahns::Wbuf.new(nil, req_res.alive)
     else
       Yahns::WbufLite.new(req_res)
     end
diff --git a/lib/yahns/wbuf.rb b/lib/yahns/wbuf.rb
index 1010c86..583df10 100644
--- a/lib/yahns/wbuf.rb
+++ b/lib/yahns/wbuf.rb
@@ -32,9 +32,8 @@ class Yahns::Wbuf # :nodoc:
   include Yahns::WbufCommon
   attr_reader :busy
 
-  def initialize(body, persist, tmpdir)
+  def initialize(body, persist)
     @tmpio = nil
-    @tmpdir = tmpdir
     @sf_offset = @sf_count = 0
     @wbuf_persist = persist # whether or not we keep the connection alive
     @body = body # something we call #close on when done writing
@@ -58,7 +57,7 @@ def wbuf_write(c, buf)
       @busy = rv
     end until @busy
 
-    @tmpio ||= Yahns::TmpIO.new(@tmpdir)
+    @tmpio ||= Yahns::TmpIO.new(c.class.output_buffer_tmpdir)
     @sf_count += String === buf ? @tmpio.write(buf) : wbuf_writev(buf)
 
     # we spent some time copying to the FS, try to write to
diff --git a/test/test_wbuf.rb b/test/test_wbuf.rb
index 990ad9d..1382086 100644
--- a/test/test_wbuf.rb
+++ b/test/test_wbuf.rb
@@ -9,6 +9,9 @@ class TestWbuf < Testcase
 
   class KgioUS < UNIXSocket
     include Kgio::SocketMethods
+    def self.output_buffer_tmpdir
+      Dir.tmpdir
+    end
   end
 
   def socketpair
@@ -20,7 +23,7 @@ def test_wbuf
     buf = "*" * (16384 * 2)
     nr = 1000
     [ true, false ].each do |persist|
-      wbuf = Yahns::Wbuf.new([], persist, Dir.tmpdir)
+      wbuf = Yahns::Wbuf.new([], persist)
       assert_equal false, wbuf.busy
       a, b = socketpair
       assert_nil wbuf.wbuf_write(a, "HIHI")
@@ -71,7 +74,7 @@ def test_wbuf_blocked
         break
       end while true
     end
-    wbuf = Yahns::Wbuf.new([], true, Dir.tmpdir)
+    wbuf = Yahns::Wbuf.new([], true)
 
     rv1 = wbuf.wbuf_write(a, buf)
     rv2 = wbuf.wbuf_flush(a)
@@ -104,7 +107,7 @@ def test_wbuf_blocked
   def test_wbuf_flush_close
     pipe = cloexec_pipe
     persist = true
-    wbuf = Yahns::Wbuf.new(pipe[0], persist, Dir.tmpdir)
+    wbuf = Yahns::Wbuf.new(pipe[0], persist)
     refute wbuf.respond_to?(:close) # we don't want this for HttpResponse body
     sp = socketpair
     rv = nil

[PATCH 2/2] wbuf_lite: fix write retries for OpenSSL sockets

[Eric Wong]

OpenSSL can't handle write retries if we append to an existing
string.  Thus we must preserve the same string object upon
retrying.

Do that by utilizing the underlying Wbuf class which could
already handles it transparently using trysendfile.
However, we still avoiding the subtlety of wbuf_close_common
reliance we previously used.

ref: commit 551e670281bea77e727a732ba94275265ccae5f6
("fix output buffering with SSL_write")
---
 lib/yahns/wbuf_lite.rb               | 46 ++++++++++++++++--------------------
 test/test_proxy_pass_no_buffering.rb |  7 +++++-
 2 files changed, 26 insertions(+), 27 deletions(-)

diff --git a/lib/yahns/wbuf_lite.rb b/lib/yahns/wbuf_lite.rb
index 5f25b2e..afee1e9 100644
--- a/lib/yahns/wbuf_lite.rb
+++ b/lib/yahns/wbuf_lite.rb
@@ -2,49 +2,43 @@
 # Copyright (C) 2016 all contributors <yahns-public@yhbt.net>
 # License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
 # frozen_string_literal: true
-require_relative 'wbuf_common'
+require_relative 'wbuf'
 
 # This is only used for "proxy_buffering: false"
-class Yahns::WbufLite # :nodoc:
-  include Yahns::WbufCommon
+class Yahns::WbufLite < Yahns::Wbuf # :nodoc:
   attr_reader :busy
 
   def initialize(req_res)
+    super(nil, :ignore)
     @req_res = req_res
-    @busy = false
-    @buf = nil
   end
 
-  # called only by proxy_write in proxy_http_response
   def wbuf_write(client, buf)
-    buf = buf.join if Array === buf
-    buf = @buf << buf if @buf # unlikely
-    do_write(client, buf) # try our best to avoid string copying/appending
+    super
+  rescue
+    @req_res = @req_res.close if @req_res
+    raise
   end
 
-  def do_write(client, buf)
-    case rv = client.kgio_trywrite(buf)
-    when String
-      buf = rv # continue looping
-    when :wait_writable, :wait_readable
-      @buf = buf
-      return @busy = rv
-    when nil
-      @buf = nil
-      return @busy = false
-    end while true
+  def wbuf_flush(client)
+    super
+  rescue
+    @req_res = @req_res.close if @req_res
+    raise
   end
 
   # called by Yahns::HttpClient#step_write
-  def wbuf_flush(client)
-    sym = do_write(client, @buf) and return sym # :wait_writable/:wait_readable
+  def wbuf_close(client)
+    wbuf_abort
 
-    # resume reading
-    client.hijack_cleanup
-    Thread.current[:yahns_queue].queue_mod(@req_res, Yahns::Queue::QEV_RD)
+    # resume reading when @blocked is empty
+    if @req_res
+      client.hijack_cleanup
+      Thread.current[:yahns_queue].queue_mod(@req_res, Yahns::Queue::QEV_RD)
+    end
     :ignore
   rescue
-    @req_res.close
+    @req_res = @req_res.close if @req_res
     raise
   end
 end
diff --git a/test/test_proxy_pass_no_buffering.rb b/test/test_proxy_pass_no_buffering.rb
index 48b8241..b2c8b48 100644
--- a/test/test_proxy_pass_no_buffering.rb
+++ b/test/test_proxy_pass_no_buffering.rb
@@ -79,6 +79,7 @@ def test_proxy_pass_no_buffering
     req = "GET /giant-body HTTP/1.1\r\nHost: example.com\r\n" \
           "Connection: close\r\n\r\n"
     s.write(req)
+    bufs = []
     sleep 1
     10.times do
       sleep 0.1
@@ -91,10 +92,14 @@ def test_proxy_pass_no_buffering
         [ deleted1, deleted2 ].each do |ary|
           ary.delete_if { |x| x =~ /\.(?:err|out) \(deleted\)/ }
         end
-        assert_equal 0, deleted1.size, "pid1=#{deleted1.inspect}"
+        assert_equal 1, deleted1.size, "pid1=#{deleted1.inspect}"
         assert_equal 0, deleted2.size, "pid2=#{deleted2.inspect}"
+        bufs.push(deleted1[0])
       end
     end
+    before = bufs.size
+    bufs.uniq!
+    assert bufs.size < before, 'unlinked buffer should not grow'
     buf = ''.dup
     slow = Digest::MD5.new
     ft = Thread.new do