How to write to and read from the same named pipe in a single ruby script?

edit: I think I fixed the issue: https://gist.github.com/niuage/c0637b8dd10549a12b6a223dbd5f158a

I might have been missing the Process.wait, hence creating a lot of zombie processes.

I have a piece of code that's working most of the time, but "locks" itself after a while, probably because of a race condition.

My code

pipe = "goals.png"

(1..100).each do |time|
  fork do
    # This runs ffmpeg, and gets it to write 1 frame of a video to the pipe 'goals.png'
    print "before ffmpeg"
    `#{ffmpeg(time, score_crop_options, pipe)}`
    exit
  end

  # Reads from 'pipe'
  print "before read"
  blob = File.read(pipe)
  image = Rocket::Image.from_blob(blob)
  # do stuff with image
end

Things to note:

#{ffmpeg(time, pipe)} writes to pipe, and is blocking until something reads from pipe
File.read(pipe) is blocking until something writes to pipe

My issue

edit: when the script is locked, and I try to read the pipe from another script, I get zsh: fork failed: resource temporarily unavailable. That's probably a good clue...

Most of the time, File.read(pipe) gets executed before the code in fork, so it works great, but after a little while the script just stops: it prints "before ffmpeg" and never gets to "before read"...

First, should I use threads instead of fork? And can I control the order the 2 statements (read and write) get run, to avoid a race condition? Or maybe it's not even about the race condition and I'm missing something?

Solution

The issue wasn't caused by a race condition, but too many zombie processes, since I wasn't calling Process.wait

The parent process should use Process.wait to collect the termination statuses of its children or use Process.detach to register disinterest in their status; otherwise, the operating system may accumulate zombie processes.

That's why I was getting zsh: fork failed: resource temporarily unavailable when trying to read from the pipe from another script probably.

Here's something that works:

(1..100) do
  if fork
    # Parent
    image = read_image(pipe)
    # do stuff with image

    Process.wait # I think that's what was missing previously
  else
    # Child
    Open3.popen3(command(time, score_crop_options, pipe)) do |stdin, stdout, stderr, wait_thr|
      # stuff
    end
    exit!(0)
  end
end