Packet loss with process pipes

[jb-alvarado]

Hello everyone,
I have a problem here that is difficult for me to debug because it only occurs sporadically every few hours.

Basically, I create an ffmpeg process that runs endlessly and is supplied with data via stdin. In a loop, I then start new ffmpeg processes that read in video files and connect their stdout to the first ffmpeg process via pipes.

In older versions of the program I did not handle this via tokio, but everything ran in different threads and with std processes. This always ran wonderfully stable for weeks.

Now with tokio it happens again and again that the endless ffmpeg process gets broken packets and my program terminates the process due to error limits.

As I said, it's difficult to debug, so I can't say exactly at what point the errors occur. I just have a strong suspicion that it happened during the transfer from stdout of one process to stdin of the other process. This is because if I write the wrong number of packets with write_all, I get exactly the same error messages in ffmpeg.

Simplified, my code looks something like this, with output_proc normally running in a different task:

let output_proc = Command::new("ffmpeg")
        .args(["-re", "-i", "pipe:0", ".."])
        .kill_on_drop(true)
        .stdin(Stdio::piped())
        .spawn()?;

let output_writer = BufWriter::with_capacity(65088, output_proc.stdin.take().unwrap());
let mut buffer = [0; 65088];

for file in files.next().await {
    let mut input_proc = Command::new("ffmpeg")
        .args(["-i", file, ".."])
        .kill_on_drop(true)
        .stdout(Stdio::piped())
        .spawn()?;

    let mut input = BufReader::with_capacity(65088, input_proc.stdout.take().unwrap());

    loop {
        let num = input.read(&mut buffer).await?;

        if num == 0 {
            break;
        }

        output_writer.write_all(&buffer[..num]).await?;
    }

    output_writer.flush().await?;

    drop(input);

    input_proc.wait().await;
}

Does anyone here have any ideas what I could do to fix this problem?

[Darksonn]

The only potential problem I can see is that because output_writer is buffered, there could be a situation where your program has things it could write to output_proc.stdin, but it is waiting for more data on input.read and will not write the data until input.read returns.

It is curious that you're using 65088 as the size everywhere. There's no guarantee that the data actually gets sent in chunks of that size, and ffmpeg should not expect it either.

[jb-alvarado]

Thank you @Darksonn for your suggestion!

If have try to work without buffers, but because I need output_proc in a mutex to be able to kill it in a separate task I did not find a better solution to get output_proc.stdin and not partial move output_proc. I tried output_writer also with a default buffer size and got the same problems, but I can skip with_capacity, my only concern was that with a 8kb buffer size I get less throughput.

Do you have another suggestion about how to deal with the chunks?

[jb-alvarado]

I removed read and write buffer, lower my buffer and use write instead of write_all. Something like:

let output_proc = Command::new("ffmpeg")
        .args(["-re", "-i", "pipe:0", ".."])
        .kill_on_drop(true)
        .stdin(Stdio::piped())
        .spawn()?;

let output_writer = output_proc.stdin.take().unwrap();
let mut buffer = [0; 4096];

for file in files.next().await {
    let mut input_proc = Command::new("ffmpeg")
        .args(["-i", file, ".."])
        .kill_on_drop(true)
        .stdout(Stdio::piped())
        .spawn()?;

    let mut input = input_proc.stdout.take().unwrap();

    loop {
        let num = input.read(&mut buffer).await?;

        if num == 0 {
            break;
        }

        let nw = output_writer.write(&buffer[..num]).await?;

        if nw != num {
            eprintln!("Read and written packages differs...");
        }
    }

    drop(input);

    input_proc.wait().await;
}

First it looks promising, but after some hours I get the same issues. I have now idea what else I can try...

[jb-alvarado]

Ok, I have to apologize for stealing your time. The problem was not with the Rust code, but simply with the way I put my ffmpeg commands together. Under certain circumstances the order of the audio and video mapping was swapped (once -map “0:v” -map “0:a” and once -map “0:a” -map “0:v”), which the receiver instance didn't like.