4 files changed, 38 insertions, 45 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
diff --git a/lib/rainbows/fiber/io.rb b/lib/rainbows/fiber/io.rb
index bc6c0fe..5c51cb9 100644
--- a/lib/rainbows/fiber/io.rb
+++ b/lib/rainbows/fiber/io.rb
@@ -36,13 +36,13 @@ module Rainbows
 
       # used for reading headers (respecting keepalive_timeout)
       def read_timeout
-        expire = false
+        expire = nil
         begin
           to_io.read_nonblock(16384)
         rescue Errno::EAGAIN
           return if expire && expire < Time.now
           RD[self] = false
-          expire = Time.now + G.kato
+          expire ||= Time.now + G.kato
           ::Fiber.yield
           RD.delete(self)
           retry
diff --git a/lib/rainbows/fiber_pool.rb b/lib/rainbows/fiber_pool.rb
index 6cb2ca6..c647676 100644
--- a/lib/rainbows/fiber_pool.rb
+++ b/lib/rainbows/fiber_pool.rb
@@ -1,6 +1,5 @@
 # -*- encoding: binary -*-
 require 'rainbows/fiber'
-require 'pp'
 
 module Rainbows
 
@@ -26,26 +25,10 @@ module Rainbows
         }.resume # resume to hit ::Fiber.yield so it waits on a client
       }
       Fiber::Base.const_set(:APP, app)
-      rd = Fiber::RD
-      wr = Fiber::WR
 
       begin
-        ret = begin
-          G.tick
-          IO.select(rd.keys.concat(LISTENERS), wr.keys, nil, timer) or next
-        rescue Errno::EINTR
-          retry
-        rescue Errno::EBADF, TypeError
-          LISTENERS.compact!
-          G.cur > 0 ? retry : break
-        end
-
-        # active writers first, then _all_ readers for keepalive timeout
-        ret[1].concat(rd.keys).each { |c| c.f.resume }
-
-        # accept() is an expensive syscall
-        (ret.first & LISTENERS).each do |l|
-          fib = pool.shift or break
+        schedule do |l|
+          fib = pool.shift or break # let another worker process take it
           io = begin
             l.accept_nonblock
           rescue Errno::EAGAIN, Errno::ECONNABORTED
diff --git a/lib/rainbows/fiber_spawn.rb b/lib/rainbows/fiber_spawn.rb
index 004976a..83b64de 100644
--- a/lib/rainbows/fiber_spawn.rb
+++ b/lib/rainbows/fiber_spawn.rb
@@ -17,28 +17,10 @@ module Rainbows
       init_worker_process(worker)
       Fiber::Base.const_set(:APP, app)
       limit = worker_connections
-      rd = Rainbows::Fiber::RD
-      wr = Rainbows::Fiber::WR
       fio = Rainbows::Fiber::IO
 
       begin
-        ret = begin
-          IO.select(rd.keys.concat(LISTENERS), wr.keys, nil, timer) or next
-        rescue Errno::EINTR
-          G.tick
-          retry
-        rescue Errno::EBADF, TypeError
-          LISTENERS.compact!
-          G.cur > 0 ? retry : break
-        end
-        G.tick
-
-        # active writers first, then _all_ readers for keepalive timeout
-        ret[1].concat(rd.keys).each { |c| c.f.resume }
-        G.tick
-
-        # accept() is an expensive syscall
-        (ret.first & LISTENERS).each do |l|
+        schedule do |l|
           break if G.cur >= limit
           io = begin
             l.accept_nonblock
@@ -47,10 +29,9 @@ module Rainbows
           end
           ::Fiber.new { process_client(fio.new(io, ::Fiber.current)) }.resume
         end
-        G.tick
       rescue => e
         listen_loop_error(e)
-      end while G.tick || G.cur > 0
+      end while G.alive || G.cur > 0
     end
 
   end