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

@@ -30,19 +30,24 @@ windows-sys = { version = "0.48", features = [
  "Win32_System_SystemServices",
 ] }
 
-# Windows specific dev dependencies
-[target.'cfg(target_os = "windows")'.dev-dependencies]
-windows-sys = { version = "0.48", features = ["Win32_Security_Authorization"] }
-
 # Unix specific dependencies
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
+os_pipe = "1"
 
 # Shared dev dependencies for all platforms
 [dev-dependencies]
 compio-runtime = { workspace = true, features = ["time"] }
 futures-util = "0.3"
 
+# Windows specific dev dependencies
+[target.'cfg(target_os = "windows")'.dev-dependencies]
+windows-sys = { version = "0.48", features = ["Win32_Security_Authorization"] }
+
+# Unix specific dev dependencies
+[target.'cfg(unix)'.dev-dependencies]
+nix = { version = "0.27", features = ["fs"] }
+
 [features]
 runtime = ["dep:compio-buf", "dep:compio-runtime"]
 

compio-fs/src/lib.rs

@@ -11,3 +11,6 @@ pub use open_options::*;
 
 #[cfg(target_os = "windows")]
 pub mod named_pipe;
+
+#[cfg(unix)]
+pub mod pipe;