Can't shutdown stdin of process

[daniel-emotech]

Version

0.2.11

Platform

Ubuntu 18.04 x64

Subcrates

process, io

Description

So I'm piping into and out of a process using tokio process. And the input goes fine, and the output pipe was getting to the last byte then hanging (when I did it in chunks) or just hanging if I tried to use read_to_end.

I ended up diagnosing this further and finding that the stdin wasn't being closed so I added a shutdown() to my code but then the write hung there (write function included).

async fn write_buffer<W>(data: &[u8], mut w: W) -> io::Result<()> where W: ASyncWriteExt + Unpin {
    let mut buffer = Cursor::new(data);
    while buffer.has_remaining() {
        w.write_buf(&mut buffer).await?;
    }
    w.shutdown().await?; // It hangs here
}

To solve this temporarily I've removed the shutdown and in my read I read chunks into a buffer and have a timeout to spot the end of stream since the program I'm piping to/from is very fast. But I'm wondering if there's an issue there

[ipetkov]

Hi @daniel-emotech thanks for the report!

Could you share the other code related to process management which leads to this? This may be a dup of #2174 but I can't tell for sure without seeing what the process related code is doing

[daniel-emotech]

Yeah see below.

I managed to get past this by removing the call to shutdown and putting an async move {} around in_hnd and the call to write_buffer so the ChildStdout was properly dropped once the future completed.

async fn write_buffer<W>(data: &[u8], w: &mut W) -> io::Result<()> where W: AsyncWriteExt {
    let mut buffer = Cursor::new(data);
    while buffer.has_remaining() {
        info!("Trying a write");
        let x = w.write_buf(&mut bufer).await?;
        info!("Wrote {} bytes", x);
    }
    w.shutdown().await?; // It hangs here
    Ok(())
}

async fn read_buffer<R>(r: &mut R) -> io::Result<Vec<u8>> where R: AsyncReadExt {
    let mut result = Vec::new();
    let mut buffer = BytesMut::with_capacity(65535);
    loop {
        info!("Trying a read");
        let x = r.read_buf(&mut bufer).await;
        match x {
            Err(e) => panic!("{}", e),
            Ok(0) => break,
            Ok(s) => {
                info!("Read {}", b);
                result.extend_from_slice(&buffer);
            },
        }
    }
}

async fn piped_process(data: &[u8]) -> io::Result<Vec<u8>> {
    let mut child = Command::new("some process")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .spawn()?;

    let mut stdin = child.stdin.take();
    let in_hnd = stdin.as_mut().unwrap()?;
    let in_fut = write_buffer(data, in_hnd);

    let mut stdout = child.stdout.take();
    let out_hnd = stdout.as_mut().unwrap();
    let out_fut = read_buffer(out_hnd);
    
    let (write, read) = tokio::join!(in_fut, out_fut);
}

[Darksonn]

I am closing this as it seems like #2218 has fixed the issue. If not, please open a new one.