From 667b05819b85165061c445fb2c75ef161a5de5b6 Mon Sep 17 00:00:00 2001
From: Eric Wong <normalperson@yhbt.net>
Date: Mon, 24 Jan 2011 20:04:20 -0800
Subject: initial XAcceptEpoll concurrency model

Edge-triggered epoll concurrency model with blocking accept() in
a (hopefully) native thread.  This is recommended over Epoll for
Ruby 1.9 users as it can workaround accept()-scalability issues
on multicore machines.
---
 lib/rainbows/http_server.rb          |  2 +-
 lib/rainbows/max_body.rb             |  2 +-
 lib/rainbows/xaccept_epoll.rb        | 25 +++++++++++++++
 lib/rainbows/xaccept_epoll/client.rb | 59 ++++++++++++++++++++++++++++++++++++
 4 files changed, 86 insertions(+), 2 deletions(-)
 create mode 100644 lib/rainbows/xaccept_epoll.rb
 create mode 100644 lib/rainbows/xaccept_epoll/client.rb

(limited to 'lib/rainbows')

diff --git a/lib/rainbows/http_server.rb b/lib/rainbows/http_server.rb
index a9767bf..91c6cad 100644
--- a/lib/rainbows/http_server.rb
+++ b/lib/rainbows/http_server.rb
@@ -72,7 +72,7 @@ class Rainbows::HttpServer < Unicorn::HttpServer
     new_defaults = {
       'rainbows.model' => (@use = model.to_sym),
       'rack.multithread' => !!(model.to_s =~ /Thread/),
-      'rainbows.autochunk' => [:Coolio,:Rev,:Epoll,
+      'rainbows.autochunk' => [:Coolio,:Rev,:Epoll,:XAcceptEpoll,
                                :EventMachine,:NeverBlock].include?(@use),
     }
     Rainbows::Const::RACK_DEFAULTS.update(new_defaults)
diff --git a/lib/rainbows/max_body.rb b/lib/rainbows/max_body.rb
index 105bbb0..68ae3e3 100644
--- a/lib/rainbows/max_body.rb
+++ b/lib/rainbows/max_body.rb
@@ -56,7 +56,7 @@ class Rainbows::MaxBody
     when :Rev, :Coolio, :EventMachine, :NeverBlock,
          :RevThreadSpawn, :RevThreadPool,
          :CoolioThreadSpawn, :CoolioThreadPool,
-         :Epoll
+         :Epoll, :XAcceptEpoll
       return
     end
 
diff --git a/lib/rainbows/xaccept_epoll.rb b/lib/rainbows/xaccept_epoll.rb
new file mode 100644
index 0000000..0690c07
--- /dev/null
+++ b/lib/rainbows/xaccept_epoll.rb
@@ -0,0 +1,25 @@
+# -*- encoding: binary -*-
+# :enddoc:
+require 'raindrops'
+require 'rainbows/epoll'
+
+# Edge-triggered epoll concurrency model with blocking accept() in
+# a (hopefully) native thread.  This is recommended over Epoll for
+# Ruby 1.9 users as it can workaround accept()-scalability issues
+# on multicore machines.
+module Rainbows::XAcceptEpoll
+  include Rainbows::Base
+  autoload :Client, 'rainbows/xaccept_epoll/client'
+
+  def init_worker_process(worker)
+    super
+    Rainbows::Epoll.const_set :EP, SleepyPenguin::Epoll.new
+    trap(:QUIT) { Rainbows::Epoll.quit! }
+    Rainbows::Client.__send__ :include, Client
+  end
+
+  def worker_loop(worker) # :nodoc:
+    init_worker_process(worker)
+    Client.run
+  end
+end
diff --git a/lib/rainbows/xaccept_epoll/client.rb b/lib/rainbows/xaccept_epoll/client.rb
new file mode 100644
index 0000000..6c3a0f5
--- /dev/null
+++ b/lib/rainbows/xaccept_epoll/client.rb
@@ -0,0 +1,59 @@
+# -*- encoding: binary -*-
+# :enddoc:
+
+module Rainbows::XAcceptEpoll::Client
+  include Rainbows::Epoll::Client
+  MAX = Rainbows.server.worker_connections
+  THRESH = MAX - 1
+  EP = Rainbows::Epoll::EP
+  N = Raindrops.new(1)
+  @timeout = Rainbows.server.timeout / 2.0
+  THREADS = Rainbows::HttpServer::LISTENERS.map do |sock|
+    Thread.new(sock) do |sock|
+      sleep
+      begin
+        if io = sock.kgio_accept
+          N.incr(0, 1)
+          io.epoll_once
+        end
+        sleep while N[0] >= MAX
+      rescue => e
+        Rainbows::Error.listen_loop(e)
+      end while Rainbows.alive
+    end
+  end
+
+  def self.run
+    THREADS.each { |t| t.run }
+    begin
+      EP.wait(nil, @timeout) { |flags, obj| obj.epoll_run }
+      Rainbows::Epoll::Client.expire
+    rescue => e
+      Rainbows::Error.listen_loop(e)
+    end while Rainbows.tick
+
+    THREADS.delete_if do |thr|
+      Rainbows.tick
+      begin
+        thr.run
+        thr.join(0.01)
+      rescue
+        true
+      end
+    end until THREADS.empty?
+  end
+
+  # only call this once
+  def epoll_once
+    @wr_queue = [] # may contain String, ResponsePipe, and StreamFile objects
+    post_init
+    EP.set(self, IN) # wake up the main thread
+    rescue => e
+      Rainbows::Error.write(self, e)
+  end
+
+  def on_close
+    KATO.delete(self)
+    N.decr(0, 1) == THRESH and THREADS.each { |t| t.run }
+  end
+end
-- 
cgit v1.2.3-24-ge0c7