add RevThreadPool concurrency model

This should be like RevThreadSpawn except with more predictable
performance (but higher memory usage under low load).

8 files changed, 203 insertions, 75 deletions

diff --git a/Documentation/comparison.haml b/Documentation/comparison.haml
index ae0ef47..db8e405 100644
--- a/Documentation/comparison.haml
+++ b/Documentation/comparison.haml
@@ -85,13 +85,17 @@
     %td.r19 Yes
     %td.rbx No
     %td.slow Yes
+  %tr.comp_row
+    %td.mod RevThreadPool
+    %td.tee No
+    %td.r18 Slow*
+    %td.r19 Yes
+    %td.rbx No
+    %td.slow Yes
 %ul
   %li
-    RevThreadSpawn + 1.8 performance is being improved, follow
-    the
-    %a(href="http://rubyforge.org/mailman/listinfo/rev-talk")
-      rev-talk mailing list
-    for details.
+    RevThread* + 1.8 performance is bad with Rev <= 0.3.1.
+    Rev 0.3.2 (when it is released) should be much faster under 1.8.
   %li
     waiting on Rubinius for better signal handling
   %li
@@ -180,6 +184,13 @@
       %a(href="http://rubyeventmachine.com") EventMachine
     %td.thr No
     %td.reent Yes
+  %tr.comp_row
+    %td.mod RevThreadPool
+    %td.slowio
+      thread-safe Ruby,
+      %a(href="http://rev.rubyforge.org/") Rev
+    %td.thr Yes
+    %td.reent No
 
 %ul
   %li
@@ -276,6 +287,12 @@
     %td.app_pool Yes*
     %td.lock Yes*
     %td.async NeverBlock, async_sinatra
+  %tr.comp_row
+    %td.mod RevThreadPool
+    %td.devfd Yes
+    %td.app_pool Yes
+    %td.lock Dumb
+    %td.async standard Ruby
 
 %ul
   %li
diff --git a/lib/rainbows.rb b/lib/rainbows.rb
index 9e07185..e42f6cf 100644
--- a/lib/rainbows.rb
+++ b/lib/rainbows.rb
@@ -80,6 +80,7 @@ module Rainbows
     :ThreadPool => 10,
     :Rev => 50,
     :RevThreadSpawn => 50,
+    :RevThreadPool => 50,
     :EventMachine => 50,
     :FiberSpawn => 50,
     :FiberPool => 50,
