lib/rainbows/xaccept_epoll/client.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3
   4 module Rainbows::XAcceptEpoll::Client
   5   include Rainbows::Epoll::Client
   6   MAX = Rainbows.server.worker_connections
   7   THRESH = MAX - 1
   8   EP = Rainbows::Epoll::EP
   9   N = Raindrops.new(1)
  10   @timeout = Rainbows.server.timeout / 2.0
  11   THREADS = Rainbows::HttpServer::LISTENERS.map do |sock|
  12     Thread.new(sock) do |sock|
  13       sleep
  14       begin
  15         if io = sock.kgio_accept
  16           N.incr(0, 1)
  17           io.epoll_once
  18         end
  19         sleep while N[0] >= MAX
  20       rescue => e
  21         Rainbows::Error.listen_loop(e)
  22       end while Rainbows.alive
  23     end
  24   end
  25
  26   def self.run
  27     THREADS.each { |t| t.run }
  28     begin
  29       EP.wait(nil, @timeout) { |flags, obj| obj.epoll_run }
  30       Rainbows::Epoll::Client.expire
  31     rescue Errno::EINTR
  32     rescue => e
  33       Rainbows::Error.listen_loop(e)
  34     end while Rainbows.tick
  35
  36     THREADS.delete_if do |thr|
  37       Rainbows.tick
  38       begin
  39         thr.run
  40         thr.join(0.01)
  41       rescue
  42         true
  43       end
  44     end until THREADS.empty?
  45   end
  46
  47   # only call this once
  48   def epoll_once
  49     @wr_queue = [] # may contain String, ResponsePipe, and StreamFile objects
  50     post_init
  51     EP.set(self, IN) # wake up the main thread
  52     rescue => e
  53       Rainbows::Error.write(self, e)
  54   end
  55
  56   def on_close
  57     KATO.delete(self)
  58     N.decr(0, 1) == THRESH and THREADS.each { |t| t.run }
  59   end
  60 end