Improved Async implementation using global thread pool.

Author: jdantonio

examples/benchmark_async.rb

@@ -105,28 +105,61 @@ def foo(latch = nil)
 Calculating -------------------------------------
            celluloid    24.000  i/100ms
 async, thread per object
-                        30.000  i/100ms
+                        31.000  i/100ms
 async, global thread pool
                         31.000  i/100ms
 -------------------------------------------------
-           celluloid    242.345  (±10.7%) i/s -      1.200k
+           celluloid    277.834  (± 9.4%) i/s -      1.392k
 async, thread per object
-                        316.387  (± 2.5%) i/s -      1.590k
+                        316.357  (± 1.9%) i/s -      1.612k
 async, global thread pool
-                        318.200  (± 1.6%) i/s -      1.612k
+                        318.707  (± 2.2%) i/s -      1.612k
 
 Comparison:
-async, global thread pool:      318.2 i/s
+async, global thread pool:      318.7 i/s
 async, thread per object:      316.4 i/s - 1.01x slower
-           celluloid:      242.3 i/s - 1.31x slower
+           celluloid:      277.8 i/s - 1.15x slower
+
+Rehearsal -------------------------------------------------------------
+celluloid                   4.110000   0.650000   4.760000 (  4.766239)
+async, thread per object    3.370000   0.100000   3.470000 (  3.420537)
+async, global thread pool   3.460000   0.240000   3.700000 (  3.598044)
+--------------------------------------------------- total: 11.930000sec
+
+                                user     system      total        real
+celluloid                   4.000000   0.640000   4.640000 (  4.652382)
+async, thread per object    3.640000   0.160000   3.800000 (  3.751535)
+async, global thread pool   3.440000   0.220000   3.660000 (  3.550602)
+
+===========================================================
+jruby 1.7.19 (1.9.3p551) 2015-01-29 20786bd on Java HotSpot(TM) 64-Bit Server VM 1.8.0_45-b14 +jit [darwin-x86_64]
+===========================================================
+
+Calculating -------------------------------------
+           celluloid     2.000  i/100ms
+async, thread per object
+                        23.000  i/100ms
+async, global thread pool
+                        60.000  i/100ms
+-------------------------------------------------
+           celluloid    155.480  (±38.6%) i/s -    606.000
+async, thread per object
+                        823.969  (±18.2%) i/s -      3.404k
+async, global thread pool
+                        852.728  (±14.7%) i/s -      4.140k
+
+Comparison:
+async, global thread pool:      852.7 i/s
+async, thread per object:      824.0 i/s - 1.03x slower
+           celluloid:      155.5 i/s - 5.48x slower
 
 Rehearsal -------------------------------------------------------------
-celluloid                   4.170000   0.630000   4.800000 (  4.812120)
-async, thread per object    3.400000   0.110000   3.510000 (  3.452749)
-async, global thread pool   3.410000   0.070000   3.480000 (  3.455878)
---------------------------------------------------- total: 11.790000sec
+celluloid                   5.640000   1.560000   7.200000 (  5.480000)
+async, thread per object    2.660000   0.240000   2.900000 (  1.670000)
+async, global thread pool   2.110000   0.240000   2.350000 (  1.360000)
+--------------------------------------------------- total: 12.450000sec
 
                                 user     system      total        real
-celluloid                   4.080000   0.620000   4.700000 (  4.687752)
-async, thread per object    3.380000   0.160000   3.540000 (  3.469882)
-async, global thread pool   3.380000   0.050000   3.430000 (  3.426759)
+celluloid                   5.650000   1.540000   7.190000 (  5.470000)
+async, thread per object    2.350000   0.230000   2.580000 (  1.532000)
+async, global thread pool   1.910000   0.220000   2.130000 (  1.272000)

lib/concurrent/async_alternate.rb

@@ -1,6 +1,6 @@
+require 'thread'
 require 'concurrent/configuration'
 require 'concurrent/ivar'