diff --git a/lib/rainbows/rev/master.rb b/lib/rainbows/rev/master.rb
new file mode 100644
index 0000000..5c112c6
--- /dev/null
+++ b/lib/rainbows/rev/master.rb
@@ -0,0 +1,29 @@
+# -*- encoding: binary -*-
+require 'rainbows/rev'
+
+RUBY_VERSION =~ %r{\A1\.8} && ::Rev::VERSION < "0.3.2" and
+  warn "Rainbows::RevThreadSpawn + Rev (< 0.3.2)" \
+       " does not work well under Ruby 1.8"
+
+module Rainbows
+
+  module Rev
+    class Master < ::Rev::AsyncWatcher
+
+      def initialize(queue)
+        super()
+        @queue = queue
+      end
+
+      def <<(output)
+        @queue << output
+        signal
+      end
+
+      def on_signal
+        client, response = @queue.pop
+        client.response_write(response)
+      end
+    end
+  end
+end
diff --git a/lib/rainbows/rev/thread.rb b/lib/rainbows/rev/thread.rb
new file mode 100644
index 0000000..8fa43ac
--- /dev/null
+++ b/lib/rainbows/rev/thread.rb
@@ -0,0 +1,53 @@
+# -*- encoding: binary -*-
+require 'thread'
+require 'rainbows/rev/master'
+
+module Rainbows
+  module Rev
+
+    class ThreadClient < Client
+
+      def app_call
+        KATO.delete(self)
+        disable
+        @env[RACK_INPUT] = @input
+        @input = nil # not sure why, @input seems to get closed otherwise...
+        app_dispatch # must be implemented by subclass
+      end
+
+      # this is only called in the master thread
+      def response_write(response)
+        enable
+        alive = @hp.keepalive? && G.alive
+        out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if @hp.headers?
+        DeferredResponse.write(self, response, out)
+        return quit unless alive && G.alive
+
+        @env.clear
+        @hp.reset
+        @state = :headers
+        # keepalive requests are always body-less, so @input is unchanged
+        if @hp.headers(@env, @buf)
+          @input = HttpRequest::NULL_IO
+          app_call
+        else
+          KATO[self] = Time.now
+        end
+      end
+
+      # fails-safe application dispatch, we absolutely cannot
+      # afford to fail or raise an exception (killing the thread)
+      # here because that could cause a deadlock and we'd leak FDs
+      def app_response
+        begin
+          @env[REMOTE_ADDR] = @remote_addr
+          APP.call(@env.update(RACK_DEFAULTS))
+        rescue => e
+          Error.app(e) # we guarantee this does not raise
+          [ 500, {}, [] ]
+        end
+      end
+
+    end
+  end
+end
diff --git a/lib/rainbows/rev_thread_pool.rb b/lib/rainbows/rev_thread_pool.rb
new file mode 100644
index 0000000..47b451e
--- /dev/null
+++ b/lib/rainbows/rev_thread_pool.rb
@@ -0,0 +1,75 @@
+# -*- encoding: binary -*-
+require 'rainbows/rev/thread'
+
+module Rainbows
+
+  # A combination of the Rev and ThreadPool models.  This allows Ruby
+  # Thread-based concurrency for application processing.  It DOES NOT
+  # expose a streamable "rack.input" for upload processing within the
+  # app.  DevFdResponse should be used with this class to proxy
+  # asynchronous responses.  All network I/O between the client and
+  # server are handled by the main thread and outside of the core
+  # application dispatch.
+  #
+  # Unlike ThreadPool, Rev makes this model highly suitable for
+  # slow clients and applications with medium-to-slow response times
+  # (I/O bound), but less suitable for sleepy applications.
+  #
+  # WARNING: this model does not currently perform well under 1.8 with
+  # Rev 0.3.1.  Rev 0.3.2 should include significant performance
+  # improvements under Ruby 1.8.
+
+  module RevThreadPool
+
+    DEFAULTS = {
+      :pool_size => 10, # same default size as ThreadPool (w/o Rev)
+    }
+
+    def self.setup
+      DEFAULTS.each { |k,v| O[k] ||= v }
+      Integer === O[:pool_size] && O[:pool_size] > 0 or
+        raise ArgumentError, "pool_size must a be an Integer > 0"
+    end
+
+    class PoolWatcher < ::Rev::TimerWatcher
+      def initialize(threads)
+        @threads = threads
+        super(G.server.timeout, true)
+      end
+
+      def on_timer
+        @threads.each { |t| t.join(0) and G.quit! }
+      end
+    end
+
+    class Client < Rainbows::Rev::ThreadClient
+      def app_dispatch
+        QUEUE << self
+      end
+    end
+
+    include Rainbows::Rev::Core
+
+    def init_worker_threads(master, queue)
+      O[:pool_size].times.map do
+        Thread.new do
+          begin
+            client = queue.pop
+            master << [ client, client.app_response ]
+          rescue => e
+            Error.listen_loop(e)
+          end while true
+        end
+      end
+    end
+
+    def init_worker_process(worker)
+      super
+      master = Rev::Master.new(Queue.new).attach(::Rev::Loop.default)
+      queue = Client.const_set(:QUEUE, Queue.new)
+      threads = init_worker_threads(master, queue)
+      PoolWatcher.new(threads).attach(::Rev::Loop.default)
+      logger.info "RevThreadPool pool_size=#{O[:pool_size]}"
+    end
+  end
+end
diff --git a/lib/rainbows/rev_thread_spawn.rb b/lib/rainbows/rev_thread_spawn.rb
index 0bfeb36..00d8b6b 100644
--- a/lib/rainbows/rev_thread_spawn.rb
+++ b/lib/rainbows/rev_thread_spawn.rb
@@ -1,9 +1,5 @@
 # -*- encoding: binary -*-
-require 'rainbows/rev'
-
-RUBY_VERSION =~ %r{\A1\.8} && ::Rev::VERSION < "0.3.2" and
-  warn "Rainbows::RevThreadSpawn + Rev (< 0.3.2)" \
-       " does not work well under Ruby 1.8"
+require 'rainbows/rev/thread'
 
 module Rainbows
 
