input and output buffers support tmpdir: arguments

This allows users to specify alternative temporary directories
in case buffers get too large for one filesystem to handle or
to give priority to some clients on certain ports.

9 files changed, 173 insertions, 19 deletions

diff --git a/Documentation/yahns_config.txt b/Documentation/yahns_config.txt
index 77825c0..2505907 100644
--- a/Documentation/yahns_config.txt
+++ b/Documentation/yahns_config.txt
@@ -258,7 +258,7 @@ Ruby it is running under.
 
     Default: $stderr
 
-* input_buffering {:lazy|true|false}
+* input_buffering {:lazy|true|false}[, OPTIONS]
 
     This controls buffering of the HTTP request body.
 
@@ -280,6 +280,16 @@ Ruby it is running under.
 
     Default: true
 
+    The following OPTIONS may be specified:
+
+      + tmpdir: DIRECTORY
+
+        Specify an alternative temporary directory of input_buffering is
+        :lazy or true.  This can be used in case the normal temporary
+        directory is too small or busy to be used for input buffering.
+
+        Default: Dir.tmpdir (usually from TMPDIR env or /tmp)
+
 * listen ADDRESS [, OPTIONS]
 
     Adds an ADDRESS to the existing listener set.  May be specified more
@@ -404,7 +414,7 @@ Ruby it is running under.
 
     Default: uses the top-level logger
 
-* output_buffering BOOLEAN
+* output_buffering BOOLEAN [, OPTIONS]
 
     This enables or disables buffering of the HTTP response.  If enabled,
     buffering is only performed lazily.  In other words, buffering only
@@ -419,6 +429,16 @@ Ruby it is running under.
 
     Default: true
 
+    The following OPTIONS may be specified:
+
+      + tmpdir: DIRECTORY
+
+        Specify an alternative temporary directory of output_buffering is
+        enabled.  This can be used in case the normal temporary directory
+        is too small or busy to be used for output buffering.
+
+        Default: Dir.tmpdir (usually from TMPDIR env or /tmp)
+
 * persistent_connections BOOLEAN
 
     Enable or disables persistent connections and pipelining for HTTP
diff --git a/lib/yahns/cap_input.rb b/lib/yahns/cap_input.rb
index 1aa10b6..313b3ce 100644
--- a/lib/yahns/cap_input.rb
+++ b/lib/yahns/cap_input.rb
@@ -7,8 +7,8 @@
 class Yahns::CapInput < Yahns::TmpIO # :nodoc:
   attr_writer :bytes_left
 
-  def self.new(limit)
-    rv = super()
+  def self.new(limit, tmpdir)
+    rv = super(tmpdir)
     rv.bytes_left = limit
     rv
   end
diff --git a/lib/yahns/config.rb b/lib/yahns/config.rb
index 1862eee..4ea51af 100644
--- a/lib/yahns/config.rb
+++ b/lib/yahns/config.rb
@@ -334,9 +334,7 @@ class Yahns::Config # :nodoc:
   end
 
   # boolean config directives for app
