Fiber*: cleanup scheduling, fix keepalive

Both FiberSpawn and FiberPool share similar main loops, the
only difference being the handling of connection acceptance.
So move the scheduler into it's own function for consistency.

We'll also correctly implement keepalive timeout so clients
get disconnected at the right time.

1 files changed, 32 insertions, 3 deletions

diff --git a/lib/rainbows/fiber.rb b/lib/rainbows/fiber.rb
index f0755aa..1927a78 100644
--- a/lib/rainbows/fiber.rb
+++ b/lib/rainbows/fiber.rb
@@ -10,6 +10,10 @@ module Rainbows
     WR = {}
     ZZ = {}
 
+    # puts the current Fiber into uninterruptible sleep for at least
+    # +seconds+.  Unlike Kernel#sleep, this it is not possible to sleep
+    # indefinitely to be woken up (nobody wants that in a web server,
+    # right?).
     def self.sleep(seconds)
       ZZ[::Fiber.current] = Time.now + seconds
       ::Fiber.yield
@@ -18,9 +22,34 @@ module Rainbows
     module Base
       include Rainbows::Base
 
+      # the scheduler method that powers both FiberSpawn and FiberPool
+      # concurrency models.  It times out idle clients and attempts to
+      # schedules ones that were blocked on I/O.  At most it'll sleep
+      # for one second (returned by the schedule_sleepers method) which
+      # will cause it.
+      def schedule(&block)
+        ret = begin
+          G.tick
+          RD.keys.each { |c| c.f.resume } # attempt to time out idle clients
+          t = schedule_sleepers
+          Kernel.select(RD.keys.concat(LISTENERS), WR.keys, nil, t) or return
+        rescue Errno::EINTR
+          retry
+        rescue Errno::EBADF, TypeError
+          LISTENERS.compact!
+          raise
+        end or return
+
+        # active writers first, then _all_ readers for keepalive timeout
+        ret[1].concat(RD.keys).each { |c| c.f.resume }
+
+        # accept is an expensive syscall, filter out listeners we don't want
+        (ret.first & LISTENERS).each(&block)
+      end
+
       # wakes up any sleepers that need to be woken and
       # returns an interval to IO.select on
-      def timer
+      def schedule_sleepers
         max = nil
         now = Time.now
         ZZ.delete_if { |fib, time|
@@ -46,7 +75,7 @@ module Rainbows
 
         begin # loop
           while ! hp.headers(env, buf)
-            buf << client.read_timeout or return
+            buf << (client.read_timeout or return)
           end
 
           env[RACK_INPUT] = 0 == hp.content_length ?
@@ -64,7 +93,6 @@ module Rainbows
           out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if hp.headers?
           HttpResponse.write(client, response, out)
         end while alive and hp.reset.nil? and env.clear
-        io.close
       rescue => e
         handle_error(io, e)
       ensure
@@ -72,6 +100,7 @@ module Rainbows
         RD.delete(client)
         WR.delete(client)
         ZZ.delete(client.f)
+        io.close unless io.closed?
       end
 
     end