Merge remote-tracking branch 'origin/freebsd'

* origin/freebsd:
  implement kqueue and sendfile compatibility support
  test: log skipped tests on non-Linux systems
  test_server: check_client_connection uses Unix sockets
  test_client_expire: add delays for non-Linux OSes
  wbuf_common: avoid trysendfile on empty sf_count
  sigevent_pipe: kgio_wait_writable takes variadic args
  queue_quitter_pipe: do not prematurely close reader

12 files changed, 172 insertions, 19 deletions

diff --git a/INSTALL b/INSTALL
index bd269f2..87acc53 100644
--- a/INSTALL
+++ b/INSTALL
@@ -6,3 +6,10 @@ No tarballs are currently provided.
 
 You may also install yahns from the git source, see the HACKING document for
 more details.
+
+Debian GNU/kFreeBSD:
+
+For now, you will need to define SENDFILE_BROKEN=1 in the env before
+running yahns, and also install the "io-extra" RubyGem (tested on 1.2.7)
+
+        gem install -v 1.2.7 io-extra
diff --git a/lib/yahns.rb b/lib/yahns.rb
index 5ffb845..5e5875a 100644
--- a/lib/yahns.rb
+++ b/lib/yahns.rb
@@ -57,7 +57,7 @@ end
 
 # FIXME: require lazily
 require_relative 'yahns/log'
-require_relative 'yahns/queue_epoll'
+require_relative 'yahns/queue'
 require_relative 'yahns/stream_input'
 require_relative 'yahns/tee_input'
 require_relative 'yahns/queue_egg'
diff --git a/lib/yahns/queue_kqueue.rb b/lib/yahns/queue_kqueue.rb
new file mode 100644
index 0000000..c502de0
--- /dev/null
+++ b/lib/yahns/queue_kqueue.rb
@@ -0,0 +1,82 @@
+# -*- encoding: binary -*-
+# Copyright (C) 2014, Eric Wong <normalperson@yhbt.net> and all contributors
+# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
+#
+# This is the dangerous, low-level kqueue interface for sleepy_penguin
+# It is safe as long as you're aware of all potential concurrency
+# issues given multithreading, GC, and kqueue itself.
+class Yahns::Queue < SleepyPenguin::Kqueue::IO # :nodoc:
+  include SleepyPenguin
+  attr_accessor :fdmap # Yahns::Fdmap
+
+  # public
+  QEV_QUIT = nil # Level Trigger for QueueQuitter
+  QEV_RD = EvFilt::READ
+  QEV_WR = EvFilt::WRITE
+
+  ADD_ONESHOT = Ev::ADD | Ev::ONESHOT # private
+
+  def self.new
+    rv = super
+    rv.close_on_exec = true
+    rv
+  end
+
+  # for HTTP and HTTPS servers, we rely on the io writing to us, first
+  # flags: QEV_RD/QEV_WR (usually QEV_RD)
+  def queue_add(io, flags)
+    # order is very important here, this thread cannot do anything with
+    # io once we've issued epoll_ctl() because another thread may use it
+    @fdmap.add(io)
+    fflags = ADD_ONESHOT
+    if flags == QEV_QUIT
+      fflags = Ev::ADD
+      flags = QEV_WR
+    end
+    kevent(Kevent[io.fileno, flags, fflags, 0, 0, io])
+  end
+
+  def thr_init
+    Thread.current[:yahns_rbuf] = ""
+    Thread.current[:yahns_fdmap] = @fdmap
+  end
+
+  def queue_del(io)
+    # do not bother with kevent EV_DELETE, it may be tricky to get right,
+    # we only did it in epoll since Eric knows the epoll internals well.
+    @fdmap.forget(io)
+  end
+
+  # returns an array of infinitely running threads
+  def worker_thread(logger, max_events)
+    Thread.new do
+      thr_init
+      begin
+        kevent(nil, max_events) do |_,_,_,_,_,io| # don't care for flags for now
+          # Note: we absolutely must not do anything with io after
+          # we've called epoll_ctl on it, io is exclusive to this
+          # thread only until epoll_ctl is called on it.
+          case rv = io.yahns_step
+          when :wait_readable
+            kevent(Kevent[io.fileno, QEV_RD, ADD_ONESHOT, 0, 0, io])
+          when :wait_writable
+            kevent(Kevent[io.fileno, QEV_WR, ADD_ONESHOT, 0, 0, io])
+          when :ignore # only used by rack.hijack
+            # we cannot EV_DELETE after hijacking, the hijacker
+            # may have already closed it  Likewise, io.fileno is not
+            # expected to work, so we had to erase it from fdmap before hijack
+          when nil, :close
+            # this must be the ONLY place where we call IO#close on
+            # things that got inside the queue
+            @fdmap.sync_close(io)
+          else
+            raise "BUG: #{io.inspect}#yahns_step returned: #{rv.inspect}"
+          end
+        end
+      rescue => e
+        break if closed? # can still happen due to shutdown_timeout
+        Yahns::Log.exception(logger, 'queue loop', e)
+      end while true
+    end
+  end
+end
diff --git a/lib/yahns/queue_quitter_pipe.rb b/lib/yahns/queue_quitter_pipe.rb
index 1aa0643..e18e249 100644
--- a/lib/yahns/queue_quitter_pipe.rb
+++ b/lib/yahns/queue_quitter_pipe.rb
@@ -5,9 +5,8 @@
 class Yahns::QueueQuitter # :nodoc:
   attr_reader :to_io
   def initialize
