Skip to content

Instantly share code, notes, and snippets.

@mfrister
Last active January 20, 2017 22:39
Show Gist options
  • Save mfrister/b591c5b09c7e66fec85b to your computer and use it in GitHub Desktop.
Save mfrister/b591c5b09c7e66fec85b to your computer and use it in GitHub Desktop.
Concurrently spawn Ruby processes, collect their output and wait for them to exit
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment