Enumerator::Parallel

parallel.rb

END {
  8.times.parallel(n: 2).each { p _1 }
}

class Enumerator
  class Parallel
    def initialize(enum, n:)
      @enum = enum
      @n = n
    end

    def each(&blk)
      pipe = Ractor.new do
        loop do
          Ractor.yield(Ractor.recv)
        end
      end
      rs = @n.times.map do |i|
        Ractor.new(pipe, blk.object_id) do |pipe, blk_id|
          b = ObjectSpace.each_object {|i| break i if i.object_id == blk_id }
          while l = pipe.take
            b.call(l)
          end
        rescue Ractor::ClosedError
        end
      end
      @enum.each do |i|
        pipe.send(i)
      end
      pipe.close
      until rs.empty?
        r, = Ractor.select(*rs)
        rs.delete r
      end
    end
  end

  def parallel(n:)
    Enumerator::Parallel.new(self, n: n)
  end
end