src/io/pipe.rs

   1 use std::convert::TryFrom;
   2 use std::io;
   3 use std::marker::PhantomData;
   4 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
   5 use std::pin::Pin;
   6 use std::task::{Context, Poll};
   7
   8 use tokio::io::{AsyncRead, AsyncWrite};
   9
  10 use crate::error::io_err_other;
  11 use crate::io::polled_fd::PolledFd;
  12 use crate::io::rw_traits;
  13 use crate::tools::Fd;
  14
  15 pub use rw_traits::{Read, Write};
  16
  17 pub struct Pipe<RW> {
  18     fd: PolledFd,
  19     _phantom: PhantomData<RW>,
  20 }
  21
  22 impl<RW> AsRawFd for Pipe<RW> {
  23     #[inline]
  24     fn as_raw_fd(&self) -> RawFd {
  25         self.fd.as_raw_fd()
  26     }
  27 }
  28
  29 impl<RW> IntoRawFd for Pipe<RW> {
  30     #[inline]
  31     fn into_raw_fd(self) -> RawFd {
  32         self.fd.into_raw_fd()
  33     }
  34 }
  35
  36 pub fn pipe() -> io::Result<(Pipe<rw_traits::Read>, Pipe<rw_traits::Write>)> {
  37     let mut pfd: [RawFd; 2] = [0, 0];
  38
  39     c_try!(unsafe { libc::pipe2(pfd.as_mut_ptr(), libc::O_CLOEXEC) });
  40
  41     let (fd_in, fd_out) = unsafe { (Fd::from_raw_fd(pfd[0]), Fd::from_raw_fd(pfd[1])) };
  42
  43     Ok((
  44         Pipe {
  45             fd: PolledFd::new(fd_in)?,
  46             _phantom: PhantomData,
  47         },
  48         Pipe {
  49             fd: PolledFd::new(fd_out)?,
  50             _phantom: PhantomData,
  51         },
  52     ))
  53 }
  54
  55 impl<RW: rw_traits::HasRead> AsyncRead for Pipe<RW> {
  56     fn poll_read(
  57         self: Pin<&mut Self>,
  58         cx: &mut Context<'_>,
  59         buf: &mut [u8],
  60     ) -> Poll<io::Result<usize>> {
  61         self.fd.wrap_read(cx, || {
  62             let fd = self.as_raw_fd();
  63             let size = libc::size_t::try_from(buf.len()).map_err(io_err_other)?;
  64             c_result!(unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, size) })
  65                 .map(|res| res as usize)
  66         })
  67     }
  68 }
  69
  70 impl<RW: rw_traits::HasWrite> AsyncWrite for Pipe<RW> {
  71     fn poll_write(
  72         self: Pin<&mut Self>,
  73         cx: &mut Context<'_>,
  74         buf: &[u8],
  75     ) -> Poll<io::Result<usize>> {
  76         self.fd.wrap_write(cx, || {
  77             let fd = self.as_raw_fd();
  78             let size = libc::size_t::try_from(buf.len()).map_err(io_err_other)?;
  79             c_result!(unsafe { libc::write(fd, buf.as_ptr() as *const libc::c_void, size) })
  80                 .map(|res| res as usize)
  81         })
  82     }
  83
  84     fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
  85         Poll::Ready(Ok(()))
  86     }
  87
  88     fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
  89         Poll::Ready(Ok(()))
  90     }
  91 }