Only one queue

Author: Fredrik Johansson

lib/forkoff.rb

@@ -137,7 +137,7 @@ def forkoff options = {}, &block
     options = { 'processes' => Integer(options) } unless Hash === options
     n = Integer( options['processes'] || options[:processes] || Forkoff.default['processes'] )
     strategy = options['strategy'] || options[:strategy] || 'pipe'
-    qs = Array.new(n){ SizedQueue.new 1 }
+    qs = SizedQueue.new 1 # One queue
     results = Array.new(n){ [] }
 
     #
@@ -151,7 +151,7 @@ def forkoff options = {}, &block
             Thread.current.abort_on_exception = true
 
             loop do
-              value = qs[i].pop
+              value = qs.pop
               break if value == Forkoff.done
               args, index = value
 
@@ -175,23 +175,12 @@ def forkoff options = {}, &block
       end
 
     #
-    # producers
+    # one producer
     #
-      producers = []
-
-      n.times do |i|
-        thread = Thread.new do
-          Thread.current.abort_on_exception = true
-          each_with_index do |args, j|
-            every_nth = j.modulo(n) == i
-            next unless every_nth
-            qs[ j.modulo(n) ].push( [args, j] )
-          end
-          qs[ i ].push( Forkoff.done )
-        end
-
-        producers << thread
+      each_with_index do |args, j|
+        qs.push( [args, j] )
       end
+      n.times do qs.push( Forkoff.done ) end
 
     #
     # wait for all consumers to complete

test/forkoff.rb

@@ -38,4 +38,14 @@ def test_0020
     end
   end
 
+# in case of different execution times for different processes
+#
+  def test_0030
+    t0 = Time.now
+    (0...4).forkoff(2) do |i|
+      sleep i.modulo(2)
+    end
+    assert (Time.now - t0) < 2
+  end
+
 end