Added Concurrent::dataflow_with for explicitly setting the executor.

Author: jdantonio

lib/concurrent/dataflow.rb

@@ -61,10 +61,16 @@ 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)
+  end
+  module_function :dataflow
+
+  def dataflow_with(executor, *inputs, &block)
+    raise ArgumentError.new('an executor must be provided') if executor.nil?
     raise ArgumentError.new('no block given') unless block_given?
     raise ArgumentError.new('not all dependencies are IVars') unless inputs.all? { |input| input.is_a? IVar }
 
-    result = Future.new do
+    result = Future.new(executor: executor) do
       values = inputs.map { |input| input.value }
       block.call(*values)
     end
@@ -81,6 +87,5 @@ def dataflow(*inputs, &block)
 
     result
   end
-
-  module_function :dataflow
+  module_function :dataflow_with
 end

spec/concurrent/dataflow_spec.rb

@@ -5,43 +5,77 @@ module Concurrent
   describe 'dataflow' do
 
     let(:executor) { ImmediateExecutor.new }
-
-    before(:each) do
-      Concurrent.configure do |config|
-        config.global_task_pool = Concurrent::PerThreadExecutor.new
-      end
-    end
+    let(:root_executor) { PerThreadExecutor.new }
 
     it 'raises an exception when no block given' do
       expect { Concurrent::dataflow }.to raise_error(ArgumentError)
+      expect { Concurrent::dataflow_with(root_executor) }.to raise_error(ArgumentError)
+    end
+
+    specify '#dataflow uses the global task pool' do
+      input = Future.execute{0}
+      Concurrent.should_receive(:dataflow_with).once.
+        with(Concurrent.configuration.global_task_pool, input)
+      Concurrent::dataflow(input){0}
+    end
+
+    specify '#dataflow_with uses the given executor' do
+      input = Future.execute{0}
+      result = Future.new{0}
+
+      Future.should_receive(:new).with(executor: root_executor).and_return(result)
+      Concurrent::dataflow_with(root_executor, input){0}
+    end
+
+    specify '#dataflow_with raises an exception when no executor given' do
+      expect {
+        Concurrent::dataflow_with(nil){ nil }
+      }.to raise_error(ArgumentError)
     end
 
     it 'accepts zero or more dependencies' do
       Concurrent::dataflow(){0}
       Concurrent::dataflow(Future.execute{0}){0}
       Concurrent::dataflow(Future.execute{0}, Future.execute{0}){0}
+
+      Concurrent::dataflow_with(root_executor, ){0}
+      Concurrent::dataflow_with(root_executor, Future.execute{0}){0}
+      Concurrent::dataflow_with(root_executor, Future.execute{0}, Future.execute{0}){0}
     end
 
     it 'accepts uncompleted dependencies' do
       d = Future.new(executor: executor){0}
       Concurrent::dataflow(d){0}
       d.execute
+
+      d = Future.new(executor: executor){0}
+      Concurrent::dataflow_with(root_executor, d){0}
+      d.execute
     end
 
     it 'accepts completed dependencies' do
       d = Future.new(executor: executor){0}
       d.execute
       Concurrent::dataflow(d){0}
+
+      d = Future.new(executor: executor){0}
+      d.execute
+      Concurrent::dataflow_with(root_executor, d){0}
     end
 
     it 'raises an exception if any dependencies are not IVars' do
       expect { Concurrent::dataflow(nil) }.to raise_error(ArgumentError)
       expect { Concurrent::dataflow(Future.execute{0}, nil) }.to raise_error(ArgumentError)
       expect { Concurrent::dataflow(nil, Future.execute{0}) }.to raise_error(ArgumentError)
+
+      expect { Concurrent::dataflow_with(root_executor, nil) }.to raise_error(ArgumentError)
+      expect { Concurrent::dataflow_with(root_executor, Future.execute{0}, nil) }.to raise_error(ArgumentError)
+      expect { Concurrent::dataflow_with(root_executor, nil, Future.execute{0}) }.to raise_error(ArgumentError)
     end
 
     it 'returns a Future' do
       Concurrent::dataflow{0}.should be_a(Future)
+      Concurrent::dataflow{0}.should be_a(Future)
     end
 
     context 'does not schedule the Future' do
@@ -51,6 +85,11 @@ module Concurrent
         f = Concurrent::dataflow(d){0}
         f.should be_unscheduled
         d.execute
+
+        d = Future.new(executor: executor){0}
+        f = Concurrent::dataflow_with(root_executor, d){0}
+        f.should be_unscheduled
+        d.execute
       end
 
       specify 'if one dependency of two is completed' do
