Add unix pipe.

compio-driver/src/iour/op.rs

@@ -21,7 +21,7 @@ impl<T: IoBufMut> OpCode for ReadAt<T> {
  let fd = Fd(self.fd);
  let slice = self.buffer.as_uninit_slice();
  opcode::Read::new(fd, slice.as_mut_ptr() as _, slice.len() as _)
- .offset(self.offset as _)
+ .offset(self.offset())
  .build()
  }
 }
@@ -30,7 +30,7 @@ impl<T: IoBuf> OpCode for WriteAt<T> {
  fn create_entry(self: Pin<&mut Self>) -> Entry {
  let slice = self.buffer.as_slice();
  opcode::Write::new(Fd(self.fd), slice.as_ptr(), slice.len() as _)
- .offset(self.offset as _)
+ .offset(self.offset())
  .build()
  }
 }

compio-driver/src/poll/op.rs

@@ -20,7 +20,7 @@ impl<T: IoBufMut> OpCode for ReadAt<T> {
  fd,
  slice.as_mut_ptr() as _,
  slice.len() as _,
- self.offset as _
+ self.offset()
  ))? as _))
  } else {
  Ok(Decision::wait_readable(self.fd))
@@ -33,14 +33,7 @@ impl<T: IoBufMut> OpCode for ReadAt<T> {
  let fd = self.fd;
  let slice = self.buffer.as_uninit_slice();
 
- syscall!(
- break pread(
- fd,
- slice.as_mut_ptr() as _,
- slice.len() as _,
- self.offset as _
- )
- )
+ syscall!(break pread(fd, slice.as_mut_ptr() as _, slice.len() as _, self.offset()))
  }
 }
 
@@ -52,7 +45,7 @@ impl<T: IoBuf> OpCode for WriteAt<T> {
  self.fd,
  slice.as_ptr() as _,
  slice.len() as _,
- self.offset as _
+ self.offset()
  ))? as _))
  } else {
  Ok(Decision::wait_writable(self.fd))
@@ -69,7 +62,7 @@ impl<T: IoBuf> OpCode for WriteAt<T> {
  self.fd,
  slice.as_ptr() as _,
  slice.len() as _,
- self.offset as _
+ self.offset()
  )
  )
  }

compio-driver/src/unix/op.rs

@@ -4,9 +4,27 @@ use compio_buf::{IntoInner, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut};
 use libc::{sockaddr_storage, socklen_t};
 use socket2::SockAddr;
 
