Merged PR #69: Do not create another thread for each post call to Agent

Author: jdantonio

lib/concurrent/agent.rb

@@ -38,19 +38,13 @@ class Agent
     include Concurrent::Observable
     include Logging
 
-    # The default timeout value (in seconds); used when no timeout option
-    # is given at initialization
-    TIMEOUT = 5
-
     attr_reader :timeout, :task_executor, :operation_executor
 
     # Initialize a new Agent with the given initial value and provided options.
     #
     # @param [Object] initial the initial value
     # @param [Hash] opts the options used to define the behavior at update and deref
     #
-    # @option opts [Fixnum] :timeout (TIMEOUT) maximum number of seconds before an update is cancelled
-    #
     # @option opts [Boolean] :operation (false) when `true` will execute the future on the global
     #   operation pool (for long-running operations), when `false` will execute the future on the
     #   global task pool (for short-running tasks)
@@ -65,7 +59,6 @@ def initialize(initial, opts = {})
       @value                = initial
       @rescuers             = []
       @validator            = Proc.new { |result| true }
-      @timeout              = opts.fetch(:timeout, TIMEOUT).freeze
       self.observers        = CopyOnWriteObserverSet.new
       @serialized_execution = SerializedExecution.new
       @task_executor        = OptionsParser.get_task_executor_from(opts)
@@ -145,12 +138,19 @@ def post(&block)
     # Update the current value with the result of the given block operation,
     # block can do blocking calls
     #
+    # @param [Fixnum, nil] timeout maximum number of seconds before an update is cancelled
+    #
     # @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)
+    def post_off(timeout = nil, &block)
+      block = if timeout
+                lambda { |value| Concurrent::timeout(timeout) { block.call(value) } }
+              else
+                block
+              end
       post_on(@operation_executor, &block)
     end
 
@@ -203,10 +203,8 @@ def work(&handler) # :nodoc:
       validator, value = mutex.synchronize { [@validator, @value] }
 
       begin
-        result, valid = Concurrent::timeout(@timeout) do
-          result = handler.call(value)
-          [result, validator.call(result)]
-        end
+        result = handler.call(value)
+        valid  = validator.call(result)
       rescue Exception => ex
         exception = ex
       end

spec/concurrent/agent_spec.rb

@@ -19,7 +19,7 @@ module Concurrent
       end.new
     end
 
-    context '#send_off' do
+    context '#post_off' do
       subject { Agent.new 2, executor: executor }
 
       it 'executes post and post-off in order' do
@@ -28,6 +28,14 @@ module Concurrent
         subject.await
         expect(subject.value).to eq 12
       end
+
+      it 'times out' do
+        ex = nil
+        subject.post_off(0.1) { |v| sleep(0.2); ex = true }
+        subject.await
+        sleep 0.3
+        expect(ex).to eq nil
+      end
     end
 
     context 'behavior' do
@@ -71,14 +79,6 @@ def trigger_observable(observable)
         expect(Agent.new(10).value).to eq 10
       end
 
-      it 'sets the timeout to the given value' do
-        expect(Agent.new(0, timeout: 5).timeout).to eq 5
-      end
-
-      it 'sets the timeout to the default when nil' do
-        expect(Agent.new(0).timeout).to eq Agent::TIMEOUT
-      end
-
       it 'uses the executor given with the :executor option' do
         expect(executor).to receive(:post).with(any_args).and_return(0)
         agent = Agent.new(0, executor: executor)
@@ -164,9 +164,9 @@ def trigger_observable(observable)
         subject.post { nil }
         sleep(0.1)
         expect(subject.
-            instance_variable_get(:@serialized_execution).
-            instance_variable_get(:@stash).
-            size).to eq 2
+                   instance_variable_get(:@serialized_execution).
+                   instance_variable_get(:@stash).
+                   size).to eq 2
       end
 
       it 'does not add to the queue when no block is given' do
@@ -221,7 +221,7 @@ def trigger_observable(observable)
       it 'passes the current value to the handler' do
         latch = Concurrent::CountDownLatch.new(5)
         Agent.new(latch.count, executor: executor).post do |i|
-          i.times{ latch.count_down }
+          i.times { latch.count_down }
         end
         expect(latch.wait(1)).to be_truthy
       end
@@ -252,7 +252,7 @@ def trigger_observable(observable)
 
       it 'passes the new value to the validator' do
         expected = Concurrent::AtomicFixnum.new(0)
-        latch = Concurrent::CountDownLatch.new(1)
+        latch    = Concurrent::CountDownLatch.new(1)
         subject.validate { |v| expected.value = v; latch.count_down; true }
         subject.post { 10 }
         latch.wait(1)
@@ -282,7 +282,7 @@ def trigger_observable(observable)
 
       it 'calls the first exception block with a matching class' do
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(StandardError) { |ex| expected = 1 }.
             rescue(StandardError) { |ex| expected = 2 }.
             rescue(StandardError) { |ex| expected = 3 }
@@ -292,7 +292,7 @@ def trigger_observable(observable)
 
       it 'matches all with a rescue with no class given' do
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(LoadError) { |ex| expected = 1 }.
             rescue { |ex| expected = 2 }.
             rescue(StandardError) { |ex| expected = 3 }
@@ -302,23 +302,23 @@ def trigger_observable(observable)
 
       it 'searches associated rescue handlers in order' do
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(ArgumentError) { |ex| expected = 1 }.
             rescue(LoadError) { |ex| expected = 2 }.
             rescue(StandardError) { |ex| expected = 3 }
         agent.post { raise ArgumentError }
         expect(expected).to eq 1
 
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(ArgumentError) { |ex| expected = 1 }.
             rescue(LoadError) { |ex| expected = 2 }.
             rescue(StandardError) { |ex| expected = 3 }
         agent.post { raise LoadError }
         expect(expected).to eq 2
 
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(ArgumentError) { |ex| expected = 1 }.
             rescue(LoadError) { |ex| expected = 2 }.
             rescue(StandardError) { |ex| expected = 3 }
@@ -328,7 +328,7 @@ def trigger_observable(observable)
 
       it 'passes the exception object to the matched block' do
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(ArgumentError) { |ex| expected = ex }.
             rescue(LoadError) { |ex| expected = ex }.
             rescue(StandardError) { |ex| expected = ex }
@@ -338,7 +338,7 @@ def trigger_observable(observable)
 
       it 'ignores rescuers without a block' do
         expected = nil
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new).
             rescue(StandardError).
             rescue(StandardError) { |ex| expected = ex }
         agent.post { raise StandardError }
@@ -464,15 +464,15 @@ def trigger_observable(observable)
       end
 
       it 'aliases #catch for #rescue' do
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new)
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new)
         expected = nil
         agent.catch { expected = true }
         agent.post { raise StandardError }
         expect(agent).to be_truthy
       end
 
       it 'aliases #on_error for #rescue' do
-        agent = Agent.new(0, executor: Concurrent::ImmediateExecutor.new)
+        agent    = Agent.new(0, executor: Concurrent::ImmediateExecutor.new)
         expected = nil
         agent.on_error { expected = true }
         agent.post { raise StandardError }