deno/ext/io/pipe.rs

// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
use std::io;
use std::pin::Pin;
use std::process::Stdio;

pub type RawPipeHandle = super::RawIoHandle;

// The synchronous read end of a unidirectional pipe.
pub struct PipeRead {
  file: std::fs::File,
}

// The asynchronous read end of a unidirectional pipe.
pub struct AsyncPipeRead {
  #[cfg(windows)]
  /// We use a `ChildStdout` here as it's a much better fit for a Windows named pipe on Windows. We
  /// might also be able to use `tokio::net::windows::named_pipe::NamedPipeClient` in the future
  /// if those can be created from raw handles down the road.
  read: tokio::process::ChildStdout,
  #[cfg(not(windows))]
  read: tokio::net::unix::pipe::Receiver,
}

// The synchronous write end of a unidirectional pipe.
pub struct PipeWrite {
  file: std::fs::File,
}

// The asynchronous write end of a unidirectional pipe.
pub struct AsyncPipeWrite {
  #[cfg(windows)]
  /// We use a `ChildStdin` here as it's a much better fit for a Windows named pipe on Windows. We
  /// might also be able to use `tokio::net::windows::named_pipe::NamedPipeClient` in the future
  /// if those can be created from raw handles down the road.
  write: tokio::process::ChildStdin,
  #[cfg(not(windows))]
  write: tokio::net::unix::pipe::Sender,
}

impl PipeRead {
  /// Converts this sync reader into an async reader. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(windows)]
  pub fn into_async(self) -> io::Result<AsyncPipeRead> {
    let owned: std::os::windows::io::OwnedHandle = self.file.into();
    let stdout = std::process::ChildStdout::from(owned);
    Ok(AsyncPipeRead {
      read: tokio::process::ChildStdout::from_std(stdout)?,
    })
  }

  /// Converts this sync reader into an async reader. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(not(windows))]
  pub fn into_async(self) -> io::Result<AsyncPipeRead> {
    Ok(AsyncPipeRead {
      read: tokio::net::unix::pipe::Receiver::from_file(self.file)?,
    })
  }

  /// Creates a new [`PipeRead`] instance that shares the same underlying file handle
  /// as the existing [`PipeRead`] instance.
  pub fn try_clone(&self) -> io::Result<Self> {
    Ok(Self {
      file: self.file.try_clone()?,
    })
  }
}

impl AsyncPipeRead {
  /// Converts this async reader into an sync reader. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(windows)]
  pub fn into_sync(self) -> io::Result<PipeRead> {
    let owned = self.read.into_owned_handle()?;
    Ok(PipeRead { file: owned.into() })
  }

  /// Converts this async reader into an sync reader. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(not(windows))]
  pub fn into_sync(self) -> io::Result<PipeRead> {
    let file = self.read.into_nonblocking_fd()?.into();
    Ok(PipeRead { file })
  }
}

impl std::io::Read for PipeRead {
  fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
    self.file.read(buf)
  }

  fn read_vectored(
    &mut self,
    bufs: &mut [io::IoSliceMut<'_>],
  ) -> io::Result<usize> {
    self.file.read_vectored(bufs)
  }
}

impl tokio::io::AsyncRead for AsyncPipeRead {
  fn poll_read(
    self: Pin<&mut Self>,
    cx: &mut std::task::Context<'_>,
    buf: &mut tokio::io::ReadBuf<'_>,
  ) -> std::task::Poll<io::Result<()>> {
    Pin::new(&mut self.get_mut().read).poll_read(cx, buf)
  }
}

impl PipeWrite {
  /// Converts this sync writer into an async writer. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(windows)]
  pub fn into_async(self) -> io::Result<AsyncPipeWrite> {
    let owned: std::os::windows::io::OwnedHandle = self.file.into();
    let stdin = std::process::ChildStdin::from(owned);
    Ok(AsyncPipeWrite {
      write: tokio::process::ChildStdin::from_std(stdin)?,
    })
  }

