Allow dereferencing of Agent values while updating

ensure only one update is running

Author: pitr-ch

lib/concurrent/agent.rb

@@ -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 = OptionsParser::get_executor_from(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          = []
       init_mutex
       set_deref_options(opts)
     end
@@ -111,7 +113,11 @@ def rescue(clazz = StandardError, &block)
     # @yieldparam [Object] value the result of the last update operation
     # @yieldreturn [Boolean] true if the value is valid else false
     def validate(&block)
-      @validator = block unless block.nil?
+      unless block.nil?
+        mutex.lock
+        @validator = block
+        mutex.unlock
+      end
       self
     end
     alias_method :validates, :validate
@@ -124,8 +130,19 @@ def validate(&block)
     #   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)
-      @executor.post{ work(&block) } unless block.nil?
+      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
+      true
     end
 
     # Update the current value with the result of the given block operation
@@ -157,22 +174,38 @@ def try_rescue(ex) # :nodoc:
     # @!visibility private
     def work(&handler) # :nodoc:
       begin
-
-        should_notify = false
+        should_notify    = false
+        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
+        end
 
         mutex.synchronize do
-          result = Concurrent::timeout(@timeout) do
-            handler.call(@value)
-          end
-          if @validator.call(result)
-            @value = result
+          if !exception && valid
+            @value        = result
             should_notify = true
           end
+
+          if (stashed = @stash.shift)
+            @executor.post { work(&stashed) }
+          else
+            @being_executed = false
+          end
         end
-        time = Time.now
-        observers.notify_observers{ [time, self.value] } if should_notify
-      rescue Exception => ex
-        try_rescue(ex)
+
+        if should_notify
+          time = Time.now
+          observers.notify_observers { [time, self.value] }
+        end
+
+        try_rescue(exception)
       end
     end
   end

spec/concurrent/agent_spec.rb

@@ -147,18 +147,17 @@ def trigger_observable(observable)
     context '#post' do
 
       it 'adds the given block to the queue' do
-        executor.should_receive(:post).with(no_args).exactly(3).times
-        subject.post { sleep(100) }
+        executor.should_receive(:post).with(no_args).exactly(1).times
+        subject.post { sleep(1) }
         subject.post { nil }
         subject.post { nil }
         sleep(0.1)
+        subject.instance_variable_get(:@stash).size.should eq 2
       end
 
       it 'does not add to the queue when no block is given' do
-        executor.should_receive(:post).with(no_args).exactly(2).times
-        subject.post { sleep(100) }
+        executor.should_receive(:post).with(no_args).exactly(0).times
         subject.post
-        subject.post { nil }
         sleep(0.1)
       end
     end
@@ -365,6 +364,46 @@ def trigger_observable(observable)
       end
     end
 
+    context 'clojure-like behaviour' do
+      it 'does not block dereferencing when updating the value' do
+        continue = IVar.new
+        agent    = Agent.new(0, executor: executor)
+        agent.post { |old| old + continue.value }
+        sleep 0.1
+        Concurrent.timeout(0.2) { agent.value.should eq 0 }
+        continue.set 1
+        sleep 0.1
+      end
+
+      it 'does not allow to execute two updates at the same time' do
+        agent     = Agent.new(0, executor: executor)
+        continue1 = IVar.new
+        continue2 = IVar.new
+        f1        = f2 = false
+        agent.post { |old| f1 = true; old + continue1.value }
+        agent.post { |old| f2 = true; old + continue2.value }
+
+        sleep 0.1
+        f1.should eq true
+        f2.should eq false
+        agent.value.should eq 0
+
+        continue1.set 1
+        sleep 0.1
+        f1.should eq true
+        f2.should eq true
+        agent.value.should eq 1
+
+        continue2.set 1
+        sleep 0.1
+        f1.should eq true
+        f2.should eq true
+        agent.value.should eq 2
+      end
+
+      it 'waits with sending functions to other agents until update is done'
+    end
+
     context 'aliases' do
 
       it 'aliases #deref for #value' do