-  %w(check_client_connection
-     output_buffering
-     persistent_connections).each do |_v|
+  %w(check_client_connection persistent_connections).each do |_v|
     eval(
     %Q(def #{_v}(bool);) <<
     %Q(  _check_in_block(:app, :#{_v});) <<
@@ -345,6 +343,21 @@ class Yahns::Config # :nodoc:
     )
   end
 
+  def output_buffering(bool, opts = {})
+    var = _check_in_block(:app, :output_buffering)
+    @block.ctx.__send__("#{var}=", _check_bool(var, bool))
+    tmpdir = opts[:tmpdir] and
+      @block.ctx.output_buffer_tmpdir = _check_tmpdir(var, tmpdir)
+  end
+
+  def _check_tmpdir(var, path)
+    File.directory?(path) or
+      raise ArgumentError, "#{var} tmpdir: #{path} is not a directory"
+    File.writable?(path) or
+      raise ArgumentError, "#{var} tmpdir: #{path} is not writable"
+    path
+  end
+
   # integer config directives for app
   {
     # config name, minimum value
@@ -371,12 +384,14 @@ class Yahns::Config # :nodoc:
     @block.ctx.__send__("#{var}=", val)
   end
 
-  def input_buffering(val)
+  def input_buffering(val, opts = {})
     var = _check_in_block(:app, :input_buffering)
     ok = [ :lazy, true, false ]
     ok.include?(val) or
       raise ArgumentError, "`#{var}' must be one of: #{ok.inspect}"
     @block.ctx.__send__("#{var}=", val)
+    tmpdir = opts[:tmpdir] and
+      @block.ctx.input_buffer_tmpdir = _check_tmpdir(var, tmpdir)
   end
 
   # used to configure rack.errors destination
diff --git a/lib/yahns/http_context.rb b/lib/yahns/http_context.rb
index 605989f..bd455bd 100644
--- a/lib/yahns/http_context.rb
+++ b/lib/yahns/http_context.rb
@@ -18,6 +18,8 @@ module Yahns::HttpContext # :nodoc:
   attr_accessor :queue # set right before spawning acceptors
   attr_reader :app
   attr_reader :app_defaults
+  attr_writer :input_buffer_tmpdir
+  attr_writer :output_buffer_tmpdir
 
   def http_ctx_init(yahns_rack)
     @yahns_rack = yahns_rack
@@ -32,6 +34,10 @@ module Yahns::HttpContext # :nodoc:
     @client_timeout = 15
     @qegg = nil
     @queue = nil
+
+    # Dir.tmpdir can change while running, so leave these as nil
+    @input_buffer_tmpdir = nil
+    @output_buffer_tmpdir = nil
   end
 
   # call this after forking
@@ -74,10 +80,20 @@ module Yahns::HttpContext # :nodoc:
 
   def tmpio_for(len)
     if len # Content-Length given
-      len <= @client_body_buffer_size ? StringIO.new("") : Yahns::TmpIO.new
+      len <= @client_body_buffer_size ? StringIO.new("")
+           : Yahns::TmpIO.new(input_buffer_tmpdir)
     else # chunked, unknown length
       mbs = @client_max_body_size
-      mbs ? Yahns::CapInput.new(mbs) : Yahns::TmpIO.new
+      tmpdir = input_buffer_tmpdir
+      mbs ? Yahns::CapInput.new(mbs, tmpdir) : Yahns::TmpIO.new(tmpdir)
     end
   end
+
+  def input_buffer_tmpdir
+    @input_buffer_tmpdir || Dir.tmpdir
+  end
+
+  def output_buffer_tmpdir
+    @output_buffer_tmpdir || Dir.tmpdir
+  end
 end
diff --git a/lib/yahns/http_response.rb b/lib/yahns/http_response.rb
index 6ddb1c8..f48b4d8 100644
--- a/lib/yahns/http_response.rb
+++ b/lib/yahns/http_response.rb
@@ -47,7 +47,7 @@ module Yahns::HttpResponse # :nodoc:
       alive = Yahns::StreamFile.new(body, alive, offset, count)
       body = nil
     end
-    wbuf = Yahns::Wbuf.new(body, alive)
+    wbuf = Yahns::Wbuf.new(body, alive, self.class.output_buffer_tmpdir)
     rv = wbuf.wbuf_write(self, header)
     body.each { |chunk| rv = wbuf.wbuf_write(self, chunk) } if body
     wbuf_maybe(wbuf, rv, alive)
@@ -165,7 +165,7 @@ module Yahns::HttpResponse # :nodoc:
           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)
+            wbuf = Yahns::Wbuf.new(body, alive, k.output_buffer_tmpdir)
             rv = wbuf.wbuf_write(self, chunk)
             break
           else
diff --git a/lib/yahns/tmpio.rb b/lib/yahns/tmpio.rb
index 4fe4794..bcf6b8a 100644
--- a/lib/yahns/tmpio.rb
+++ b/lib/yahns/tmpio.rb
@@ -11,10 +11,10 @@ class Yahns::TmpIO < File # :nodoc:
   # creates and returns a new File object.  The File is unlinked
   # immediately, switched to binary mode, and userspace output
   # buffering is disabled
-  def self.new
+  def self.new(tmpdir = Dir.tmpdir)
     retried = false
     begin
-      fp = super("#{Dir.tmpdir}/#{rand}", RDWR|CREAT|EXCL, 0600)
+      fp = super("#{tmpdir}/#{rand}", RDWR|CREAT|EXCL, 0600)
     rescue Errno::EEXIST
       retry
     rescue Errno::EMFILE, Errno::ENFILE
diff --git a/lib/yahns/wbuf.rb b/lib/yahns/wbuf.rb
index 8b2b82e..e6039b7 100644
--- a/lib/yahns/wbuf.rb
+++ b/lib/yahns/wbuf.rb
@@ -6,8 +6,8 @@ require_relative 'wbuf_common'
 class Yahns::Wbuf # :nodoc:
   include Yahns::WbufCommon
 
-  def initialize(body, persist)
-    @tmpio = Yahns::TmpIO.new
+  def initialize(body, persist, tmpdir)
+    @tmpio = Yahns::TmpIO.new(tmpdir)
     @sf_offset = @sf_count = 0
     @wbuf_persist = persist # whether or not we keep the connection alive
     @body = body
diff --git a/test/test_buffer_tmpdir.rb b/test/test_buffer_tmpdir.rb
new file mode 100644
index 0000000..c7665f6
--- /dev/null
+++ b/test/test_buffer_tmpdir.rb
@@ -0,0 +1,103 @@
+# Copyright (C) 2013, Eric Wong <normalperson@yhbt.net> and all contributors
+# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
+require_relative 'server_helper'
+require 'sleepy_penguin'
+
+class TestBufferTmpdir < Testcase
+  ENV["N"].to_i > 1 and parallelize_me!
+  include ServerHelper
+  attr_reader :ino, :tmpdir
+
+  def setup
+    @ino = SleepyPenguin::Inotify.new(:CLOEXEC)
+    @tmpdir = Dir.mktmpdir
+    server_helper_setup
+  end
+
+  def teardown
+    server_helper_teardown
+    @ino.close
+    FileUtils.rm_rf @tmpdir
+  end
+
+  class GiantBody
+    # just spew until the client gives up
+    def each
+      nr = 16384
+      buf = "#{nr.to_s(16)}\r\n#{("!" * nr)}\r\n"
+      loop do
+        yield buf
+      end
+    end
+  end
+
+  def test_output_buffer_tmpdir
+    opts = { tmpdir: @tmpdir }
+    err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
+    pid = mkserver(cfg) do
+      cfg.instance_eval do
+        ru = lambda { |e|
+          h = {
+            "Transfer-Encoding" => "chunked",
+            "Content-Type" => "text/plain"
+          }
+          [ 200, h, GiantBody.new ]
+        }
+        app(:rack, ru) do
+          listen "#{host}:#{port}"
+          output_buffering true, opts
+        end
+        stderr_path err.path
+      end
+    end
+    @ino.add_watch @tmpdir, [:CREATE, :DELETE]
+    c = get_tcp_client(host, port)
+    c.write "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
+    Timeout.timeout(30) do
+      event = @ino.take
+      assert_equal [:CREATE], event.events
+      name = event.name
+      event = @ino.take
+      assert_equal [:DELETE], event.events
+      assert_equal name, event.name
+      refute File.exist?("#@tmpdir/#{name}")
+    end
+  ensure
+    c.close if c
+    quit_wait(pid)
+  end
+
+  def test_input_buffer_lazy; input_buffer(:lazy); end
+  def test_input_buffer_true; input_buffer(true); end
+
+  def input_buffer(btype)
+    opts = { tmpdir: @tmpdir }
+    err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
+    pid = mkserver(cfg) do
+      cfg.instance_eval do
+        require 'rack/lobster'
+        app(:rack, Rack::Lobster.new) do
+          listen "#{host}:#{port}"
+          input_buffering btype, opts
+        end
+        stderr_path err.path
+      end
+    end
+    @ino.add_watch tmpdir, [:CREATE, :DELETE]
+    c = get_tcp_client(host, port)
+    nr = 16384 # must be > client_body_buffer_size
+    c.write "POST / HTTP/1.0\r\nContent-Length: #{nr}\r\n\r\n"
+    Timeout.timeout(30) do
+      event = ino.take
+      assert_equal [:CREATE], event.events
+      name = event.name
+      event = ino.take
+      assert_equal [:DELETE], event.events
+      assert_equal name, event.name
+      refute File.exist?("#{tmpdir}/#{name}")
+    end
+  ensure
+    c.close if c
+    quit_wait(pid)
+  end
+end if SleepyPenguin.const_defined?(:Inotify)
diff --git a/test/test_wbuf.rb b/test/test_wbuf.rb
index 03abbac..a9dc717 100644
--- a/test/test_wbuf.rb
+++ b/test/test_wbuf.rb
@@ -18,7 +18,7 @@ class TestWbuf < Testcase
     buf = "*" * (16384 * 2)
     nr = 1000
     [ true, false ].each do |persist|
-      wbuf = Yahns::Wbuf.new([], persist)
+      wbuf = Yahns::Wbuf.new([], persist, Dir.tmpdir)
       a, b = socketpair
       assert_nil wbuf.wbuf_write(a, "HIHI")
       assert_equal "HIHI", b.read(4)
@@ -67,7 +67,7 @@ class TestWbuf < Testcase
         break
       end while true
     end
-    wbuf = Yahns::Wbuf.new([], true)
+    wbuf = Yahns::Wbuf.new([], true, Dir.tmpdir)
     assert_equal :wait_writable, wbuf.wbuf_write(a, buf)
     assert_equal :wait_writable, wbuf.wbuf_flush(a)
 
@@ -96,7 +96,7 @@ class TestWbuf < Testcase
   def test_wbuf_flush_close
     pipe = cloexec_pipe
     persist = true
-    wbuf = Yahns::Wbuf.new(pipe[0], persist)
+    wbuf = Yahns::Wbuf.new(pipe[0], persist, Dir.tmpdir)
     refute wbuf.respond_to?(:close) # we don't want this for HttpResponse body
     sp = socketpair
     rv = nil