  /// Converts this sync writer into an async writer. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(not(windows))]
  pub fn into_async(self) -> io::Result<AsyncPipeWrite> {
    Ok(AsyncPipeWrite {
      write: tokio::net::unix::pipe::Sender::from_file(self.file)?,
    })
  }

  /// Creates a new [`PipeWrite`] instance that shares the same underlying file handle
  /// as the existing [`PipeWrite`] instance.
  pub fn try_clone(&self) -> io::Result<Self> {
    Ok(Self {
      file: self.file.try_clone()?,
    })
  }
}

impl AsyncPipeWrite {
  /// Converts this async writer into an sync writer. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(windows)]
  pub fn into_sync(self) -> io::Result<PipeWrite> {
    let owned = self.write.into_owned_handle()?;
    Ok(PipeWrite { file: owned.into() })
  }

  /// Converts this async writer into an sync writer. May fail if the Tokio runtime is
  /// unavailable.
  #[cfg(not(windows))]
  pub fn into_sync(self) -> io::Result<PipeWrite> {
    let file = self.write.into_nonblocking_fd()?.into();
    Ok(PipeWrite { file })
  }
}

impl std::io::Write for PipeWrite {
  fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
    self.file.write(buf)
  }

  fn flush(&mut self) -> io::Result<()> {
    self.file.flush()
  }

  fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
    self.file.write_vectored(bufs)
  }
}

impl tokio::io::AsyncWrite for AsyncPipeWrite {
  #[inline(always)]
  fn poll_write(
    self: std::pin::Pin<&mut Self>,
    cx: &mut std::task::Context<'_>,
    buf: &[u8],
  ) -> std::task::Poll<Result<usize, io::Error>> {
    Pin::new(&mut self.get_mut().write).poll_write(cx, buf)
  }

  #[inline(always)]
  fn poll_flush(
    self: Pin<&mut Self>,
    cx: &mut std::task::Context<'_>,
  ) -> std::task::Poll<Result<(), io::Error>> {
    Pin::new(&mut self.get_mut().write).poll_flush(cx)
  }

  #[inline(always)]
  fn poll_shutdown(
    self: Pin<&mut Self>,
    cx: &mut std::task::Context<'_>,
  ) -> std::task::Poll<Result<(), io::Error>> {
    Pin::new(&mut self.get_mut().write).poll_shutdown(cx)
  }

  #[inline(always)]
  fn is_write_vectored(&self) -> bool {
    self.write.is_write_vectored()
  }

  #[inline(always)]
  fn poll_write_vectored(
    self: Pin<&mut Self>,
    cx: &mut std::task::Context<'_>,
    bufs: &[io::IoSlice<'_>],
  ) -> std::task::Poll<Result<usize, io::Error>> {
    Pin::new(&mut self.get_mut().write).poll_write_vectored(cx, bufs)
  }
}

impl From<PipeRead> for Stdio {
  fn from(val: PipeRead) -> Self {
    Stdio::from(val.file)
  }
}

impl From<PipeWrite> for Stdio {
  fn from(val: PipeWrite) -> Self {
    Stdio::from(val.file)
  }
}

impl From<PipeRead> for std::fs::File {
  fn from(val: PipeRead) -> Self {
    val.file
  }
}

impl From<PipeWrite> for std::fs::File {
  fn from(val: PipeWrite) -> Self {
    val.file
  }
}

#[cfg(not(windows))]
impl From<PipeRead> for std::os::unix::io::OwnedFd {
  fn from(val: PipeRead) -> Self {
    val.file.into()
  }
}

#[cfg(not(windows))]
impl From<PipeWrite> for std::os::unix::io::OwnedFd {
  fn from(val: PipeWrite) -> Self {
    val.file.into()
  }
}

#[cfg(windows)]
impl From<PipeRead> for std::os::windows::io::OwnedHandle {
  fn from(val: PipeRead) -> Self {
    val.file.into()
  }
}

#[cfg(windows)]
impl From<PipeWrite> for std::os::windows::io::OwnedHandle {
  fn from(val: PipeWrite) -> Self {
    val.file.into()
  }
}

/// Create a unidirectional pipe pair that starts off as a pair of synchronous file handles,
/// but either side may be promoted to an async-capable reader/writer.
///
/// On Windows, we use a named pipe because that's the only way to get reliable async I/O
/// support. On Unix platforms, we use the `os_pipe` library, which uses `pipe2` under the hood
/// (or `pipe` on OSX).
pub fn pipe() -> io::Result<(PipeRead, PipeWrite)> {
  pipe_impl()
}

/// Creates a unidirectional pipe on top of a named pipe (which is technically bidirectional).
#[cfg(windows)]
pub fn pipe_impl() -> io::Result<(PipeRead, PipeWrite)> {
  // SAFETY: We're careful with handles here
  unsafe {
    use std::os::windows::io::FromRawHandle;
    use std::os::windows::io::OwnedHandle;
    let (server, client) = crate::winpipe::create_named_pipe()?;
    let read = std::fs::File::from(OwnedHandle::from_raw_handle(client));
    let write = std::fs::File::from(OwnedHandle::from_raw_handle(server));
    Ok((PipeRead { file: read }, PipeWrite { file: write }))
  }
}

/// Creates a unidirectional pipe for unix platforms.
#[cfg(not(windows))]
pub fn pipe_impl() -> io::Result<(PipeRead, PipeWrite)> {
  use std::os::unix::io::OwnedFd;
  let (read, write) = os_pipe::pipe()?;
  let read = std::fs::File::from(Into::<OwnedFd>::into(read));
  let write = std::fs::File::from(Into::<OwnedFd>::into(write));
  Ok((PipeRead { file: read }, PipeWrite { file: write }))
}

#[cfg(test)]
mod tests {
  use super::*;

