Extract OneByOne behavior from Agent

it can be also later used for Actors

Author: pitr-ch

lib/concurrent/agent.rb

@@ -40,7 +40,7 @@ class Agent
     # is given at initialization
     TIMEOUT = 5
 
-    attr_reader :timeout
+    attr_reader :timeout, :executor
 
     # Initialize a new Agent with the given initial value and provided options.
     #
@@ -60,14 +60,12 @@ 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       = OptionsParser::get_executor_from(opts)
-      @being_executed = false
-      @stash          = []
+      @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)
       init_mutex
       set_deref_options(opts)
     end
@@ -133,15 +131,7 @@ def validate(&block)
     # @return [true, nil] nil when no block is given
     def post(&block)
       return nil if block.nil?
-      mutex.lock
-      post = if @being_executed
-               @stash << block
-               false
-             else
-               @being_executed = true
-             end
-      mutex.unlock
-      @executor.post { work(&block) } if post
+      @executor.post { work(&block) }
       true
     end
 
@@ -184,36 +174,31 @@ def try_rescue(ex) # :nodoc:
 
     # @!visibility private
     def work(&handler) # :nodoc:
+      validator, value = mutex.synchronize { [@validator, @value] }
+
       begin
-        validator, value = mutex.synchronize { [@validator, @value] }
-
-        begin
-          # FIXME creates second thread
-          result, valid = Concurrent::timeout(@timeout) do
-            [result = handler.call(value),
-             validator.call(result)]
-          end
-        rescue Exception => ex
-          exception = ex
+        # FIXME creates second thread
+        result, valid = Concurrent::timeout(@timeout) do
+          [result = handler.call(value),
+           validator.call(result)]
         end
+      rescue Exception => ex
+        exception = ex
+      end
 
-        mutex.lock
-        should_notify = if !exception && valid
-                          @value = result
-                          true
-                        end
-        stashed = @stash.shift || (@being_executed = false)
-        mutex.unlock
-
-        @executor.post { work(&stashed) } if stashed
-
-        if should_notify
-          time = Time.now
-          observers.notify_observers { [time, self.value] }
-        end
+      mutex.lock
+      should_notify = if !exception && valid
+                        @value = result
+                        true
+                      end
+      mutex.unlock
 
-        try_rescue(exception)
+      if should_notify
+        time = Time.now
+        observers.notify_observers { [time, self.value] }
       end
+
+      try_rescue(exception)
     end
   end
 end

lib/concurrent/executor/one_by_one.rb

@@ -0,0 +1,71 @@
+module Concurrent
+
+  # Ensures that jobs are passed to the underlying executor one by one,
+  # never running at the same time.
+  class OneByOne
+
+    attr_reader :executor
+
+    Job = Struct.new(:args, :block) do
+      def call
+        block.call *args
+      end
+    end
+
+    # @param [Executor] executor
+    def initialize(executor)
+      @executor       = executor
+      @being_executed = false
+      @stash          = []
+      @mutex          = Mutex.new
+    end
+
+    # Submit a task to the executor for asynchronous processing.
+    #
+    # @param [Array] args zero or more arguments to be passed to the task
+    #
+    # @yield the asynchronous task to perform
+    #
+    # @return [Boolean] `true` if the task is queued, `false` if the executor
+    #   is not running
+    #
+    # @raise [ArgumentError] if no task is given
+    def post(*args, &task)
+      return nil if task.nil?
+      job = Job.new args, task
+      @mutex.lock
+      post = if @being_executed
+               @stash << job
+               false
+             else
+               @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
+    end
+
+    private
+
+    # 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
+    end
+
+  end
+end

lib/concurrent/executors.rb

@@ -6,3 +6,4 @@
 require 'concurrent/executor/single_thread_executor'
 require 'concurrent/executor/thread_pool_executor'
 require 'concurrent/executor/timer_set'
+require 'concurrent/executor/one_by_one'

spec/concurrent/agent_spec.rb

@@ -152,7 +152,10 @@ def trigger_observable(observable)
         subject.post { nil }
         subject.post { nil }
         sleep(0.1)
-        subject.instance_variable_get(:@stash).size.should eq 2
+        subject.
+            executor.
+            instance_variable_get(:@stash).
+            size.should eq 2
       end
 
       it 'does not add to the queue when no block is given' do