From: Erik Veenstra <erikveen@...>
Date: 2008-04-19T03:19:17+09:00
Subject: Re: forkoff - parallel processing for ruby enumerables

I've once implemented Enumerable#fork myself. It doesn't use
queues, or a producer-consumer like pattern. It simply tells a
generic ThreadLimiter to spawn a new thread. Within this
thread, a new process is spawned. The number of concurrent
threads, and thus the number of concurrent processes, is
controlled by ThreadLimiter.

We might learn from both implementations.

gegroet,
Erik V. - http://www.erikveen.dds.nl/

----------------------------------------------------------------

Here's my code:

----------------------------------------------------------------

 module Enumerable
   def fork(max_number_of_threads=nil, &block)
     thread_limiter      =
EV::ThreadLimiter.new(max_number_of_threads)

     collect do |x|
       thread_limiter.fork do
         Thread.current.abort_on_exception = true

         r, w    = IO.pipe

         if pid = Process.fork
           w.close
           Process.wait(pid)
           data  = r.read
           r.close
           Marshal.load(data)
         else
           r.close
           Marshal.dump(block.call(x), w)
           w.close
           exit
         end
       end
     end.collect do |t|
       t.value
     end
   end
 end

----------------------------------------------------------------

 module EV
   class ThreadLimiter
     def initialize(max_number_of_threads)
       @number_of_threads        = 0
       @max_number_of_threads    = max_number_of_threads

       yield(self)       if block_given?
     end

     def fork(*args, &block)
       Thread.pass       while @max_number_of_threads and
                               @max_number_of_threads > 0 and
                               @number_of_threads >
@max_number_of_threads

       # If this methods is called from several threads, then
       # @number_of_threads might get bigger than
@max_number_of_threads.
       # This usually a) isn't the case and b) doesn't really matter
(to me...).
       # I'm willing to accept this "risk", because a)
Thread.exclusive is
       # much, much faster than Mutex#synchronize and b) we can't run
into
       # deadlocks.

       Thread.exclusive{@number_of_threads += 1}

       Thread.fork do
         begin
           res   = block.call(*args)
         ensure
           Thread.exclusive{@number_of_threads -= 1}
         end

         res
       end
     end
   end
 end

----------------------------------------------------------------

Here's a benchmark:

 require "benchmark"

 Benchmark.bm(15) do |bm|
   rc    = nil
   r2    = nil
   r4    = nil
   rx    = nil

   data  = 1..10
   test  = lambda{|x| 1_000_000.times{7+8}; [x, Process.pid]}

   bm.report("    collect  "){rc = data.collect(&test)}
   bm.report("  2 processes"){r2 = data.fork(2, &test)}
   bm.report("  4 processes"){r4 = data.fork(4, &test)}
   bm.report("inf processes"){rx = data.fork(-1, &test)}

   p rc
   p r2
   p r4
   p rx
 end

It produces these results on a dual core machine:

                      user     system      total        real
     collect      4.530000   0.000000   4.530000 (  4.527982)
   2 processes    0.030000   0.050000   3.170000 (  1.733209)
   4 processes    0.160000   0.370000   3.610000 (  1.927826)
 inf processes    0.000000   0.000000   3.080000 (  1.691932)
 [[1, 18732], [2, 18732], [3, 18732], [4, 18732], [5, 18732], [6,
18732], [7, 18732], [8, 18732], [9, 18732], [10, 18732]]
 [[1, 18733], [2, 18734], [3, 18735], [4, 18736], [5, 18737], [6,
18738], [7, 18739], [8, 18740], [9, 18741], [10, 18742]]
 [[1, 18743], [2, 18744], [3, 18745], [4, 18746], [5, 18747], [6,
18748], [7, 18749], [8, 18750], [9, 18751], [10, 18752]]
 [[1, 18753], [2, 18754], [3, 18755], [4, 18756], [5, 18757], [6,
18758], [7, 18759], [8, 18760], [9, 18761], [10, 18762]]

----------------------------------------------------------------