-#[cfg(doc)]
-use crate::op::*;
-use crate::RawFd;
+use crate::{op::*, RawFd};
+
+impl<T: IoBufMut> ReadAt<T> {
+ pub(crate) fn offset(&self) -> u64 {
+ if self.offset == usize::MAX {
+ u64::MAX
+ } else {
+ self.offset as _
+ }
+ }
+}
+
+impl<T: IoBuf> WriteAt<T> {
+ pub(crate) fn offset(&self) -> u64 {
+ if self.offset == usize::MAX {
+ u64::MAX
+ } else {
+ self.offset as _
+ }
+ }
+}
 
 /// Accept a connection.
 pub struct Accept {

compio-fs/Cargo.toml

@@ -33,6 +33,7 @@ windows-sys = { version = "0.48", features = [
 # Unix specific dependencies
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
+os_pipe = "1"
 
 # Shared dev dependencies for all platforms
 [dev-dependencies]

compio-fs/src/pipe.rs

@@ -2,7 +2,7 @@
 
 use std::{io, os::unix::fs::FileTypeExt, path::Path};
 
-use compio_driver::{impl_raw_fd, syscall, AsRawFd, FromRawFd};
+use compio_driver::{impl_raw_fd, syscall, AsRawFd, FromRawFd, IntoRawFd};
 #[cfg(feature = "runtime")]
 use {
  compio_buf::{BufResult, IoBuf, IoBufMut},
@@ -14,21 +14,20 @@ use crate::File;
 /// Creates a pair of anonymous pipe.
 ///
 /// ```
-/// use compio_fs::pipe::pipe;
+/// use compio_fs::pipe::anon_pipe;
 ///
 /// # compio_runtime::block_on(async {
-/// let (rx, tx) = pipe().unwrap();
+/// let (rx, tx) = anon_pipe().unwrap();
 ///
 /// tx.write_all("Hello world!").await.unwrap();
 /// let (_, buf) = rx.read_exact(Vec::with_capacity(12)).await.unwrap();
 /// assert_eq!(&buf, b"Hello world!");
 /// # });
 /// ```
-pub fn pipe() -> io::Result<(Receiver, Sender)> {
- let mut fds = [-1, -1];
- syscall!(pipe(fds.as_mut_ptr()))?;
- let receiver = unsafe { Receiver::from_raw_fd(fds[0]) };
- let sender = unsafe { Sender::from_raw_fd(fds[1]) };
+pub fn anon_pipe() -> io::Result<(Receiver, Sender)> {
+ let (receiver, sender) = os_pipe::pipe()?;
+ let receiver = Receiver::from_file(unsafe { File::from_raw_fd(receiver.into_raw_fd()) })?;
+ let sender = Sender::from_file(unsafe { File::from_raw_fd(sender.into_raw_fd()) })?;
  Ok((receiver, sender))
 }
 
@@ -315,9 +314,7 @@ pub struct Sender {
 
 impl Sender {
  pub(crate) fn from_file(file: File) -> io::Result<Sender> {
- if cfg!(not(all(target_os = "linux", feature = "io-uring"))) {
- set_nonblocking(&file)?;
- }
+ set_nonblocking(&file)?;
  Ok(Sender { file })
  }
 
@@ -413,9 +410,7 @@ pub struct Receiver {
 
 impl Receiver {
  pub(crate) fn from_file(file: File) -> io::Result<Receiver> {
- if cfg!(not(all(target_os = "linux", feature = "io-uring"))) {
- set_nonblocking(&file)?;
- }
+ set_nonblocking(&file)?;
  Ok(Receiver { file })
  }
 
@@ -444,16 +439,14 @@ fn is_fifo(file: &File) -> io::Result<bool> {
 }
 
 /// Sets file's flags with O_NONBLOCK by fcntl.
-fn set_nonblocking(file: &File) -> io::Result<()> {
- let fd = file.as_raw_fd();
-
- let current_flags = syscall!(fcntl(fd, libc::F_GETFL))?;
-
- let flags = current_flags | libc::O_NONBLOCK;
-
- if flags != current_flags {
- syscall!(fcntl(fd, libc::F_SETFL, flags))?;
+fn set_nonblocking(file: &impl AsRawFd) -> io::Result<()> {
+ if cfg!(not(all(target_os = "linux", feature = "io-uring"))) {
+ let fd = file.as_raw_fd();
+ let current_flags = syscall!(fcntl(fd, libc::F_GETFL))?;
+ let flags = current_flags | libc::O_NONBLOCK;
+ if flags != current_flags {
+ syscall!(fcntl(fd, libc::F_SETFL, flags))?;
+ }
  }
-
  Ok(())
 }

compio-runtime/Cargo.toml

@@ -44,6 +44,7 @@ windows-sys = { version = "0.48", features = ["Win32_System_IO"] }
 
 # Unix specific dependencies
 [target.'cfg(unix)'.dependencies]
+os_pipe = "1"
 libc = "0.2"
 
 [features]

compio-runtime/src/event/eventfd.rs

@@ -19,7 +19,7 @@ pub struct Event {
 impl Event {
  /// Create [`Event`].
  pub fn new() -> io::Result<Self> {
- let fd = syscall!(eventfd(0, 0))?;
+ let fd = syscall!(eventfd(0, libc::EFD_CLOEXEC))?;
  let fd = unsafe { OwnedFd::from_raw_fd(fd) };
  Ok(Self {
  fd,

compio-runtime/src/event/pipe.rs

@@ -1,6 +1,6 @@
 use std::{
  io,
- os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd},
+ os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd},
 };
 
 use compio_buf::{arrayvec::ArrayVec, BufResult};
@@ -20,14 +20,11 @@ pub struct Event {
 impl Event {
  /// Create [`Event`].
  pub fn new() -> io::Result<Self> {
- let mut fds = [-1, -1];
- syscall!(pipe(fds.as_mut_ptr()))?;
- let receiver = unsafe { OwnedFd::from_raw_fd(fds[0]) };
- let sender = unsafe { OwnedFd::from_raw_fd(fds[1]) };
+ let (receiver, sender) = os_pipe::pipe()?;
+ let receiver = unsafe { OwnedFd::from_raw_fd(receiver.into_raw_fd()) };
+ let sender = unsafe { OwnedFd::from_raw_fd(sender.into_raw_fd()) };
 
- syscall!(fcntl(receiver.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC))?;
  syscall!(fcntl(receiver.as_raw_fd(), libc::F_SETFL, libc::O_NONBLOCK))?;
- syscall!(fcntl(sender.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC))?;
  Ok(Self {
  sender,
  receiver,