From 65f149a90bd760b19c0ef5937ef8083666f3da27 Mon Sep 17 00:00:00 2001 From: itsmunzir Date: Sun, 27 Sep 2026 15:13:47 +0300 Subject: [PATCH 1/4] feat(transport): add non-blocking stdio_pipes transport `transport::stdio()` reads and writes through Tokio's blocking thread pool, so every message on a piped stdio connection pays a cross-thread hand-off. MCP clients launch stdio servers with FIFOs for stdin/stdout, which Tokio's Unix pipe driver can drive from the IO reactor instead. `stdio_pipes()` wraps the standard streams as non-blocking pipes when both are FIFOs and otherwise falls back to the blocking `stdio()` handles, so the returned `(StdioReader, StdioWriter)` pair still converts into a transport on every platform. TTYs, files, sockets, and Windows keep today's behavior. `stdio()` is left unchanged because changing its return type would be a breaking public API change; this constructor is additive. The `transport-io` feature now enables `tokio/net` for the pipe types. Refs #1301 --- crates/rmcp/Cargo.toml | 2 +- crates/rmcp/src/transport.rs | 2 +- crates/rmcp/src/transport/io.rs | 241 ++++++++++++++++++++++++++++++++ 3 files changed, 243 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/Cargo.toml b/crates/rmcp/Cargo.toml index 772844d86..101a1ef64 100644 --- a/crates/rmcp/Cargo.toml +++ b/crates/rmcp/Cargo.toml @@ -173,7 +173,7 @@ transport-streamable-http-client-unix-socket = [ ] transport-async-rw = ["tokio/io-util", "tokio-util/codec"] -transport-io = ["transport-async-rw", "tokio/io-std"] +transport-io = ["transport-async-rw", "tokio/io-std", "tokio/net"] transport-child-process = [ "transport-async-rw", "tokio/process", diff --git a/crates/rmcp/src/transport.rs b/crates/rmcp/src/transport.rs index fdedc9c3a..0843f16d9 100644 --- a/crates/rmcp/src/transport.rs +++ b/crates/rmcp/src/transport.rs @@ -91,7 +91,7 @@ pub use child_process::{ConfigureCommandExt, TokioChildProcess}; #[cfg(feature = "transport-io")] pub mod io; #[cfg(feature = "transport-io")] -pub use io::stdio; +pub use io::{StdioReader, StdioWriter, stdio, stdio_pipes}; #[cfg(feature = "auth")] pub mod auth; diff --git a/crates/rmcp/src/transport/io.rs b/crates/rmcp/src/transport/io.rs index adc2dba0e..381c6f872 100644 --- a/crates/rmcp/src/transport/io.rs +++ b/crates/rmcp/src/transport/io.rs @@ -1,6 +1,247 @@ +use std::{ + io, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + /// # StdIO Transport /// /// Create a pair of [`tokio::io::Stdin`] and [`tokio::io::Stdout`]. +/// +/// Every read from and write to these handles is executed on Tokio's blocking +/// thread pool. When both standard streams are pipes, prefer [`stdio_pipes`], +/// which avoids the per-operation thread hand-off on Unix. pub fn stdio() -> (tokio::io::Stdin, tokio::io::Stdout) { (tokio::io::stdin(), tokio::io::stdout()) } + +/// # Non-blocking StdIO Transport +/// +/// Like [`stdio`], but uses Tokio's non-blocking Unix pipe driver when both +/// standard streams are FIFOs — the way MCP clients launch stdio servers — so +/// reads and writes are driven by the IO reactor instead of Tokio's blocking +/// thread pool. This avoids a cross-thread hand-off per message under +/// concurrent requests. +/// +/// Falls back to the blocking [`stdio`] handles when the streams are not +/// pipes (TTYs, files, sockets) or on platforms without Unix pipes, so it can +/// be used unconditionally. +/// +/// On Unix the returned reader and writer own descriptors duplicated from +/// standard input and output; closing them does not close the process's +/// standard streams. +/// +/// # Panics +/// +/// On Unix, panics if called outside of a Tokio runtime with IO enabled, +/// mirroring `tokio::net::unix::pipe::Receiver::from_owned_fd`. +pub fn stdio_pipes() -> (StdioReader, StdioWriter) { + #[cfg(unix)] + if let Some(pair) = try_pipe_transport() { + return pair; + } + ( + StdioReader::Blocking(tokio::io::stdin()), + StdioWriter::Blocking(tokio::io::stdout()), + ) +} + +#[cfg(unix)] +fn try_pipe_transport() -> Option<(StdioReader, StdioWriter)> { + use std::os::fd::AsFd; + + let read = std::io::stdin().as_fd().try_clone_to_owned().ok()?; + let write = std::io::stdout().as_fd().try_clone_to_owned().ok()?; + pipe_transport(read, write) +} + +/// Wraps two FIFO descriptors as a non-blocking pipe transport. +/// +/// Returns `None` unless both descriptors are FIFOs. Both are checked before +/// either is handed to Tokio: `pipe::Receiver::from_owned_fd` and +/// `pipe::Sender::from_owned_fd` put the descriptor into non-blocking mode, +/// and `O_NONBLOCK` is shared by every descriptor of the same open file +/// description, so a partial conversion would leave the blocking fallback +/// with non-blocking stdio. +#[cfg(unix)] +fn pipe_transport( + read: std::os::fd::OwnedFd, + write: std::os::fd::OwnedFd, +) -> Option<(StdioReader, StdioWriter)> { + if !is_fifo(&read) || !is_fifo(&write) { + return None; + } + let read = tokio::net::unix::pipe::Receiver::from_owned_fd(read).ok()?; + let write = tokio::net::unix::pipe::Sender::from_owned_fd(write).ok()?; + Some((StdioReader::Pipe(read), StdioWriter::Pipe(write))) +} + +#[cfg(unix)] +fn is_fifo(fd: &std::os::fd::OwnedFd) -> bool { + use std::os::unix::fs::FileTypeExt; + + fd.try_clone() + .map(std::fs::File::from) + .and_then(|file| file.metadata()) + .is_ok_and(|metadata| metadata.file_type().is_fifo()) +} + +/// Reader half of the transport returned by [`stdio_pipes`]. +#[non_exhaustive] +pub enum StdioReader { + /// Blocking standard input, as returned by [`stdio`]. + Blocking(tokio::io::Stdin), + /// Non-blocking standard input backed by a Unix pipe. + #[cfg(unix)] + Pipe(tokio::net::unix::pipe::Receiver), +} + +impl AsyncRead for StdioReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + match self.get_mut() { + Self::Blocking(reader) => Pin::new(reader).poll_read(cx, buf), + #[cfg(unix)] + Self::Pipe(reader) => Pin::new(reader).poll_read(cx, buf), + } + } +} + +/// Writer half of the transport returned by [`stdio_pipes`]. +#[non_exhaustive] +pub enum StdioWriter { + /// Blocking standard output, as returned by [`stdio`]. + Blocking(tokio::io::Stdout), + /// Non-blocking standard output backed by a Unix pipe. + #[cfg(unix)] + Pipe(tokio::net::unix::pipe::Sender), +} + +impl AsyncWrite for StdioWriter { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + match self.get_mut() { + Self::Blocking(writer) => Pin::new(writer).poll_write(cx, buf), + #[cfg(unix)] + Self::Pipe(writer) => Pin::new(writer).poll_write(cx, buf), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Self::Blocking(writer) => Pin::new(writer).poll_flush(cx), + #[cfg(unix)] + Self::Pipe(writer) => Pin::new(writer).poll_flush(cx), + } + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.get_mut() { + Self::Blocking(writer) => Pin::new(writer).poll_shutdown(cx), + #[cfg(unix)] + Self::Pipe(writer) => Pin::new(writer).poll_shutdown(cx), + } + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[io::IoSlice<'_>], + ) -> Poll> { + match self.get_mut() { + Self::Blocking(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), + #[cfg(unix)] + Self::Pipe(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), + } + } + + fn is_write_vectored(&self) -> bool { + match self { + Self::Blocking(writer) => writer.is_write_vectored(), + #[cfg(unix)] + Self::Pipe(writer) => writer.is_write_vectored(), + } + } +} + +#[cfg(test)] +mod tests { + #[cfg(unix)] + fn fifo_pair() -> ( + std::process::Child, + std::os::fd::OwnedFd, + std::os::fd::OwnedFd, + ) { + use std::process::{Command, Stdio}; + + let mut child = Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .expect("failed to spawn `cat`"); + let write = child.stdin.take().expect("child stdin").into(); + let read = child.stdout.take().expect("child stdout").into(); + (child, read, write) + } + + #[cfg(unix)] + #[tokio::test] + async fn pipe_transport_round_trips_over_fifos() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use super::{StdioReader, StdioWriter, pipe_transport}; + + let (mut child, read, write) = fifo_pair(); + let (mut reader, mut writer) = pipe_transport(read, write).expect("child pipes are FIFOs"); + assert!(matches!(reader, StdioReader::Pipe(_))); + assert!(matches!(writer, StdioWriter::Pipe(_))); + + let message = b"{\"jsonrpc\":\"2.0\",\"id\":1}\n"; + writer.write_all(message).await.unwrap(); + writer.flush().await.unwrap(); + let mut echoed = vec![0; message.len()]; + reader.read_exact(&mut echoed).await.unwrap(); + assert_eq!(echoed, message); + + drop(writer); + assert!(child.wait().unwrap().success()); + } + + #[cfg(unix)] + #[tokio::test] + async fn pipe_transport_falls_back_for_non_fifos() { + use super::pipe_transport; + + let read = std::fs::File::open("/dev/null").unwrap().into(); + let write = std::fs::File::open("/dev/null").unwrap().into(); + assert!(pipe_transport(read, write).is_none()); + } + + #[cfg(feature = "server")] + #[tokio::test] + async fn stdio_pipes_builds_an_async_rw_transport() { + use super::stdio_pipes; + use crate::{RoleServer, transport::async_rw::AsyncRwTransport}; + + let (reader, writer) = stdio_pipes(); + let _transport: AsyncRwTransport = AsyncRwTransport::new(reader, writer); + } + + #[cfg(windows)] + #[tokio::test] + async fn stdio_pipes_falls_back_to_blocking_handles_on_windows() { + use super::{StdioReader, StdioWriter, stdio_pipes}; + + let (reader, writer) = stdio_pipes(); + assert!(matches!(reader, StdioReader::Blocking(_))); + assert!(matches!(writer, StdioWriter::Blocking(_))); + } +} From f2e2d6bfc46fe576a672369ba3ab13e854949691 Mon Sep 17 00:00:00 2001 From: itsmunzir Date: Tue, 29 Sep 2026 09:34:31 +0300 Subject: [PATCH 2/4] fix(transport): reopen the standard streams on Linux try_clone_to_owned duplicates the descriptor, so the duplicate shares its open file description with fd 0 or 1. Tokio's O_NONBLOCK then leaked into the process's own standard streams, and into children that inherit them. stdio_pipes() now opens /proc/self/fd/0 and /1 with O_NONBLOCK on Linux. The kernel gives the new descriptors their own file descriptions, so the flags stay private to the transport. macOS has no /proc, so it keeps the blocking fallback; the pipe variant is Linux-only now. Tests: reopening a FIFO keeps the source descriptor blocking, descriptor 0 keeps its flags across stdio_pipes(), and the round-trip and non-FIFO cases moved to Linux. Signed-off-by: itsmunzir --- crates/rmcp/Cargo.toml | 4 + crates/rmcp/src/transport/io.rs | 171 +++++++++++++++++++++++--------- 2 files changed, 128 insertions(+), 47 deletions(-) diff --git a/crates/rmcp/Cargo.toml b/crates/rmcp/Cargo.toml index 101a1ef64..ebb97cd9e 100644 --- a/crates/rmcp/Cargo.toml +++ b/crates/rmcp/Cargo.toml @@ -121,6 +121,10 @@ chrono = { version = "0.4.38", default-features = false, features = [ "oldtime", ] } +# the stdio pipe transport reopens `/proc/self/fd` with `O_NONBLOCK` +[target.'cfg(target_os = "linux")'.dependencies] +libc = "0.2" + [features] default = ["base64", "macros", "server"] local = ["rmcp-macros?/local"] diff --git a/crates/rmcp/src/transport/io.rs b/crates/rmcp/src/transport/io.rs index 381c6f872..3aaba2cd7 100644 --- a/crates/rmcp/src/transport/io.rs +++ b/crates/rmcp/src/transport/io.rs @@ -12,33 +12,32 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; /// /// Every read from and write to these handles is executed on Tokio's blocking /// thread pool. When both standard streams are pipes, prefer [`stdio_pipes`], -/// which avoids the per-operation thread hand-off on Unix. +/// which avoids the per-operation thread hand-off on Linux. pub fn stdio() -> (tokio::io::Stdin, tokio::io::Stdout) { (tokio::io::stdin(), tokio::io::stdout()) } /// # Non-blocking StdIO Transport /// -/// Like [`stdio`], but uses Tokio's non-blocking Unix pipe driver when both -/// standard streams are FIFOs — the way MCP clients launch stdio servers — so -/// reads and writes are driven by the IO reactor instead of Tokio's blocking -/// thread pool. This avoids a cross-thread hand-off per message under +/// Like [`stdio`], but drives the standard streams through Tokio's +/// non-blocking pipe driver when both are FIFOs — the way MCP clients launch +/// stdio servers — so reads and writes run on the IO reactor instead of the +/// blocking thread pool. This avoids a cross-thread hand-off per message under /// concurrent requests. /// -/// Falls back to the blocking [`stdio`] handles when the streams are not -/// pipes (TTYs, files, sockets) or on platforms without Unix pipes, so it can -/// be used unconditionally. -/// -/// On Unix the returned reader and writer own descriptors duplicated from -/// standard input and output; closing them does not close the process's -/// standard streams. +/// On Linux the pipe descriptors are reopened through `/proc/self/fd`, which +/// gives them their own open file descriptions: the process's own standard +/// streams, and any child process that inherits them, keep their blocking +/// flags. Every other platform (macOS, Windows) and any stream that is not a +/// FIFO falls back to the blocking [`stdio`] handles, so this can be used +/// unconditionally. /// /// # Panics /// -/// On Unix, panics if called outside of a Tokio runtime with IO enabled, +/// On Linux, panics if called outside of a Tokio runtime with IO enabled, /// mirroring `tokio::net::unix::pipe::Receiver::from_owned_fd`. pub fn stdio_pipes() -> (StdioReader, StdioWriter) { - #[cfg(unix)] + #[cfg(target_os = "linux")] if let Some(pair) = try_pipe_transport() { return pair; } @@ -48,37 +47,54 @@ pub fn stdio_pipes() -> (StdioReader, StdioWriter) { ) } -#[cfg(unix)] +/// Takes the pipe fast path only when both standard streams are FIFOs. +/// +/// The descriptors are reopened through `/proc/self/fd` rather than +/// duplicated. A duplicate shares its open file description with the original, +/// and `/proc/self/fd` opens a fresh one, so `O_NONBLOCK` stays private to the +/// transport instead of leaking into the process's standard streams (and into +/// children spawned with the inherited handles). +#[cfg(target_os = "linux")] fn try_pipe_transport() -> Option<(StdioReader, StdioWriter)> { use std::os::fd::AsFd; - let read = std::io::stdin().as_fd().try_clone_to_owned().ok()?; - let write = std::io::stdout().as_fd().try_clone_to_owned().ok()?; + let stdin = std::io::stdin().as_fd().try_clone_to_owned().ok()?; + if !is_fifo(&stdin) { + return None; + } + let stdout = std::io::stdout().as_fd().try_clone_to_owned().ok()?; + if !is_fifo(&stdout) { + return None; + } + + let read = reopen_nonblocking("/proc/self/fd/0", false)?; + let write = reopen_nonblocking("/proc/self/fd/1", true)?; pipe_transport(read, write) } +/// Reopens `path` with `O_NONBLOCK`, on its own open file description. +#[cfg(target_os = "linux")] +fn reopen_nonblocking(path: &str, write: bool) -> Option { + use std::os::unix::fs::OpenOptionsExt; + + let mut options = std::fs::OpenOptions::new(); + options.read(!write).write(write); + options.custom_flags(libc::O_NONBLOCK); + options.open(path).ok().map(Into::into) +} + /// Wraps two FIFO descriptors as a non-blocking pipe transport. -/// -/// Returns `None` unless both descriptors are FIFOs. Both are checked before -/// either is handed to Tokio: `pipe::Receiver::from_owned_fd` and -/// `pipe::Sender::from_owned_fd` put the descriptor into non-blocking mode, -/// and `O_NONBLOCK` is shared by every descriptor of the same open file -/// description, so a partial conversion would leave the blocking fallback -/// with non-blocking stdio. -#[cfg(unix)] +#[cfg(target_os = "linux")] fn pipe_transport( read: std::os::fd::OwnedFd, write: std::os::fd::OwnedFd, ) -> Option<(StdioReader, StdioWriter)> { - if !is_fifo(&read) || !is_fifo(&write) { - return None; - } let read = tokio::net::unix::pipe::Receiver::from_owned_fd(read).ok()?; let write = tokio::net::unix::pipe::Sender::from_owned_fd(write).ok()?; Some((StdioReader::Pipe(read), StdioWriter::Pipe(write))) } -#[cfg(unix)] +#[cfg(target_os = "linux")] fn is_fifo(fd: &std::os::fd::OwnedFd) -> bool { use std::os::unix::fs::FileTypeExt; @@ -93,8 +109,8 @@ fn is_fifo(fd: &std::os::fd::OwnedFd) -> bool { pub enum StdioReader { /// Blocking standard input, as returned by [`stdio`]. Blocking(tokio::io::Stdin), - /// Non-blocking standard input backed by a Unix pipe. - #[cfg(unix)] + /// Non-blocking standard input backed by a Linux pipe. + #[cfg(target_os = "linux")] Pipe(tokio::net::unix::pipe::Receiver), } @@ -106,7 +122,7 @@ impl AsyncRead for StdioReader { ) -> Poll> { match self.get_mut() { Self::Blocking(reader) => Pin::new(reader).poll_read(cx, buf), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(reader) => Pin::new(reader).poll_read(cx, buf), } } @@ -117,8 +133,8 @@ impl AsyncRead for StdioReader { pub enum StdioWriter { /// Blocking standard output, as returned by [`stdio`]. Blocking(tokio::io::Stdout), - /// Non-blocking standard output backed by a Unix pipe. - #[cfg(unix)] + /// Non-blocking standard output backed by a Linux pipe. + #[cfg(target_os = "linux")] Pipe(tokio::net::unix::pipe::Sender), } @@ -130,7 +146,7 @@ impl AsyncWrite for StdioWriter { ) -> Poll> { match self.get_mut() { Self::Blocking(writer) => Pin::new(writer).poll_write(cx, buf), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(writer) => Pin::new(writer).poll_write(cx, buf), } } @@ -138,7 +154,7 @@ impl AsyncWrite for StdioWriter { fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Self::Blocking(writer) => Pin::new(writer).poll_flush(cx), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(writer) => Pin::new(writer).poll_flush(cx), } } @@ -146,7 +162,7 @@ impl AsyncWrite for StdioWriter { fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { match self.get_mut() { Self::Blocking(writer) => Pin::new(writer).poll_shutdown(cx), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(writer) => Pin::new(writer).poll_shutdown(cx), } } @@ -158,7 +174,7 @@ impl AsyncWrite for StdioWriter { ) -> Poll> { match self.get_mut() { Self::Blocking(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), } } @@ -166,7 +182,7 @@ impl AsyncWrite for StdioWriter { fn is_write_vectored(&self) -> bool { match self { Self::Blocking(writer) => writer.is_write_vectored(), - #[cfg(unix)] + #[cfg(target_os = "linux")] Self::Pipe(writer) => writer.is_write_vectored(), } } @@ -174,7 +190,7 @@ impl AsyncWrite for StdioWriter { #[cfg(test)] mod tests { - #[cfg(unix)] + #[cfg(target_os = "linux")] fn fifo_pair() -> ( std::process::Child, std::os::fd::OwnedFd, @@ -192,7 +208,13 @@ mod tests { (child, read, write) } - #[cfg(unix)] + #[cfg(target_os = "linux")] + fn open_flags(fd: std::os::fd::RawFd) -> i32 { + // SAFETY: every caller passes a live descriptor and F_GETFL only reads. + unsafe { libc::fcntl(fd, libc::F_GETFL) } + } + + #[cfg(target_os = "linux")] #[tokio::test] async fn pipe_transport_round_trips_over_fifos() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -215,14 +237,69 @@ mod tests { assert!(child.wait().unwrap().success()); } - #[cfg(unix)] + /// The reopen exists to keep `O_NONBLOCK` off the process's own standard + /// streams, so pin the property on a FIFO: the reopened descriptor is + /// non-blocking and the descriptor it came from stays blocking. + #[cfg(target_os = "linux")] + #[tokio::test] + async fn reopening_a_fifo_keeps_the_original_flags() { + use std::os::fd::AsRawFd; + + use super::{is_fifo, reopen_nonblocking}; + + let path = std::env::temp_dir().join(format!("rmcp-fifo-reopen-{}", std::process::id())); + let _ = std::fs::remove_file(&path); + assert!( + std::process::Command::new("mkfifo") + .arg(&path) + .status() + .expect("failed to run `mkfifo`") + .success() + ); + + // O_RDWR on a FIFO does not wait for a peer, so the reopen cannot block. + let original = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .expect("failed to open the fifo"); + assert_eq!(open_flags(original.as_raw_fd()) & libc::O_NONBLOCK, 0); + + let reopened = + reopen_nonblocking(&format!("/proc/self/fd/{}", original.as_raw_fd()), false) + .expect("failed to reopen the fifo"); + assert!(is_fifo(&reopened)); + assert_ne!(open_flags(reopened.as_raw_fd()) & libc::O_NONBLOCK, 0); + assert_eq!(open_flags(original.as_raw_fd()) & libc::O_NONBLOCK, 0); + + drop(reopened); + drop(original); + let _ = std::fs::remove_file(&path); + } + + /// The check the reviewer asked for: descriptor 0 keeps its flags. + #[cfg(target_os = "linux")] + #[tokio::test] + async fn stdio_pipes_leaves_descriptor_zero_flags_alone() { + use std::os::fd::AsRawFd; + + use super::stdio_pipes; + + let before = open_flags(std::io::stdin().as_raw_fd()); + let _transport = stdio_pipes(); + let after = open_flags(std::io::stdin().as_raw_fd()); + assert_eq!(before, after); + } + + #[cfg(target_os = "linux")] #[tokio::test] - async fn pipe_transport_falls_back_for_non_fifos() { - use super::pipe_transport; + async fn fifo_detection_rejects_character_devices() { + use super::is_fifo; - let read = std::fs::File::open("/dev/null").unwrap().into(); - let write = std::fs::File::open("/dev/null").unwrap().into(); - assert!(pipe_transport(read, write).is_none()); + let dev_null: std::os::fd::OwnedFd = std::fs::File::open("/dev/null") + .expect("failed to open /dev/null") + .into(); + assert!(!is_fifo(&dev_null)); } #[cfg(feature = "server")] From 8957ae10c274be02250fb326f33269ca123ccfaf Mon Sep 17 00:00:00 2001 From: itsmunzir Date: Mon, 12 Oct 2026 10:06:18 +0300 Subject: [PATCH 3/4] fix(transport): address review feedback on stdio_pipes - Reopen fd 0 and 1 first, then check the FIFOs on the descriptors that go to Tokio. - Keep the platform-specific handle in private enums with Debug. Callers only need AsyncRead/AsyncWrite. - Log the taken path with tracing::debug!. - Drop the vacuous stdin-flags test; the owned-FIFO round trip covers the reopen path on the runner. Signed-off-by: itsmunzir --- crates/rmcp/src/transport/io.rs | 179 +++++++++++++++++++++----------- 1 file changed, 118 insertions(+), 61 deletions(-) diff --git a/crates/rmcp/src/transport/io.rs b/crates/rmcp/src/transport/io.rs index 3aaba2cd7..7a7d08fcf 100644 --- a/crates/rmcp/src/transport/io.rs +++ b/crates/rmcp/src/transport/io.rs @@ -38,37 +38,48 @@ pub fn stdio() -> (tokio::io::Stdin, tokio::io::Stdout) { /// mirroring `tokio::net::unix::pipe::Receiver::from_owned_fd`. pub fn stdio_pipes() -> (StdioReader, StdioWriter) { #[cfg(target_os = "linux")] - if let Some(pair) = try_pipe_transport() { - return pair; + { + use std::os::fd::AsRawFd; + + if let Some(pair) = + reopen_fifo_pair(std::io::stdin().as_raw_fd(), std::io::stdout().as_raw_fd()) + { + tracing::debug!("stdio_pipes: using the non-blocking Linux pipe transport"); + return pair; + } + tracing::debug!("stdio_pipes: falling back to the blocking stdio handles"); } ( - StdioReader::Blocking(tokio::io::stdin()), - StdioWriter::Blocking(tokio::io::stdout()), + StdioReader { + imp: StdioReaderImp::Blocking(tokio::io::stdin()), + }, + StdioWriter { + imp: StdioWriterImp::Blocking(tokio::io::stdout()), + }, ) } -/// Takes the pipe fast path only when both standard streams are FIFOs. +/// Reopens the given descriptors through `/proc/self/fd` and wraps them when +/// both reopened descriptors are FIFOs. /// -/// The descriptors are reopened through `/proc/self/fd` rather than -/// duplicated. A duplicate shares its open file description with the original, -/// and `/proc/self/fd` opens a fresh one, so `O_NONBLOCK` stays private to the -/// transport instead of leaking into the process's standard streams (and into +/// The reopened descriptors are the ones that are checked and the ones that +/// are handed to Tokio; the originals are only read. The reopen gives the +/// transport its own open file descriptions, so `O_NONBLOCK` stays private to +/// it instead of leaking into the process's standard streams (and into /// children spawned with the inherited handles). #[cfg(target_os = "linux")] -fn try_pipe_transport() -> Option<(StdioReader, StdioWriter)> { - use std::os::fd::AsFd; - - let stdin = std::io::stdin().as_fd().try_clone_to_owned().ok()?; - if !is_fifo(&stdin) { +fn reopen_fifo_pair( + read_fd: std::os::fd::RawFd, + write_fd: std::os::fd::RawFd, +) -> Option<(StdioReader, StdioWriter)> { + let read = reopen_nonblocking(&format!("/proc/self/fd/{read_fd}"), false)?; + if !is_fifo(&read) { return None; } - let stdout = std::io::stdout().as_fd().try_clone_to_owned().ok()?; - if !is_fifo(&stdout) { + let write = reopen_nonblocking(&format!("/proc/self/fd/{write_fd}"), true)?; + if !is_fifo(&write) { return None; } - - let read = reopen_nonblocking("/proc/self/fd/0", false)?; - let write = reopen_nonblocking("/proc/self/fd/1", true)?; pipe_transport(read, write) } @@ -91,7 +102,14 @@ fn pipe_transport( ) -> Option<(StdioReader, StdioWriter)> { let read = tokio::net::unix::pipe::Receiver::from_owned_fd(read).ok()?; let write = tokio::net::unix::pipe::Sender::from_owned_fd(write).ok()?; - Some((StdioReader::Pipe(read), StdioWriter::Pipe(write))) + Some(( + StdioReader { + imp: StdioReaderImp::Pipe(read), + }, + StdioWriter { + imp: StdioWriterImp::Pipe(write), + }, + )) } #[cfg(target_os = "linux")] @@ -105,8 +123,16 @@ fn is_fifo(fd: &std::os::fd::OwnedFd) -> bool { } /// Reader half of the transport returned by [`stdio_pipes`]. -#[non_exhaustive] -pub enum StdioReader { +/// +/// Implements [`AsyncRead`]. The underlying handle depends on the platform and +/// on whether standard input is a pipe. +#[derive(Debug)] +pub struct StdioReader { + imp: StdioReaderImp, +} + +#[derive(Debug)] +enum StdioReaderImp { /// Blocking standard input, as returned by [`stdio`]. Blocking(tokio::io::Stdin), /// Non-blocking standard input backed by a Linux pipe. @@ -120,17 +146,25 @@ impl AsyncRead for StdioReader { cx: &mut Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll> { - match self.get_mut() { - Self::Blocking(reader) => Pin::new(reader).poll_read(cx, buf), + match &mut self.get_mut().imp { + StdioReaderImp::Blocking(reader) => Pin::new(reader).poll_read(cx, buf), #[cfg(target_os = "linux")] - Self::Pipe(reader) => Pin::new(reader).poll_read(cx, buf), + StdioReaderImp::Pipe(reader) => Pin::new(reader).poll_read(cx, buf), } } } /// Writer half of the transport returned by [`stdio_pipes`]. -#[non_exhaustive] -pub enum StdioWriter { +/// +/// Implements [`AsyncWrite`]. The underlying handle depends on the platform and +/// on whether standard output is a pipe. +#[derive(Debug)] +pub struct StdioWriter { + imp: StdioWriterImp, +} + +#[derive(Debug)] +enum StdioWriterImp { /// Blocking standard output, as returned by [`stdio`]. Blocking(tokio::io::Stdout), /// Non-blocking standard output backed by a Linux pipe. @@ -144,26 +178,26 @@ impl AsyncWrite for StdioWriter { cx: &mut Context<'_>, buf: &[u8], ) -> Poll> { - match self.get_mut() { - Self::Blocking(writer) => Pin::new(writer).poll_write(cx, buf), + match &mut self.get_mut().imp { + StdioWriterImp::Blocking(writer) => Pin::new(writer).poll_write(cx, buf), #[cfg(target_os = "linux")] - Self::Pipe(writer) => Pin::new(writer).poll_write(cx, buf), + StdioWriterImp::Pipe(writer) => Pin::new(writer).poll_write(cx, buf), } } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match self.get_mut() { - Self::Blocking(writer) => Pin::new(writer).poll_flush(cx), + match &mut self.get_mut().imp { + StdioWriterImp::Blocking(writer) => Pin::new(writer).poll_flush(cx), #[cfg(target_os = "linux")] - Self::Pipe(writer) => Pin::new(writer).poll_flush(cx), + StdioWriterImp::Pipe(writer) => Pin::new(writer).poll_flush(cx), } } fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - match self.get_mut() { - Self::Blocking(writer) => Pin::new(writer).poll_shutdown(cx), + match &mut self.get_mut().imp { + StdioWriterImp::Blocking(writer) => Pin::new(writer).poll_shutdown(cx), #[cfg(target_os = "linux")] - Self::Pipe(writer) => Pin::new(writer).poll_shutdown(cx), + StdioWriterImp::Pipe(writer) => Pin::new(writer).poll_shutdown(cx), } } @@ -172,18 +206,18 @@ impl AsyncWrite for StdioWriter { cx: &mut Context<'_>, bufs: &[io::IoSlice<'_>], ) -> Poll> { - match self.get_mut() { - Self::Blocking(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), + match &mut self.get_mut().imp { + StdioWriterImp::Blocking(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), #[cfg(target_os = "linux")] - Self::Pipe(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), + StdioWriterImp::Pipe(writer) => Pin::new(writer).poll_write_vectored(cx, bufs), } } fn is_write_vectored(&self) -> bool { - match self { - Self::Blocking(writer) => writer.is_write_vectored(), + match &self.imp { + StdioWriterImp::Blocking(writer) => writer.is_write_vectored(), #[cfg(target_os = "linux")] - Self::Pipe(writer) => writer.is_write_vectored(), + StdioWriterImp::Pipe(writer) => writer.is_write_vectored(), } } } @@ -219,12 +253,12 @@ mod tests { async fn pipe_transport_round_trips_over_fifos() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use super::{StdioReader, StdioWriter, pipe_transport}; + use super::{StdioReaderImp, StdioWriterImp, pipe_transport}; let (mut child, read, write) = fifo_pair(); let (mut reader, mut writer) = pipe_transport(read, write).expect("child pipes are FIFOs"); - assert!(matches!(reader, StdioReader::Pipe(_))); - assert!(matches!(writer, StdioWriter::Pipe(_))); + assert!(matches!(reader.imp, StdioReaderImp::Pipe(_))); + assert!(matches!(writer.imp, StdioWriterImp::Pipe(_))); let message = b"{\"jsonrpc\":\"2.0\",\"id\":1}\n"; writer.write_all(message).await.unwrap(); @@ -241,8 +275,8 @@ mod tests { /// streams, so pin the property on a FIFO: the reopened descriptor is /// non-blocking and the descriptor it came from stays blocking. #[cfg(target_os = "linux")] - #[tokio::test] - async fn reopening_a_fifo_keeps_the_original_flags() { + #[test] + fn reopening_a_fifo_keeps_the_original_flags() { use std::os::fd::AsRawFd; use super::{is_fifo, reopen_nonblocking}; @@ -277,23 +311,46 @@ mod tests { let _ = std::fs::remove_file(&path); } - /// The check the reviewer asked for: descriptor 0 keeps its flags. + /// The reopen path is taken on real pipes: CI runs `cargo test` without + /// piped standard streams, so build the FIFOs here. The reopened pair must + /// round-trip, and the descriptors it came from must keep their flags. #[cfg(target_os = "linux")] #[tokio::test] - async fn stdio_pipes_leaves_descriptor_zero_flags_alone() { + async fn reopened_fifo_pair_round_trips_and_keeps_originals_blocking() { use std::os::fd::AsRawFd; - use super::stdio_pipes; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let before = open_flags(std::io::stdin().as_raw_fd()); - let _transport = stdio_pipes(); - let after = open_flags(std::io::stdin().as_raw_fd()); - assert_eq!(before, after); + use super::{StdioReaderImp, StdioWriterImp, reopen_fifo_pair}; + + let (mut child, read, write) = fifo_pair(); + let read_flags = open_flags(read.as_raw_fd()); + let write_flags = open_flags(write.as_raw_fd()); + assert_eq!(read_flags & libc::O_NONBLOCK, 0); + assert_eq!(write_flags & libc::O_NONBLOCK, 0); + + let (mut reader, mut writer) = + reopen_fifo_pair(read.as_raw_fd(), write.as_raw_fd()).expect("child pipes are FIFOs"); + assert!(matches!(reader.imp, StdioReaderImp::Pipe(_))); + assert!(matches!(writer.imp, StdioWriterImp::Pipe(_))); + + let message = b"{\"jsonrpc\":\"2.0\",\"id\":2}\n"; + writer.write_all(message).await.unwrap(); + writer.flush().await.unwrap(); + let mut echoed = vec![0; message.len()]; + reader.read_exact(&mut echoed).await.unwrap(); + assert_eq!(echoed, message); + + assert_eq!(open_flags(read.as_raw_fd()), read_flags); + assert_eq!(open_flags(write.as_raw_fd()), write_flags); + + drop(writer); + assert!(child.wait().unwrap().success()); } #[cfg(target_os = "linux")] - #[tokio::test] - async fn fifo_detection_rejects_character_devices() { + #[test] + fn fifo_detection_rejects_character_devices() { use super::is_fifo; let dev_null: std::os::fd::OwnedFd = std::fs::File::open("/dev/null") @@ -312,13 +369,13 @@ mod tests { let _transport: AsyncRwTransport = AsyncRwTransport::new(reader, writer); } - #[cfg(windows)] + #[cfg(not(target_os = "linux"))] #[tokio::test] - async fn stdio_pipes_falls_back_to_blocking_handles_on_windows() { - use super::{StdioReader, StdioWriter, stdio_pipes}; + async fn stdio_pipes_falls_back_to_blocking_handles_off_linux() { + use super::{StdioReaderImp, StdioWriterImp, stdio_pipes}; let (reader, writer) = stdio_pipes(); - assert!(matches!(reader, StdioReader::Blocking(_))); - assert!(matches!(writer, StdioWriter::Blocking(_))); + assert!(matches!(reader.imp, StdioReaderImp::Blocking(_))); + assert!(matches!(writer.imp, StdioWriterImp::Blocking(_))); } } From 175495be13d0b6dad94433eb2b595ce98786760f Mon Sep 17 00:00:00 2001 From: itsmunzir Date: Wed, 14 Oct 2026 13:40:19 +0300 Subject: [PATCH 4/4] fix(transport): let the FIFO child see EOF in the round-trip test Signed-off-by: itsmunzir --- crates/rmcp/src/transport/io.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/rmcp/src/transport/io.rs b/crates/rmcp/src/transport/io.rs index 7a7d08fcf..666228407 100644 --- a/crates/rmcp/src/transport/io.rs +++ b/crates/rmcp/src/transport/io.rs @@ -344,7 +344,12 @@ mod tests { assert_eq!(open_flags(read.as_raw_fd()), read_flags); assert_eq!(open_flags(write.as_raw_fd()), write_flags); + // The originals hold the pipe's ends too, and a live write end would + // keep `cat` from ever seeing EOF. + drop(reader); drop(writer); + drop(read); + drop(write); assert!(child.wait().unwrap().success()); }