ChildStdin does not send EOF on .shutdown()

[coral]

Version
1.16.1

❯ cargo tree | grep tokio
│   │   │   ├── tokio v1.16.1
│   │   │   │   └── tokio-macros v1.7.0 (proc-macro)
│   │   │   ├── tokio-util v0.6.9
│   │   │   │   └── tokio v1.16.1 (*)
│   │   │   ├── tokio v1.16.1 (*)
│   │   │   ├── tokio v1.16.1 (*)
│   │   │   └── tokio-native-tls v0.3.0
│   │   │       └── tokio v1.16.1 (*)
│   │   ├── tokio v1.16.1 (*)
│   │   ├── tokio-native-tls v0.3.0 (*)
│   ├── tokio v1.16.1 (*)
├── tokio v1.16.1 (*)

Platform

Mac OSX Monterey 12.1

Darwin 21.2.0 Darwin Kernel Version 21.2.0: Sun Nov 28 20:28:54 PST 2021; root:xnu-8019.61.5~1/RELEASE_X86_64 x86_64

Description

ChildStdin does not seem to properly send EOF on shutdown(). I am porting some code from another project and one step is triggering file syncing using rsync as a child process, expecting a pipe of files for input.

 let mut cmd = Command::new("rsync");
cmd.args([
    "--ignore-existing",
    "-r",
    "-v",
    "--files-from=-",
    &self.source,
    &self.dest,
]);

cmd.stdout(Stdio::inherit());
cmd.stderr(Stdio::inherit());
cmd.stdin(Stdio::piped());

let mut child = cmd.spawn()?;

let mut tn = child.stdin.take();
let stdin = tn.as_mut().ok_or(Error::CouldNotGetStdin)?;

let mut filelist: String = "".to_string();
for f in files {
    filelist.push_str(&format!("{}\n", f));
}
stdin.write_all(filelist.as_bytes()).await?;

stdin.shutdown().await?;

What ends up happening is the process keeps waiting for the EOF without any closure. I tried by running the command manually and sending Ctrl + D (EOF) which does work. The same command also works in Go by expecting EOF

// This is in Go
// EXAMPLE CODE
stdin, _ := cmd.StdinPipe()

go func() {
	defer stdin.Close()
	for _, p := range filesToSync {
		io.WriteString(stdin, p+"\n")
	}
}()

stdin.Close() in this case will send EOF upon completion whereas ChildStdin seems to not properly do this.

[Darksonn]

Thoughts @ipetkov?

[coral]

tokio/tokio/src/process/mod.rs

Lines 1233 to 1249 in fc4deaa

	impl AsyncWrite for ChildStdin {
	fn poll_write(
	self: Pin<&mut Self>,
	cx: &mut Context<'_>,
	buf: &[u8],
	) -> Poll<io::Result<usize>> {
	self.inner.poll_write(cx, buf)
	}
	fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
	Poll::Ready(Ok(()))
	}
	fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
	Poll::Ready(Ok(()))
	}
	}

Alice Ryhl pointed out in the Discord that the trait for shutdown seems broken. From my understanding the polling of shutdown doesn't actually do anything?

Edit: "darksonn" is alice ryhl in discord I now realize...

[ipetkov]

The AsyncWriteExt::shutdown is supposed to encompass flushing any buffered state along with any other necessary graceful shutdown (e.g. like tearing down a TLS connection), but it doesn't actually close the pipe.

@coral what comes after the shutdown().await call? You can try explicitly dropping the stdin handle to make sure it gets closed, sending EOF to the child process so it can complete and exit.

let mut cmd = Command::new("rsync");
cmd.args([
    "--ignore-existing",
    "-r",
    "-v",
    "--files-from=-",
    &self.source,
    &self.dest,
]);

cmd.stdout(Stdio::inherit());
cmd.stderr(Stdio::inherit());
cmd.stdin(Stdio::piped());

let mut child = cmd.spawn()?;

let mut tn = child.stdin.take();
let stdin = tn.as_mut().ok_or(Error::CouldNotGetStdin)?;

let mut filelist: String = "".to_string();
for f in files {
    filelist.push_str(&format!("{}\n", f));
}
stdin.write_all(filelist.as_bytes()).await?;

stdin.shutdown().await?;
drop(stdin);

child.await // etc.

[coral]

@ipetkov

Dropping works yes (that's how I ended up working around this problem) but I'm curious if this is the intended design of the API? As in the stdin basically holding onto the pipe handle for the duration of the lifetime, leaving "shutdown" basically unused? If so, is the methods for AsyncWriter on ChildStdin just assumed to basically be NOP?

[ipetkov]

@coral

Dropping works yes (that's how I ended up working around this problem) but I'm curious if this is the intended design of the API?

Yep, this is the idiomatic way of handling resource cleanup in Rust, similar to closing channels, dropping the io handle leads to closing it. (in fact this is the only way to close the handle)

As in the stdin basically holding onto the pipe handle for the duration of the lifetime, leaving "shutdown" basically unused? If so, is the methods for AsyncWriter on ChildStdin just assumed to basically be NOP?

Yeah shutdown and flush are basically NOPs on the ChildStd{in,out,err} handles since they don't do much on their own besides give us some type safety when working with the child process. If you were using something like a buffered reader/writer, for example, then it may be necessary to call flush to get all the data sent, but as the example code stands, the shutdown call could be elided!

[coral]

@ipetkov

#4479

Took a pass at creating a simple example for the documentation which captures a lot of what I've learned from this.

[Darksonn]

@ipetkov Are we sure this is what we want? I mean, pipes can send an EOF signal, and other types of channels (e.g. TcpStream, UnixStream) will send EOF when their poll_shutdown method is called.

[ipetkov]

@Darksonn

pipes can send an EOF signal

Is this true without explicitly closing the pipe end? AFAIU shutdown(3) is specific to sockets. Even the stdlib only has shutdown for sockets and not for regular pipes.

Not opposed to adding this behavior if possible, just not sure it exists

[Darksonn]

Hm, interesting. I'll have to think about what to do about that. It is a difference from std that our AsyncWrite trait has a method for shutdown.