lib/rainbows/fiber/base.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 require 'rainbows/fiber/io'
   4
   5 module Rainbows::Fiber::Base
   6
   7   include Rainbows::Base
   8
   9   # :stopdoc:
  10   RD = Rainbows::Fiber::RD
  11   WR = Rainbows::Fiber::WR
  12   ZZ = Rainbows::Fiber::ZZ
  13   # :startdoc:
  14
  15   # the scheduler method that powers both FiberSpawn and FiberPool
  16   # concurrency models.  It times out idle clients and attempts to
  17   # schedules ones that were blocked on I/O.  At most it'll sleep
  18   # for one second (returned by the schedule_sleepers method) which
  19   # will cause it.
  20   def schedule
  21     begin
  22       Rainbows.tick
  23       t = schedule_sleepers
  24       ret = select(RD.compact.concat(LISTENERS), WR.compact, nil, t)
  25     rescue Errno::EINTR
  26       retry
  27     rescue Errno::EBADF, TypeError
  28       LISTENERS.compact!
  29       raise
  30     end or return
  31
  32     # active writers first, then readers
  33     ret[1].concat(RD.compact & ret[0]).each { |c| c.f.resume }
  34
  35     # accept is an expensive syscall, filter out listeners we don't want
  36     (ret[0] & LISTENERS).each { |x| yield x }
  37   end
  38
  39   # wakes up any sleepers or keepalive-timeout violators that need to be
  40   # woken and returns an interval to IO.select on
  41   def schedule_sleepers
  42     max = nil
  43     now = Time.now
  44     fibs = []
  45     ZZ.delete_if { |fib, time|
  46       if now >= time
  47         fibs << fib
  48       else
  49         max = time
  50         false
  51       end
  52     }
  53     fibs.each { |fib| fib.resume }
  54
  55     max_sleep = 1.0 # wake up semi-frequently to prevent SIGKILL from master
  56     if max
  57       max -= Time.now
  58       return 0 if max < 0.0
  59       return max_sleep if max > max_sleep
  60       max
  61     else
  62       max_sleep
  63     end
  64   end
  65
  66   def process(client)
  67     Rainbows.cur += 1
  68     client.process_loop
  69   ensure
  70     Rainbows.cur -= 1
  71     ZZ.delete(client.f)
  72   end
  73
  74   def self.setup(klass, app)
  75     Rainbows::Client.__send__(:include, Rainbows::Fiber::IO::Methods)
  76     require 'rainbows/fiber/body'
  77     Rainbows::Client.__send__(:include, Rainbows::Fiber::Body)
  78     self.const_set(:APP, app)
  79   end
  80 end