@@ -15,73 +11,19 @@ module Rainbows
   # server are handled by the main thread and outside of the core
   # application dispatch.
   #
-  # WARNING: this model does not currently perform well under 1.8.  See the
-  # {rev-talk mailing list}[http://rubyforge.org/mailman/listinfo/rev-talk]
-  # for ongoing performance work that will hopefully make it into the
-  # next release of {Rev}[http://rev.rubyforge.org/].
+  # Unlike ThreadSpawn, Rev makes this model highly suitable for
+  # slow clients and applications with medium-to-slow response times
+  # (I/O bound), but less suitable for sleepy applications.
+  #
+  # WARNING: this model does not currently perform well under 1.8 with
+  # Rev 0.3.1.  Rev 0.3.2 should include significant performance
+  # improvements under Ruby 1.8.
 
   module RevThreadSpawn
 
-    class Master < ::Rev::AsyncWatcher
-
-      def initialize
-        super
-        @queue = Queue.new
-      end
-
-      def <<(output)
-        @queue << output
-        signal
-      end
-
-      def on_signal
-        client, response = @queue.pop
-        client.response_write(response)
-      end
-    end
-
-    class Client < Rainbows::Rev::Client
-      DR = Rainbows::Rev::DeferredResponse
-      KATO = Rainbows::Rev::KATO
-
-      def response_write(response)
-        enable
-        alive = @hp.keepalive? && G.alive
-        out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if @hp.headers?
-        DR.write(self, response, out)
-        return quit unless alive && G.alive
-
-        @env.clear
-        @hp.reset
-        @state = :headers
-        # keepalive requests are always body-less, so @input is unchanged
-        if @hp.headers(@env, @buf)
-          @input = HttpRequest::NULL_IO
-          app_call
-        else
-          KATO[self] = Time.now
-        end
-      end
-
-      # fails-safe application dispatch, we absolutely cannot
-      # afford to fail or raise an exception (killing the thread)
-      # here because that could cause a deadlock and we'd leak FDs
-      def app_response
-        begin
-          @env[REMOTE_ADDR] = @remote_addr
-          APP.call(@env.update(RACK_DEFAULTS))
-        rescue => e
-          Error.app(e) # we guarantee this does not raise
-          [ 500, {}, [] ]
-        end
-      end
-
-      def app_call
-        KATO.delete(client = self)
-        disable
-        @env[RACK_INPUT] = @input
-        @input = nil # not sure why, @input seems to get closed otherwise...
-        Thread.new { MASTER << [ client, app_response ] }
+    class Client < Rainbows::Rev::ThreadClient
+      def app_dispatch
+        Thread.new(self) { |client| MASTER << [ client, app_response ] }
       end
     end
 
@@ -89,7 +31,8 @@ module Rainbows
 
     def init_worker_process(worker)
       super
-      Client.const_set(:MASTER, Master.new.attach(::Rev::Loop.default))
+      master = Rev::Master.new(Queue.new).attach(::Rev::Loop.default)
+      Client.const_set(:MASTER, master)
     end
 
   end
diff --git a/t/GNUmakefile b/t/GNUmakefile
index 32893ce..ec04607 100644
--- a/t/GNUmakefile
+++ b/t/GNUmakefile
@@ -32,6 +32,7 @@ ifeq ($(ONENINE),true)
 
   # technically this works under 1.8, but wait until rev 0.3.2
   models += RevThreadSpawn
+  models += RevThreadPool
 endif
 all_models := $(models) Base
 
diff --git a/t/simple-http_RevThreadPool.ru b/t/simple-http_RevThreadPool.ru
new file mode 100644
index 0000000..43dde1a
--- /dev/null
+++ b/t/simple-http_RevThreadPool.ru
@@ -0,0 +1,9 @@
+use Rack::ContentLength
+use Rack::ContentType
+run lambda { |env|
+  if env['rack.multithread'] && env['rainbows.model'] == :RevThreadPool
+    [ 200, {}, [ env.inspect << "\n" ] ]
+  else
+    raise "rack.multithread is false"
+  end
+}