Parallel process barriers in Ruby

parallel_process_barriers.rb

#
# Runs two blocks in parallel, yielding a barrier to each for
# synchronization. The barrier may be called multiple times in each
# block to establish multiple synchronization points.
#
#     parallel do
#       process do |barrier|
#         ...
#         barrier.call
#         ...
#       end
#       process do |barrier|
#         ...
#         barrier.call
#         ...
#       end
#     end
#
def parallel(&block)
  p1 = p2 = nil
  object = Object.new
  object.singleton_class.__send__(:define_method, :process) do |&body|
    if !p1
      p1 = body
    elsif !p2
      p2 = body
    else
      raise ArgumentError, "more than 2 processes given"
    end
  end
  object.instance_eval(&block)
  r1, w1 = IO.pipe
  r2, w2 = IO.pipe
  r1.sync = w1.sync = r2.sync = w2.sync = true
  pid = fork do
    r1.close
    w2.close
    barrier = lambda do
      w1.print('.')
      r2.read(1)
    end
    instance_exec(barrier, &p2)
  end
  begin
    r2.close
    w1.close
    barrier = lambda do
      w2.print('.')
      r1.read(1)
    end
    instance_exec(barrier, &p1)
  ensure
    Process.waitpid(pid)
  end
end