Concurrently spawn Ruby processes, collect their output and wait for them to exit

concurrent_spawn.rb

module ConcurrentSpawn
  def self.run(commands, &block)
    processes = spawn(commands)
    collect_output(processes, &block)
  end

  def self.spawn(commands)
    processes = commands.map do |cmd|
      r, w = IO.pipe
      pid = Process.spawn(cmd, out: w, err: [:child, :out], in: :close)

      [
        pid,
        {
          command:      cmd,
          pid:          pid,
          read_pipe:    r,
          exit_status:  nil,
          output:       '',
        }
      ]
    end.to_h
  end

  # Collect output and wait for exit of all child processes
  def self.collect_output(processes, &block)
    processes_by_read_pipe = processes.map {|k,v| [v[:read_pipe], v]}.to_h

    while !processes_by_read_pipe.empty?
      readable, _, _ = IO.select(processes_by_read_pipe.keys, nil, nil, 0.1)

      if readable
        readable.each do |read_pipe|
          process = processes_by_read_pipe[read_pipe]
          begin
            process[:output] << read_pipe.read_nonblock(64 * 1024)
          rescue EOFError
            # They happen sometimes and do not mean the command failed.
          rescue => e
            puts "Warning: Failed to read (cmd: '#{process[:command]}'): #{e} (#{e.class})"
          end
        end
      end

      processes_by_read_pipe.each do |_, process|
        if Process.waitpid(process[:pid], Process::WNOHANG)
          process[:read_pipe].close
          process[:exit_status] = $?.exitstatus
          processes_by_read_pipe.delete(process[:read_pipe])
          block.call(process) if block
        end
      end
    end

    processes.values
  end
end