Merge pull request #71 from pitr-ch/actor

Support `send` and `send-off` methods for Agent

Author: Petr Chalupa

lib/concurrent/actor/simple_actor_ref.rb

@@ -9,20 +9,21 @@ class SimpleActorRef
     include ActorRef
 
     def initialize(actor, opts = {})
-      @actor = actor
-      @mutex = Mutex.new
-      @executor = OneByOne.new OptionsParser::get_executor_from(opts)
-      @stop_event = Event.new
-      @reset_on_error = opts.fetch(:reset_on_error, true)
+      @actor           = actor
+      @mutex           = Mutex.new
+      @one_by_one      = OneByOne.new
+      @executor        = OptionsParser::get_executor_from(opts)
+      @stop_event      = Event.new
+      @reset_on_error  = opts.fetch(:reset_on_error, true)
       @exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError
-      @args = opts.fetch(:args, []) if @reset_on_error
+      @args            = opts.fetch(:args, []) if @reset_on_error
 
       @actor.define_singleton_method(:shutdown, &method(:set_stop_event))
       @actor.on_start
     end
 
     def running?
-      ! @stop_event.set?
+      not @stop_event.set?
     end
 
     def shutdown?
@@ -32,7 +33,7 @@ def shutdown?
     def post(*msg, &block)
       raise ArgumentError.new('message cannot be empty') if msg.empty?
       ivar = IVar.new
-      @executor.post(Message.new(msg, ivar, block), &method(:process_message))
+      @one_by_one.post(@executor, Message.new(msg, ivar, block), &method(:process_message))
       ivar
     end
 

lib/concurrent/agent.rb

@@ -40,7 +40,7 @@ class Agent
     # is given at initialization
     TIMEOUT = 5
 
-    attr_reader :timeout, :executor
+    attr_reader :timeout, :task_executor, :operation_executor
 
     # Initialize a new Agent with the given initial value and provided options.
     #
@@ -60,12 +60,14 @@ class Agent
     # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
     #   returning the value returned from the proc
     def initialize(initial, opts = {})
-      @value         = initial
-      @rescuers      = []
-      @validator     = Proc.new { |result| true }
-      @timeout       = opts.fetch(:timeout, TIMEOUT).freeze
-      self.observers = CopyOnWriteObserverSet.new
-      @executor      = OneByOne.new OptionsParser::get_executor_from(opts)
+      @value              = initial
+      @rescuers           = []
+      @validator          = Proc.new { |result| true }
+      @timeout            = opts.fetch(:timeout, TIMEOUT).freeze
+      self.observers      = CopyOnWriteObserverSet.new
+      @one_by_one         = OneByOne.new
+      @task_executor      = OptionsParser.get_task_executor_from(opts)
+      @operation_executor = OptionsParser.get_operation_executor_from(opts)
       init_mutex
       set_deref_options(opts)
     end
@@ -122,27 +124,39 @@ def validate(&block)
     alias_method :validate_with, :validate
     alias_method :validates_with, :validate
 
-    # Update the current value with the result of the given block operation
+    # Update the current value with the result of the given block operation,
+    # block should not do blocking calls, use #post_off for blocking calls
     #
     # @yield the operation to be performed with the current value in order to calculate
     #   the new value
     # @yieldparam [Object] value the current value
     # @yieldreturn [Object] the new value
     # @return [true, nil] nil when no block is given
     def post(&block)
-      return nil if block.nil?
-      @executor.post { work(&block) }
-      true
+      post_on(@task_executor, &block)
+    end
+
+    # Update the current value with the result of the given block operation,
+    # block can do blocking calls
+    #
+    # @yield the operation to be performed with the current value in order to calculate
+    #   the new value
+    # @yieldparam [Object] value the current value
+    # @yieldreturn [Object] the new value
+    # @return [true, nil] nil when no block is given
+    def post_off(&block)
+      post_on(@operation_executor, &block)
     end
 
-    # Update the current value with the result of the given block operation
+    # Update the current value with the result of the given block operation,
+    # block should not do blocking calls, use #post_off for blocking calls
     #
     # @yield the operation to be performed with the current value in order to calculate
     #   the new value
     # @yieldparam [Object] value the current value
     # @yieldreturn [Object] the new value
     def <<(block)
-      self.post(&block)
+      post(&block)
       self
     end
 
@@ -152,12 +166,18 @@ def <<(block)
     # @return [Boolean] false on timeout, true otherwise
     def await(timeout = nil)
       done = Event.new
-      post { done.set }
+      post { |val| done.set; val }
       done.wait timeout
     end
 
     private
 
+    def post_on(executor, &block)
+      return nil if block.nil?
+      @one_by_one.post(executor) { work(&block) }
+      true
+    end
+
     # @!visibility private
     Rescuer = Struct.new(:clazz, :block) # :nodoc:
 