@@ -60,6 +99,13 @@ module Concurrent
         d1.execute
         f.should be_unscheduled
         d2.execute
+
+        d1 = Future.new(executor: executor){0}
+        d2 = Future.new(executor: executor){0}
+        f = Concurrent::dataflow_with(root_executor, d1, d2){0}
+        d1.execute
+        f.should be_unscheduled
+        d2.execute
       end
     end
 
@@ -70,6 +116,11 @@ module Concurrent
         f = Concurrent::dataflow(d){0}
         d.execute
         f.value.should eq 0
+
+        d = Future.new(executor: executor){0}
+        f = Concurrent::dataflow_with(root_executor, d){0}
+        d.execute
+        f.value.should eq 0
       end
 
       specify 'if there is more than one' do
@@ -79,6 +130,13 @@ module Concurrent
         d1.execute
         d2.execute
         f.value.should eq 0
+
+        d1 = Future.new(executor: executor){0}
+        d2 = Future.new(executor: executor){0}
+        f = Concurrent::dataflow_with(root_executor, d1, d2){0}
+        d1.execute
+        d2.execute
+        f.value.should eq 0
       end
     end
 
@@ -89,6 +147,11 @@ module Concurrent
         d.execute
         f = Concurrent::dataflow(d){0}
         f.value.should eq 0
+
+        d = Future.new(executor: executor){0}
+        d.execute
+        f = Concurrent::dataflow_with(root_executor, d){0}
+        f.value.should eq 0
       end
 
       specify 'if there is more than one' do
@@ -98,26 +161,41 @@ module Concurrent
         d2.execute
         f = Concurrent::dataflow(d1, d2){0}
         f.value.should eq 0
+
+        d1 = Future.new(executor: executor){0}
+        d2 = Future.new(executor: executor){0}
+        d1.execute
+        d2.execute
+        f = Concurrent::dataflow_with(root_executor, d1, d2){0}
+        f.value.should eq 0
       end
     end
 
     context 'passes the values of dependencies into the block' do
 
       specify 'if there is just one' do
         d = Future.new(executor: executor){14}
-        f = Concurrent::dataflow(d) do |v|
-          v
-        end
+        f = Concurrent::dataflow(d){|v| v }
+        d.execute
+        f.value.should eq 14
+
+        d = Future.new(executor: executor){14}
+        f = Concurrent::dataflow_with(root_executor, d){|v| v }
         d.execute
         f.value.should eq 14
       end
 
       specify 'if there is more than one' do
         d1 = Future.new(executor: executor){14}
         d2 = Future.new(executor: executor){2}
-        f = Concurrent::dataflow(d1, d2) do |v1, v2|
-          v1 + v2
-        end
+        f = Concurrent::dataflow(d1, d2) {|v1, v2| v1 + v2}
+        d1.execute
+        d2.execute
+        f.value.should eq 16
+
+        d1 = Future.new(executor: executor){14}
+        d2 = Future.new(executor: executor){2}
+        f = Concurrent::dataflow_with(root_executor, d1, d2) {|v1, v2| v1 + v2}
         d1.execute
         d2.execute
         f.value.should eq 16
@@ -126,15 +204,15 @@ module Concurrent
 
     context 'module function' do
 
-      it 'can be called as Concurrent.dataflow' do
+      it 'can be called as Concurrent.dataflow and Concurrent.dataflow_with' do
 
         def fib_with_dot(n)
           if n < 2
             Concurrent.dataflow { n }
           else
             n1 = fib_with_dot(n - 1)
             n2 = fib_with_dot(n - 2)
-            Concurrent.dataflow(n1, n2) { n1.value + n2.value }
+            Concurrent.dataflow_with(root_executor, n1, n2) { n1.value + n2.value }
           end
         end
 
@@ -143,15 +221,15 @@ def fib_with_dot(n)
         expected.value.should eq 377
       end
 
-      it 'can be called as Concurrent::dataflow' do
+      it 'can be called as Concurrent::dataflow and Concurrent::dataflow_with' do
 
         def fib_with_colons(n)
           if n < 2
             Concurrent::dataflow { n }
           else
             n1 = fib_with_colons(n - 1)
             n2 = fib_with_colons(n - 2)
-            Concurrent::dataflow(n1, n2) { n1.value + n2.value }
+            Concurrent::dataflow_with(root_executor, n1, n2) { n1.value + n2.value }
           end
         end