diff options
author | Eric Wong <normalperson@yhbt.net> | 2011-01-24 20:04:20 -0800 |
---|---|---|
committer | Eric Wong <normalperson@yhbt.net> | 2011-01-24 20:07:09 -0800 |
commit | 667b05819b85165061c445fb2c75ef161a5de5b6 (patch) | |
tree | af69ab3fe1f143c408c6c9013debf1c9afe26c72 /lib/rainbows/xaccept_epoll/client.rb | |
parent | 9f7f497dc21d904a1af20465318d85811daf0652 (diff) | |
download | rainbows-667b05819b85165061c445fb2c75ef161a5de5b6.tar.gz |
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.
Diffstat (limited to 'lib/rainbows/xaccept_epoll/client.rb')
-rw-r--r-- | lib/rainbows/xaccept_epoll/client.rb | 59 |
1 files changed, 59 insertions, 0 deletions
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 |