-    reader, @to_io = IO.pipe
+    @reader, @to_io = IO.pipe
     @to_io.close_on_exec = true
-    reader.close
   end
 
   def yahns_step
@@ -19,6 +18,7 @@ class Yahns::QueueQuitter # :nodoc:
   end
 
   def close
+    @reader.close
     @to_io.close
   end
 end
diff --git a/lib/yahns/sendfile_compat.rb b/lib/yahns/sendfile_compat.rb
new file mode 100644
index 0000000..e3f53d1
--- /dev/null
+++ b/lib/yahns/sendfile_compat.rb
@@ -0,0 +1,29 @@
+# -*- encoding: binary -*-
+# Copyright (C) 2009-2014, Eric Wong <normalperson@yhbt.net> et. al.
+# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
+require 'io/extra' # gem install io-extra
+
+module Yahns::SendfileCompat
+  def trysendfile(io, offset, count)
+    return 0 if count == 0
+    count = 0x4000 if count > 0x4000
+    str = IO.pread(io.fileno, count, offset)
+    if count > str.bytesize
+      raise EOFError, "end of file reached"
+    end
+    n = 0
+    case rv = kgio_trywrite(str)
+    when String # partial write, keep trying
+      n += (str.bytesize - rv.bytesize)
+      str = rv
+    when :wait_writable, :wait_readable
+      return n > 0 ? n : rv
+    when nil
+      return n + str.bytesize # yay!
+    end while true
+  end
+end
+
+class IO
+  include Yahns::SendfileCompat
+end
diff --git a/lib/yahns/sigevent_pipe.rb b/lib/yahns/sigevent_pipe.rb
index 8bd083f..ee0f14e 100644
--- a/lib/yahns/sigevent_pipe.rb
+++ b/lib/yahns/sigevent_pipe.rb
@@ -8,8 +8,8 @@ class Yahns::Sigevent # :nodoc:
     @to_io.close_on_exec = @wr.close_on_exec = true
   end
 
-  def kgio_wait_readable
-    @to_io.kgio_wait_readable
+  def kgio_wait_readable(*args)
+    @to_io.kgio_wait_readable(*args)
   end
 
   def sev_signal
diff --git a/lib/yahns/wbuf_common.rb b/lib/yahns/wbuf_common.rb
index b3a4502..01e20bb 100644
--- a/lib/yahns/wbuf_common.rb
+++ b/lib/yahns/wbuf_common.rb
@@ -1,7 +1,12 @@
 # -*- encoding: binary -*-
 # Copyright (C) 2009-2013, Eric Wong <normalperson@yhbt.net> et. al.
 # License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
-require 'sendfile'
+if ENV["SENDFILE_BROKEN"]
+  require_relative 'sendfile_compat'
+else
+  require 'sendfile'
+end
+
 module Yahns::WbufCommon # :nodoc:
   # returns nil on success, :wait_*able when blocked
   # currently, we rely on each thread having exclusive access to the
@@ -25,7 +30,8 @@ module Yahns::WbufCommon # :nodoc:
       raise "BUG: rv=#{rv.inspect} " \
             "on tmpio=#{@tmpio.inspect} " \
             "sf_offset=#@sf_offset sf_count=#@sf_count"
-    end while true
+    end while @sf_count > 0
+    wbuf_close(client)
   end
 
   def wbuf_close_common(client)
diff --git a/test/test_buffer_tmpdir.rb b/test/test_buffer_tmpdir.rb
index c7665f6..8461346 100644
--- a/test/test_buffer_tmpdir.rb
+++ b/test/test_buffer_tmpdir.rb
@@ -9,12 +9,18 @@ class TestBufferTmpdir < Testcase
   attr_reader :ino, :tmpdir
 
   def setup
-    @ino = SleepyPenguin::Inotify.new(:CLOEXEC)
+    @ino = nil
+    begin
+      @ino = SleepyPenguin::Inotify.new(:CLOEXEC)
+    rescue
+      skip "test needs inotify"
+    end
     @tmpdir = Dir.mktmpdir
     server_helper_setup
   end
 
   def teardown
+    return unless @ino
     server_helper_teardown
     @ino.close
     FileUtils.rm_rf @tmpdir
@@ -100,4 +106,4 @@ class TestBufferTmpdir < Testcase
     c.close if c
     quit_wait(pid)
   end
