@@ -2,13 +2,17 @@ require 'thread' module Forkoff def version - '0.0.4' + '1.1.0' end def done @done ||= Object.new end + def ready + @ready ||= Object.new + end + def default @default ||= { 'processes' => 2 } end @@ -137,7 +141,8 @@ module Enumerable 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 } + p2c = Queue.new + c2p = Queue.new results = Array.new(n){ [] } # @@ -151,7 +156,8 @@ module Enumerable Thread.current.abort_on_exception = true loop do - value = qs[i].pop + c2p.push( Forkoff.ready ) + value = p2c.pop break if value == Forkoff.done args, index = value @@ -177,22 +183,18 @@ module Enumerable # # producers # - producers = [] - - n.times do |i| - thread = Thread.new do + producer = + 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] ) + each_with_index do |args, i| + ready = c2p.pop + p2c.push( [args, i] ) + end + n.times do |i| + p2c.push( Forkoff.done ) end - qs[ i ].push( Forkoff.done ) end - producers << thread - end - # # wait for all consumers to complete # lib/forkoff.rb @@ -9,6 +9,7 @@ SYNOPSIS URI http://rubyforge.org/projects/codeforpeople + http://github.com/ahoward/forkoff INSTALL @@ -25,6 +26,14 @@ SAMPLES <%= samples %> HISTORY + 1.1.0 + - move to a model with one work queue and signals sent from consumers to + producer to noitify ready state. this let's smaller jobs race through a + single process even while a larger job may have one sub-process bound up. + incorporates a fix from http://github.com/fredrikj/forkoff which meant + some processes would lag behind when jobs didn't have similar execution + times. + 1.0.0 - move to github readme.erb @@ -38,4 +38,17 @@ class T < Test::Unit::TestCase end end +# in case of different execution times for different processes +# + def test_0030 + a = Time.now.to_f + (0...4).forkoff(2) do |i| + sleep i.modulo(2) + end + b = Time.now.to_f + elapsed = b - a + assert elapsed < 2 + end + + end test/forkoff.rb 673cf3dffe955fd4a6d288fc157e0eca97eed0a5 ara.t.howard ara.t.howard@gmail.com http://github.com/ahoward/forkoff/commit/202886f4138e0321790bd54610bb99ecf058278b 202886f4138e0321790bd54610bb99ecf058278b 2009-10-11T01:36:05-07:00 2009-10-11T01:36:05-07:00 single work queue producer -> consumers and notifications redarding readines sent via a consumer -> producer 8e1d79164b8d7b444544e69d99fb5622eb4ddd6a ara.t.howard ara.t.howard@gmail.com