@@ -168,7 +188,6 @@ def try_rescue(ex) # :nodoc:
       end
       rescuer.block.call(ex) if rescuer
     rescue Exception => ex
-      # puts "#{ex} (#{ex.class})\n#{ex.backtrace.join("\n")}"
       # supress
     end
 
@@ -179,8 +198,8 @@ def work(&handler) # :nodoc:
       begin
         # FIXME creates second thread
         result, valid = Concurrent::timeout(@timeout) do
-          [result = handler.call(value),
-           validator.call(result)]
+          result = handler.call(value)
+          [result, validator.call(result)]
         end
       rescue Exception => ex
         exception = ex

lib/concurrent/executor/one_by_one.rb

@@ -4,24 +4,22 @@ module Concurrent
   # never running at the same time.
   class OneByOne
 
-    attr_reader :executor
-
-    Job = Struct.new(:args, :block) do
+    Job = Struct.new(:executor, :args, :block) do
       def call
         block.call *args
       end
     end
 
-    # @param [Executor] executor
-    def initialize(executor)
-      @executor       = executor
+    def initialize
       @being_executed = false
       @stash          = []
       @mutex          = Mutex.new
     end
 
     # Submit a task to the executor for asynchronous processing.
     #
+    # @param [Executor] executor to be used for this job
+    #
     # @param [Array] args zero or more arguments to be passed to the task
     #
     # @yield the asynchronous task to perform
@@ -30,9 +28,10 @@ def initialize(executor)
     #   is not running
     #
     # @raise [ArgumentError] if no task is given
-    def post(*args, &task)
+    def post(executor, *args, &task)
       return nil if task.nil?
-      job = Job.new args, task
+      job = Job.new executor, args, task
+
       @mutex.lock
       post = if @being_executed
                @stash << job
@@ -41,30 +40,25 @@ def post(*args, &task)
                @being_executed = true
              end
       @mutex.unlock
-      @executor.post { work(job) } if post
-      true
-    end
 
-    # Submit a task to the executor for asynchronous processing.
-    #
-    # @param [Proc] task the asynchronous task to perform
-    #
-    # @return [self] returns itself
-    def <<(task)
-      post(&task)
-      self
+      call_job job if post
+      true
     end
 
     private
 
+    def call_job(job)
+      job.executor.post { work(job) }
+    end
+
     # ensures next job is executed if any is stashed
     def work(job)
       job.call
     ensure
       @mutex.lock
       job = @stash.shift || (@being_executed = false)
       @mutex.unlock
-      @executor.post { work(job) } if job
+      call_job job if job
     end
 
   end

lib/concurrent/options_parser.rb

@@ -20,6 +20,27 @@ def get_executor_from(opts = {})
         Concurrent.configuration.global_task_pool
       end
     end
-    module_function :get_executor_from
+
+    # Get the requested `Executor` based on the values set in the options hash.
+    #
+    # @param [Hash] opts the options defining the requested executor
+    # @option opts [Executor] :task_executor (`nil`) when set use the given `Executor` instance
+    #
+    # @return [Executor] the requested thread pool (default: global task pool)
+    def get_task_executor_from(opts = {})
+      opts[:task_executor] || opts[:executor] || Concurrent.configuration.global_task_pool
+    end
+
+    # Get the requested `Executor` based on the values set in the options hash.
+    #
+    # @param [Hash] opts the options defining the requested executor
+    # @option opts [Executor] :task_executor (`nil`) when set use the given `Executor` instance
+    #
+    # @return [Executor] the requested thread pool (default: global operation pool)
+    def get_operation_executor_from(opts = {})
+      opts[:operation_executor] || opts[:executor] || Concurrent.configuration.global_operation_pool
+    end
+
+    extend self
   end
 end

spec/concurrent/agent_spec.rb

@@ -19,6 +19,17 @@ module Concurrent
       end.new
     end
 
+    context '#send_off' do
+      subject { Agent.new 2 }
+
+      it 'executes post and post-off in order' do
+        subject.post { |v| v + 2 }
+        subject.post_off { |v| v * 3 }
+        subject.await
+        subject.value.should eq 12
+      end
+    end
+
     context 'behavior' do
 
       # dereferenceable
@@ -153,7 +164,7 @@ def trigger_observable(observable)
         subject.post { nil }
         sleep(0.1)
         subject.
-            executor.
+            instance_variable_get(:@one_by_one).
             instance_variable_get(:@stash).
             size.should eq 2
       end
@@ -189,6 +200,12 @@ def trigger_observable(observable)
         fn.should be_false
       end
 
+      it 'does not alter the value' do
+        subject.post { |v| v + 1 }
+        subject.await
+        subject.value.should eq 1
+      end
+
     end
 
     context 'fulfillment' do