Use global_operation_pool for jobs which may block

Author: pitr-ch

lib/concurrent/dataflow.rb

@@ -61,7 +61,7 @@ def update(time, value, reason)
   # @raise [ArgumentError] if no block is given
   # @raise [ArgumentError] if any of the inputs are not `IVar`s
   def dataflow(*inputs, &block)
-    dataflow_with(Concurrent.configuration.global_task_pool, *inputs, &block)
+    dataflow_with(Concurrent.configuration.global_operation_pool, *inputs, &block)
   end
   module_function :dataflow
 

lib/concurrent/executor/timer_set.rb

@@ -23,7 +23,7 @@ class TimerSet
     #   this executor rather than the global thread pool (overrides :operation)
     def initialize(opts = {})
       @queue = PriorityQueue.new(order: :min)
-      @task_executor = OptionsParser::get_executor_from(opts)
+      @task_executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_task_pool
       @timer_executor = SingleThreadExecutor.new
       @condition = Condition.new
       init_executor

lib/concurrent/future.rb

@@ -61,7 +61,7 @@ def initialize(opts = {}, &block)
       super(IVar::NO_VALUE, opts)
       @state = :unscheduled
       @task = block
-      @executor = OptionsParser::get_executor_from(opts)
+      @executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
     end
 
     # Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and

lib/concurrent/options_parser.rb

@@ -10,14 +10,16 @@ module OptionsParser
     # @option opts [Boolean] :operation (`false`) when true use the global operation pool
     # @option opts [Boolean] :task (`true`) when true use the global task pool
     #
-    # @return [Executor] the requested thread pool (default: global task pool)
+    # @return [Executor, nil] the requested thread pool, or nil when no option specified
     def get_executor_from(opts = {})
       if opts[:executor]
         opts[:executor]
       elsif opts[:operation] == true || opts[:task] == false
         Concurrent.configuration.global_operation_pool
-      else
+      elsif opts[:operation] == false || opts[:task] == true
         Concurrent.configuration.global_task_pool
+      else
+        nil
       end
     end
 

lib/concurrent/promise.rb

@@ -5,6 +5,7 @@
 
 module Concurrent
 
+  # TODO unify promise and future to single class, with dataflow
   class Promise
     include Obligation
 
@@ -32,7 +33,7 @@ class Promise
     def initialize(opts = {}, &block)
       opts.delete_if { |k, v| v.nil? }
 
-      @executor = OptionsParser::get_executor_from(opts)
+      @executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
       @parent = opts.fetch(:parent) { nil }
       @on_fulfill = opts.fetch(:on_fulfill) { Proc.new { |result| result } }
       @on_reject = opts.fetch(:on_reject) { Proc.new { |reason| raise reason } }

lib/concurrent/scheduled_task.rb

@@ -15,16 +15,17 @@ def initialize(intended_time, opts = {}, &block)
       super(NO_VALUE, opts)
 
       self.observers = CopyOnNotifyObserverSet.new
-      @intended_time =  intended_time
-      @state = :unscheduled
-      @task = block
+      @intended_time = intended_time
+      @state         = :unscheduled
+      @task          = block
+      @executor      = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
     end
 
     # @since 0.5.0
     def execute
       if compare_and_set_state(:pending, :unscheduled)
         @schedule_time = TimerSet.calculate_schedule_time(@intended_time)
-        Concurrent::timer(@schedule_time.to_f - Time.now.to_f, &method(:process_task))
+        Concurrent::timer(@schedule_time.to_f - Time.now.to_f) { @executor.post &method(:process_task) }
         self
       end
     end
@@ -71,7 +72,7 @@ def process_task
         end
 
         time = Time.now
-        observers.notify_and_delete_observers{ [time, self.value, reason] }
+        observers.notify_and_delete_observers { [time, self.value, reason] }
       end
     end
   end

spec/concurrent/dataflow_spec.rb

@@ -12,10 +12,10 @@ module Concurrent
       expect { Concurrent::dataflow_with(root_executor) }.to raise_error(ArgumentError)
     end
 
-    specify '#dataflow uses the global task pool' do
+    specify '#dataflow uses the global operation pool' do
       input = Future.execute{0}
       expect(Concurrent).to receive(:dataflow_with).once.
-        with(Concurrent.configuration.global_task_pool, input)
+        with(Concurrent.configuration.global_operation_pool, input)
       Concurrent::dataflow(input){0}
     end
 
@@ -229,9 +229,9 @@ def fib_with_dot(n)
           end
         end
 
-        expected = fib_with_dot(14)
+        expected = fib_with_dot(7)
         sleep(0.1)
-        expect(expected.value).to eq 377
+        expect(expected.value).to eq 13
       end
 
     end

spec/concurrent/future_spec.rb

@@ -114,8 +114,8 @@ def trigger_observable(observable)
         Future.execute(task: true){ nil }
       end
 
-      it 'uses the global task pool by default' do
-        expect(Concurrent.configuration).to receive(:global_task_pool).and_return(executor)
+      it 'uses the global operation pool by default' do
+        expect(Concurrent.configuration).to receive(:global_operation_pool).and_return(executor)
         Future.execute{ nil }
       end
     end

spec/concurrent/options_parser_spec.rb

@@ -39,16 +39,12 @@ module Concurrent
         OptionsParser::get_executor_from(task: true)
       end
 
-      it 'returns the global task pool when :executor is nil' do
-        expect(Concurrent.configuration).to receive(:global_task_pool).
-          and_return(:task_pool)
-        OptionsParser::get_executor_from(executor: nil)
+      it 'returns nil when :executor is nil' do
+        expect(OptionsParser::get_executor_from(executor: nil)).to be_nil
       end
 
-      it 'returns the global task pool when no option is given' do
-        expect(Concurrent.configuration).to receive(:global_task_pool).
-          and_return(:task_pool)
-        OptionsParser::get_executor_from
+      it 'returns nil task pool when no option is given' do
+        expect(OptionsParser::get_executor_from).to be_nil
       end
 
       specify ':executor overrides :operation' do