lib/rainbows/xepoll/client.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3
   4 module Rainbows::XEpoll::Client
   5   N = Raindrops.new(1)
   6   include Rainbows::Epoll::Client
   7   ACCEPTORS = Rainbows::HttpServer::LISTENERS.dup
   8   extend Rainbows::WorkerYield
   9
  10   def self.included(klass)
  11     max = Rainbows.server.worker_connections
  12     ACCEPTORS.map! do |sock|
  13       Thread.new do
  14         begin
  15           if io = sock.kgio_accept(klass)
  16             N.incr(0, 1)
  17             io.epoll_once
  18           end
  19           worker_yield while N[0] >= max
  20         rescue => e
  21           Rainbows::Error.listen_loop(e)
  22         end while Rainbows.alive
  23       end
  24     end
  25   end
  26
  27   def self.loop
  28     begin
  29       EP.wait(nil, 1000) { |_, obj| obj.epoll_run }
  30       while obj = ReRun.shift
  31         obj.epoll_run
  32       end
  33       Rainbows::Epoll::Client.expire
  34     rescue Errno::EINTR
  35     rescue => e
  36       Rainbows::Error.listen_loop(e)
  37     end while Rainbows.tick || N[0] > 0
  38     Rainbows::JoinThreads.acceptors(ACCEPTORS)
  39   end
  40
  41   # only call this once
  42   def epoll_once
  43     @wr_queue = [] # may contain String, ResponsePipe, and StreamFile objects
  44     post_init
  45     EP.set(self, IN) # wake up the main thread
  46     rescue => e
  47       Rainbows::Error.write(self, e)
  48   end
  49
  50   def on_close
  51     KATO.delete(self)
  52     N.decr(0, 1)
  53   end
  54 end