-require 'concurrent/executor/immediate_executor'
 
 module Concurrent
 
@@ -42,10 +42,9 @@ module Concurrent
   # ## Basic Usage
   #
   # When this module is mixed into a class, objects of the class become inherently
-  # asynchronous. Each object gets its own background thread (specifically,
-  # `SingleThreadExecutor`) on which to post asynchronous method calls.
-  # Asynchronous method calls are executed in the background one at a time in
-  # the order they are received.
+  # asynchronous. Each object gets its own background thread on which to post
+  # asynchronous method calls. Asynchronous method calls are executed in the
+  # background one at a time in the order they are received.
   #
   # To create an asynchronous class, simply mix in the `Concurrent::Async` module:
   #
@@ -232,7 +231,6 @@ module Concurrent
   #                          # returns an IVar in the :complete state
   #
   # @see Concurrent::Actor
-  # @see Concurrent::SingleThreadExecutor
   # @see https://en.wikipedia.org/wiki/Actor_model "Actor Model" at Wikipedia
   # @see http://www.erlang.org/doc/man/gen_server.html Erlang gen_server
   # @see http://c2.com/cgi/wiki?LetItCrash "Let It Crash" at http://c2.com/
@@ -307,13 +305,15 @@ class AsyncDelegator
       # given executor. Block if necessary.
       #
       # @param [Object] delegate the object to wrap and delegate method calls to
-      # @param [Concurrent::ExecutorService] executor the executor on which to execute delegated method calls
+      # @param [Array] job queue which guarantees serialization of method calls
+      # @param [Mutex] mutex which synchronizes queue operations
       # @param [Boolean] blocking will block awaiting result when `true`
-      def initialize(delegate, executor, serializer, blocking)
+      def initialize(delegate, queue, mutex, blocking)
         @delegate = delegate
-        @executor = executor
-        @serializer = serializer
+        @queue = queue
+        @mutex = mutex
         @blocking = blocking
+        @executor = Concurrent.global_io_executor
       end
 
       # Delegates method calls to the wrapped object. For performance,
@@ -332,15 +332,28 @@ def method_missing(method, *args, &block)
         Async::validate_argc(@delegate, method, *args)
 
         ivar = Concurrent::IVar.new
-        @serializer.post(@executor, args) do |arguments|
+        @mutex.synchronize do
+          @queue.push [ivar, method, args, block]
+          @executor.post { perform } if @queue.length == 1
+        end
+
+        ivar.wait if @blocking
+        ivar
+      end
+
+      def perform
+        loop do
+          ivar, method, args, block = @mutex.synchronize { @queue.first }
+          break unless ivar # queue is empty
+
           begin
-            ivar.set(@delegate.send(method, *arguments, &block))
+            ivar.set(@delegate.send(method, *args, &block))
           rescue => error
             ivar.fail(error)
           end
+
+          @mutex.synchronize { @queue.shift }
         end
-        ivar.wait if @blocking
-        ivar
       end
     end
     private_constant :AsyncDelegator
@@ -387,8 +400,6 @@ def await
     end
     alias_method :call, :await
 
-    private
-
     # Initialize the internal serializer and other stnchronization mechanisms.
     #
     # @note This method *must* be called immediately upon object construction.
@@ -398,9 +409,10 @@ def await
     def init_synchronization
       return self if @__async_initialized__
       @__async_initialized__ = true
-      serializer = Concurrent::SerializedExecution.new
-      @__await_delegator__ = AsyncDelegator.new(self, Concurrent::ImmediateExecutor.new, serializer, true)
-      @__async_delegator__ = AsyncDelegator.new(self, Concurrent.global_io_executor, serializer, false)
+      queue = []
+      mutex = Mutex.new
+      @__await_delegator__ = AsyncDelegator.new(self, queue, mutex, true)
+      @__async_delegator__ = AsyncDelegator.new(self, queue, mutex, false)
       self
     end
   end