Rust: stdout von Prozess lesen und in stdin von zweiten Prozess schreiben

[jb_alvarado]

Hallo mal wieder,

ich möchte einen Prozess mit offenem stdin starten, anschließend über einen Liste von Dateien loopen, die mit einem zweiten Prozess gelesen werden sollen. Dann soll der stdout vom zweiten Prozess in einem weiteren Loop gelesen werden und in den stdin vom ersten Prozess geschrieben werden.

In Python würde das in etwas so funktionieren.

Mein erster Rust-Versuch schaut so aus:

let mut enc_proc = Command::new("ffplay")
    .args(enc_cmd)
    .stdin(Stdio::piped())
    .spawn()
    .unwrap();

if let Some(mut enc_input) = enc_proc.stdin.take() {
    for node in get_source {
        let mut dec_proc = Command::new("ffmpeg")
            .args(dec_cmd)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();

        if let Some(mut dec_output) = dec_proc.stdout.take() {
            io::copy(&mut dec_output, &mut enc_input).expect("Write to streaming pipe failed!");

            dec_proc.wait()?;
            let dec_output = dec_proc.wait_with_output()?;

            if dec_output.stderr.len() > 0 {
                println!(
                    "[Encoder] <red>{:?}</red>",
                    String::from_utf8(dec_output.stderr).unwrap()
                );
            }
        }
    }
}

Das würde so auch gehen, allerdings brauche ich unbedingt statt io::copy einen Loop der mir Häppchenweise die Bytes aus dem Zweiten Prozess ließt und den Ersten schreibt.

Hier habe ich ein Beispiel gefunden, wo in stdin geschrieben wird, aber das hat mir auch nicht viel geholfen.

Könnt ihr mir hier ein paar Tips geben?

 

[KitKat::new()]

Das würde so auch gehen, allerdings brauche ich unbedingt statt io::copy einen Loop der mir Häppchenweise die Bytes aus dem Zweiten Prozess ließt und den Ersten schreibt.

Woran scheiterts?

Mal mit read probiert?

 

[jb_alvarado]

Habe es jetzt tatsächlich hinbekommen. Habe es vorher auch schon mit read() versucht, aber mir die Doku dazu nicht angeschaut. In python gibt man die Menge der Bytes an die man lesen möchte, und nimmt den Rückgabewert als Objekt womit man wieder den Stdin vom anderen Prozess füttert. In Rust bewirkt read() dass es des Buffer-Objekt befüllt und man dieses für den Stdin nehmen muss.

Das war ein Problem, ein andere war, dass ich Prozess 1 nach dem verlassen des loops nicht mehr schließen konnte, weil ein Teil davon ausgeliehen war. Der Compiler sagt dann zwar immer, man soll diesen Teil als Referenz nehmen, aber das hatte nicht geklappt.

Auf jeden Fall geht es nun so:

use std::io::prelude::*;
use std::io::Read;
use std::process::{Command, Stdio};

fn main() {
    let mut enc_proc = match Command::new("ffplay")
        .args(["-v", "error", "-hide_banner", "-nostats", "-i", "pipe:0"])
        .stdin(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
    {
        Err(e) => panic!("couldn't spawn ffplay: {}", e),
        Ok(proc) => proc,
    };

    let mut buffer = [0; 65424];

    let dec_proc = match Command::new("ffmpeg")
        .args([
            "-f",
            "lavfi",
            "-i",
            "testsrc=duration=6:size=1280x720:rate=25",
            "-f",
            "lavfi",
            "-i",
            "anoisesrc=d=6:c=pink:r=48000:a=0.5",
            "-pix_fmt",
            "yuv420p",
            "-c:v",
            "mpeg2video",
            "-g",
            "1",
            "-b:v",
            "50000k",
            "-minrate",
            "50000k",
            "-maxrate",
            "50000k",
            "-bufsize",
            "25000k",
            "-c:a",
            "s302m",
            "-strict",
            "-2",
            "-ar",
            "48000",
            "-ac",
            "2",
            "-f",
            "mpegts",
            "-",
        ])
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
    {
        Err(e) => panic!("couldn't spawn ffmpeg: {}", e),
        Ok(proc) => proc,
    };

    let mut enc_writer = enc_proc.stdin.as_ref().unwrap();
    let mut dec_reader = dec_proc.stdout.unwrap();

    loop {
        match dec_reader.read_exact(&mut buffer[..]) {
            Ok(_) => println!("Do some stuff between reading and writing"),
            Err(_) => break,
        };

        match enc_writer.write_all(&buffer) {
            Ok(_) => {}
            Err(e) => panic!("Err: {:?}", e),
        };
    }

    match enc_proc.kill() {
        Ok(_) => println!("Playout done..."),
        Err(e) => panic!("Enc error: {:?}", e)
    }
}

Weißt du zufällig ob es sicher ist, den Stderr des jeweiligen Prozesses an einen Thread zu senden? Würde das gerne machen, um dort Fehler usw. zu loggen.

 

[KitKat::new()]

Weißt du zufällig ob es sicher ist, den Stderr des jeweiligen Prozesses an einen Thread zu senden?

wenn du es ohne "unsafe" hinbekommst, normal schon

ChildStdErr scheint ja Send zu implementieren

 

[jb_alvarado]

Habe gerade noch festgestellt, dass einmal dieser Loop mehr Speicher verbraucht als io::copy, was ok ist - solange er nicht wächst.

Aber das größere Problem ist, dass der fertige Prozess "dec_proc" nach dem Durchlauf weiterlebt. Macht sich bemerkbar, wenn man dann über Dateien Schleift und neue Prozesse startet. Die tauchen dann im System nach Abschluss mit [ffmpeg] <defunct> auf.

Wenn ich dec_proc.wait(); nach dem loop einsetzten will, habe ich wieder das Problem, dass ein Teil ausgeliehen ist, und referenzieren klappt in dem Fall auch nicht.

 

[KitKat::new()]

Wenn ich dec_proc.wait(); nach dem loop einsetzten will, habe ich wieder das Problem, dass ein Teil ausgeliehen ist, und referenzieren klappt in dem Fall auch nicht.

Kannst du das genau nochmal als Code posten?

Es sei denn ich rate richtig:
Hier nutzt du ja nicht mehr take

let mut dec_reader = dec_proc.stdout.unwrap();

https://doc.rust-lang.org/std/process/struct.Child.html#structfield.stdin

D.h. du willst dec_proc exklusiv nutzen (wait nimmt ja &mut) während dec_reader noch lebt.

 

[jb_alvarado]

Ja du hast richtig gerate. Der Compiler hat daraufhin vorgeschlagen ich soll .as_ref() nehmen. Also:

let dec_reader = dec_proc.stdout.as_ref().unwrap();

Das war aber falsch, so ging es dann:

let dec_reader = dec_proc.stdout.as_mut().unwrap();

Habe mir zwar eine 1,5 Stunden lange Vorlesung zu Borrowing angeschaut, aber das ist immer noch die größte Hürde für mich.

 

[KitKat::new()]

Alternativ könnte es auch funktionieren, dass du dec_reader in einem extra Scope aufmachst, in etwa so:

{
    let mut dec_reader = dec_proc.stdout.unwrap();
    //use dec_reader
}
dec_proc.wait();

 

[jb_alvarado]

Ne, klappt leider nicht, bekomme dann wieder die gleiche Fehlermeldung.