Merge pull request #95 from jdantonio/fix/async-initialization

Safer initialization of the Async internal synchronization objects.

Author: jdantonio

README.md

@@ -126,14 +126,6 @@ task = Concurrent::ScheduledTask.execute(2){ Ticker.new.get_year_end_closing('IN
 task.state #=> :pending
 sleep(3)   # do other stuff
 task.value #=> 25.96
-
-# Async
-ticker = Ticker.new
-ticker.extend(Concurrent::Async)
-hpq = ticker.async.get_year_end_closing('HPQ', 2013)
-ibm = ticker.await.get_year_end_closing('IBM', 2013)
-hpq.value #=> 27.98
-ibm.value #=> 187.57
 ```
 
 ## Contributors

lib/concurrent/async.rb

@@ -1,11 +1,14 @@
 require 'thread'
 require 'concurrent/configuration'
+require 'concurrent/delay'
 require 'concurrent/ivar'
 require 'concurrent/future'
 require 'concurrent/executor/thread_pool_executor'
 
 module Concurrent
 
+  InitializationError = Class.new(StandardError)
+
   # A mixin module that provides simple asynchronous behavior to any standard
   # class/object or object. 
   #
@@ -36,6 +39,10 @@ module Concurrent
   #   class Echo
   #     include Concurrent::Async
   #
+  #     def initialize
+  #       init_mutex # initialize the internal synchronization objects
+  #     end
+  #
   #     def echo(msg)
   #       sleep(rand)
   #       print "#{msg}\n"
@@ -52,6 +59,7 @@ module Concurrent
   # @example Monkey-patching an existing object
   #   numbers = 1_000_000.times.collect{ rand }
   #   numbers.extend(Concurrent::Async)
+  #   numbers.init_mutex # initialize the internal synchronization objects
   #   
   #   future = numbers.async.max
   #   future.state #=> :pending
@@ -61,6 +69,16 @@ module Concurrent
   #   future.state #=> :fulfilled
   #   future.value #=> 0.999999138918843
   #
+  # @note This module depends on several internal synchronization objects that
+  #       must be initialized prior to calling any of the async/await/executor methods.
+  #       The best practice is to call `init_mutex` from within the constructor
+  #       of the including class. A less ideal but acceptable practice is for the
+  #       thread creating the asynchronous object to explicitly call the `init_mutex`
+  #       method prior to calling any of the async/await/executor methods. If
+  #       `init_mutex` is *not* called explicitly the async/await/executor methods
+  #       will raize a `Concurrent::InitializationError`. This is the only way 
+  #       thread-safe initialization can be guaranteed.
+  #
   # @note Thread safe guarantees can only be made when asynchronous method calls
   #       are not mixed with synchronous method calls. Use only synchronous calls
   #       when the object is used exclusively on a single thread. Use only
@@ -142,7 +160,7 @@ def method_missing(method, *args, &block)
           ivar = Concurrent::IVar.new
           value, reason = nil, nil
           begin
-            mutex.synchronize do
+            @mutex.synchronize do
               value = @delegate.send(method, *args, &block)
             end
           rescue => reason
@@ -154,11 +172,6 @@ def method_missing(method, *args, &block)
 
         self.send(method, *args)
       end
-
-      # The lock used when delegating methods to the wrapped object.
-      #
-      # @!visibility private
-      attr_reader :mutex # :nodoc:
     end
 
     # Delegates asynchronous, thread-safe method calls to the wrapped object.
@@ -194,22 +207,15 @@ def method_missing(method, *args, &block)
 
         self.define_singleton_method(method) do |*args|
           Async::validate_argc(@delegate, method, *args)
-          Concurrent::Future.execute(executor: @executor) do
-            mutex.synchronize do
+          Concurrent::Future.execute(executor: @executor.value) do
+            @mutex.synchronize do
               @delegate.send(method, *args, &block)
             end
           end
         end
 
         self.send(method, *args)
       end
-
-      private
-
-      # The lock used when delegating methods to the wrapped object.
-      #
-      # @!visibility private
-      attr_reader :mutex # :nodoc:
     end
 
     # Causes the chained method call to be performed asynchronously on the
@@ -230,17 +236,19 @@ def method_missing(method, *args, &block)
     #   either `async` or `await`. The mutable nature of Ruby references
     #   (and object orientation in general) prevent any other thread safety
     #   guarantees. Do NOT mix non-protected method calls with protected
-    #   method call. Use ONLY protected method calls when sharing the object
+    #   method call. Use *only* protected method calls when sharing the object
     #   between threads.
     #
     # @return [Concurrent::Future] the pending result of the asynchronous operation
     #
+    # @raise [Concurrent::InitializationError] `#init_mutex` has not been called
     # @raise [NameError] the object does not respond to `method` method
     # @raise [ArgumentError] the given `args` do not match the arity of `method`
     #
     # @see Concurrent::Future
     def async
-      @__async_delegator__ ||= AsyncDelegator.new(self, executor, await.mutex)
+      raise InitializationError.new('#init_mutex was never called') unless @mutex
+      @__async_delegator__.value
     end
     alias_method :future, :async
 
@@ -262,29 +270,51 @@ def async
     #   either `async` or `await`. The mutable nature of Ruby references
     #   (and object orientation in general) prevent any other thread safety
     #   guarantees. Do NOT mix non-protected method calls with protected
-    #   method call. Use ONLY protected method calls when sharing the object
+    #   method call. Use *only* protected method calls when sharing the object
     #   between threads.
     #
     # @return [Concurrent::IVar] the completed result of the synchronous operation
     #
+    # @raise [Concurrent::InitializationError] `#init_mutex` has not been called
     # @raise [NameError] the object does not respond to `method` method
     # @raise [ArgumentError] the given `args` do not match the arity of `method`
     #
     # @see Concurrent::IVar
     def await
-      @__await_delegator__ ||= AwaitDelegator.new(self, Mutex.new)
+      raise InitializationError.new('#init_mutex was never called') unless @mutex
+      @__await_delegator__.value
     end
     alias_method :delay, :await
 
+    # Set a new executor
+    #
+    # @raise [Concurrent::InitializationError] `#init_mutex` has not been called
+    # @raise [ArgumentError] executor has already been set
     def executor=(executor)
-      raise ArgumentError.new('executor has already been set') unless @__async__executor__.nil?
-      @__async__executor__ = executor
+      raise InitializationError.new('#init_mutex was never called') unless @mutex
+      @__async__executor__.reconfigure { executor } or
+        raise ArgumentError.new('executor has already been set')
     end
 
-    private
-
-    def executor
-      @__async__executor__ ||= Concurrent.configuration.global_task_pool
+    # Initialize the internal mutex and other synchronization objects. This method
+    # *must* be called from the constructor of the including class or explicitly
+    # by the caller prior to calling any other methods. If `init_mutex` is *not*
+    # called explicitly the async/await/executor methods will raize a
+    # `Concurrent::InitializationError`. This is the only way thread-safe
+    # initialization can be guaranteed.
+    #
+    # @note This method *must* be called from the constructor of the including
+    #       class or explicitly by the caller prior to calling any other methods.
+    #       This is the only way thread-safe initialization can be guaranteed.
+    #
+    # @raise [Concurrent::InitializationError] when called more than once
+    def init_mutex
+      raise InitializationError.new('#init_mutex was already called') if @mutex
+      (@mutex = Mutex.new).lock
+      @__async__executor__ = Delay.new{ Concurrent.configuration.global_operation_pool }
+      @__await_delegator__ = Delay.new{ AwaitDelegator.new(self, @mutex) }
+      @__async_delegator__ = Delay.new{ AsyncDelegator.new(self, @__async__executor__, @mutex) }
+      @mutex.unlock
     end
   end
 end

spec/concurrent/async_spec.rb

@@ -10,6 +10,9 @@ module Concurrent
       Class.new do
         include Concurrent::Async
         attr_accessor :accessor
+        def initialize
+          init_mutex
+        end
         def echo(msg)
           msg
         end
@@ -103,7 +106,7 @@ def many(*args, &block) nil; end
     context 'executor' do
 
       it 'returns the default executor when #executor= has never been called' do
-        Concurrent.configuration.should_receive(:global_task_pool).
+        Concurrent.configuration.should_receive(:global_operation_pool).
           and_return(ImmediateExecutor.new)
         subject = async_class.new
         subject.async.echo(:foo)
@@ -117,10 +120,10 @@ def many(*args, &block) nil; end
         subject.async.echo(:foo)
       end
 
-      it 'raises an exception if #executor= is called multiple times' do
+      it 'raises an exception if #executor= is called after initialization complete' do
         executor = ImmediateExecutor.new
         subject = async_class.new
-        subject.executor = executor
+        subject.async.echo(:foo)
         expect {
           subject.executor = executor
         }.to raise_error(ArgumentError)
@@ -303,6 +306,7 @@ def many(*args, &block) nil; end
         object = Class.new {
           include Concurrent::Async
           attr_reader :bucket
+          def initialize() init_mutex; end
           def gather(seconds, first, *rest)
             sleep(seconds)
             (@bucket ||= []).concat([first])
@@ -315,6 +319,34 @@ def gather(seconds, first, *rest)
         object.await.gather(0, :c, :d)
         object.bucket.should eq [:a, :b, :c, :d]
       end
+
+      context 'raises an InitializationError' do
+
+        let(:async_class) do
+          Class.new do
+            include Concurrent::Async
+            def echo(msg) msg; end
+          end
+        end
+
+        it 'when #async is called before #init_mutex' do
+          expect {
+            async_class.new.async.echo(:foo)
+          }.to raise_error(Concurrent::InitializationError)
+        end
+
+        it 'when #await is called before #init_mutex' do
+          expect {
+            async_class.new.async.echo(:foo)
+          }.to raise_error(Concurrent::InitializationError)
+        end
+
+        it 'when #executor= is called before #init_mutex' do
+          expect {
+            async_class.new.executor = Concurrent::ImmediateExecutor.new
+          }.to raise_error(Concurrent::InitializationError)
+        end
+      end
     end
   end
 end