  use std::io::Read;
  use std::io::Write;
  use tokio::io::AsyncReadExt;
  use tokio::io::AsyncWriteExt;

  #[test]
  fn test_pipe() {
    let (mut read, mut write) = pipe().unwrap();
    // Write to the server and read from the client
    write.write_all(b"hello").unwrap();
    let mut buf: [u8; 5] = Default::default();
    read.read_exact(&mut buf).unwrap();
    assert_eq!(&buf, b"hello");
  }

  #[tokio::test]
  async fn test_async_pipe() {
    let (read, write) = pipe().unwrap();
    let mut read = read.into_async().unwrap();
    let mut write = write.into_async().unwrap();

    write.write_all(b"hello").await.unwrap();
    let mut buf: [u8; 5] = Default::default();
    read.read_exact(&mut buf).await.unwrap();
    assert_eq!(&buf, b"hello");
  }

  /// Test a round-trip through async mode and back.
  #[tokio::test]
  async fn test_pipe_transmute() {
    let (mut read, mut write) = pipe().unwrap();

    // Sync
    write.write_all(b"hello").unwrap();
    let mut buf: [u8; 5] = Default::default();
    read.read_exact(&mut buf).unwrap();
    assert_eq!(&buf, b"hello");

    let mut read = read.into_async().unwrap();
    let mut write = write.into_async().unwrap();

    // Async
    write.write_all(b"hello").await.unwrap();
    let mut buf: [u8; 5] = Default::default();
    read.read_exact(&mut buf).await.unwrap();
    assert_eq!(&buf, b"hello");

    let mut read = read.into_sync().unwrap();
    let mut write = write.into_sync().unwrap();

    // Sync
    write.write_all(b"hello").unwrap();
    let mut buf: [u8; 5] = Default::default();
    read.read_exact(&mut buf).unwrap();
    assert_eq!(&buf, b"hello");
  }

  #[tokio::test]
  async fn test_async_pipe_is_nonblocking() {
    let (read, write) = pipe().unwrap();
    let mut read = read.into_async().unwrap();
    let mut write = write.into_async().unwrap();

    let a = tokio::spawn(async move {
      let mut buf: [u8; 5] = Default::default();
      read.read_exact(&mut buf).await.unwrap();
      assert_eq!(&buf, b"hello");
    });
    let b = tokio::spawn(async move {
      write.write_all(b"hello").await.unwrap();
    });

    a.await.unwrap();
    b.await.unwrap();
  }
}