-end if SleepyPenguin.const_defined?(:Inotify)
+end
diff --git a/test/test_client_expire.rb b/test/test_client_expire.rb
index bdf0cb4..4f20803 100644
--- a/test/test_client_expire.rb
+++ b/test/test_client_expire.rb
@@ -149,9 +149,17 @@ class TestClientExpire < Testcase
         Process.waitpid2(_pid)
       end
     end
+
+    # this seems to be needed in Debian GNU/kFreeBSD
+    linux = !!(RUBY_PLATFORM =~ /linux/)
+    sleep(1) unless linux
+
     [ f, s ].each do |io|
       assert_raises(Errno::EPIPE,Errno::ECONNRESET) do
-        req.each_byte { |b| io.write(b.chr) }
+        req.each_byte do |b|
+          io.write(b.chr)
+          sleep(0.01) unless linux
+        end
       end
       io.close
     end
diff --git a/test/test_mt_accept.rb b/test/test_mt_accept.rb
index e006af8..37813d4 100644
--- a/test/test_mt_accept.rb
+++ b/test/test_mt_accept.rb
@@ -10,6 +10,8 @@ class TestMtAccept < Testcase
   alias teardown server_helper_teardown
 
   def test_mt_accept
+    skip "Linux kernel required" unless RUBY_PLATFORM =~ /linux/
+    skip "/proc not mounted" unless File.directory?("/proc")
     err, cfg, host, port = @err, Yahns::Config.new, @srv.addr[3], @srv.addr[1]
     opts = { threads: 1 }
     cfg.instance_eval do
@@ -45,4 +47,4 @@ class TestMtAccept < Testcase
   ensure
     quit_wait(pid)
   end
-end if RUBY_PLATFORM =~ /linux/ && File.directory?("/proc")
+end
diff --git a/test/test_server.rb b/test/test_server.rb
index f4c66ae..96fc704 100644
--- a/test/test_server.rb
+++ b/test/test_server.rb
@@ -178,11 +178,14 @@ class TestServer < Testcase
   end
 
   def test_check_client_connection
+    tmpdir = Dir.mktmpdir
+    sock = "#{tmpdir}/sock"
+    unix_srv = UNIXServer.new(sock)
+    unix_srv.close_on_exec = true
     msgs = %w(ZZ zz)
     err = @err
     cfg = Yahns::Config.new
     bpipe = cloexec_pipe
-    host, port = @srv.addr[3], @srv.addr[1]
     cfg.instance_eval do
       ru = lambda { |e|
         case e['PATH_INFO']
@@ -205,7 +208,7 @@ class TestServer < Testcase
       }
       GTL.synchronize {
         app(:rack, ru) {
-          listen "#{host}:#{port}"
+          listen sock
           check_client_connection true
           # needed to avoid concurrency with check_client_connection
           queue { worker_threads 1 }
@@ -223,12 +226,13 @@ class TestServer < Testcase
 
     pid = fork do
       bpipe[1].close
-      ENV["YAHNS_FD"] = @srv.fileno.to_s
+      ENV["YAHNS_FD"] = unix_srv.fileno.to_s
       srv.start.join
     end
     bpipe[0].close
-    a = get_tcp_client(host, port)
-    b = get_tcp_client(host, port)
+    a = UNIXSocket.new(sock)
+    b = UNIXSocket.new(sock)
+    b.close_on_exec = a.close_on_exec = true
     a.write("GET /sleep HTTP/1.0\r\n\r\n")
     r = IO.select([a], nil, nil, 4)
     assert r, "nothing ready"
@@ -250,10 +254,19 @@ class TestServer < Testcase
     assert_equal msgs.join, buf.split(/\r\n\r\n/)[1]
 
     # do things still work?
-    run_client(host, port) { |res| assert_equal "HI", res.body }
+    c = UNIXSocket.new(sock)
+    c.write "GET /\r\n\r\n"
+    assert_equal "HI", c.read
+    c.close
     a.close
+  rescue => e
+    warn e.class
+    warn e.message
+    warn e.backtrace.join("\n")
   ensure
+    unix_srv.close
     quit_wait(pid)
+    FileUtils.rm_rf(tmpdir)
   end
 
   def test_mp
diff --git a/test/test_wbuf.rb b/test/test_wbuf.rb
index 25cdeba..644c76b 100644
--- a/test/test_wbuf.rb
+++ b/test/test_wbuf.rb
@@ -81,8 +81,8 @@ class TestWbuf < Testcase
     assert_equal b, IO.select([b], nil, nil, 5)[0][0]
     b.read(nr - 2) if nr > 2
     assert_equal b, IO.select([b], nil, nil, 5)[0][0]
-    assert_equal "HI", b.read(2)
-    assert_equal false, wbuf.wbuf_flush(a)
+    assert_equal "HI", b.read(2), "read the end of the response"
+    assert_equal true, wbuf.wbuf_flush(a)
   ensure
     a.close
     b.close