feat: QUIC

Cargo.toml

@@ -13,6 +13,7 @@ members = [
     "compio-tls",
     "compio-log",
     "compio-process",
+    "compio-quic",
 ]
 resolver = "2"
 
@@ -36,6 +37,7 @@ compio-dispatcher = { path = "./compio-dispatcher", version = "0.3.0" }
 compio-log = { path = "./compio-log", version = "0.1.0" }
 compio-tls = { path = "./compio-tls", version = "0.2.0", default-features = false }
 compio-process = { path = "./compio-process", version = "0.1.0" }
+compio-quic = { path = "./compio-quic", version = "0.1.0" }
 
 flume = "0.11.0"
 cfg-if = "1.0.0"
@@ -49,6 +51,7 @@ nix = "0.29.0"
 once_cell = "1.18.0"
 os_pipe = "1.1.4"
 paste = "1.0.14"
+rustls = { version = "0.23.1", default-features = false }
 slab = "0.4.9"
 socket2 = "0.5.6"
 tempfile = "3.8.1"

compio-driver/src/iocp/op.rs

@@ -1,13 +1,7 @@
 #[cfg(feature = "once_cell_try")]
 use std::sync::OnceLock;
 use std::{
-    io,
-    marker::PhantomPinned,
-    net::Shutdown,
-    os::windows::io::AsRawSocket,
-    pin::Pin,
-    ptr::{null, null_mut},
-    task::Poll,
+    io, marker::PhantomPinned, net::Shutdown, os::windows::io::AsRawSocket, pin::Pin, ptr::{null, null_mut}, task::Poll
 };
 
 use aligned_array::{Aligned, A8};
@@ -781,12 +775,12 @@ static WSA_RECVMSG: OnceLock<LPFN_WSARECVMSG> = OnceLock::new();
 
 /// Receive data and source address with ancillary data into vectored buffer.
 pub struct RecvMsg<T: IoVectoredBufMut, C: IoBufMut, S> {
+    msg: WSAMSG,
     addr: SOCKADDR_STORAGE,
-    addr_len: socklen_t,
     fd: SharedFd<S>,
     buffer: T,
     control: C,
-    control_len: u32,
+    slices: Vec<IoSliceMut>,
     _p: PhantomPinned,
 }
 
@@ -802,12 +796,12 @@ impl<T: IoVectoredBufMut, C: IoBufMut, S> RecvMsg<T, C, S> {
             "misaligned control message buffer"
         );
         Self {
+            msg: unsafe { std::mem::zeroed() },
             addr: unsafe { std::mem::zeroed() },
-            addr_len: std::mem::size_of::<SOCKADDR_STORAGE>() as _,
             fd,
             buffer,
             control,
-            control_len: 0,
+            slices: vec![],
             _p: PhantomPinned,
         }
     }
@@ -820,8 +814,8 @@ impl<T: IoVectoredBufMut, C: IoBufMut, S> IntoInner for RecvMsg<T, C, S> {
         (
             (self.buffer, self.control),
             self.addr,
-            self.addr_len,
-            self.control_len as _,
+            self.msg.namelen,
+            self.msg.Control.len as _,
         )
     }
 }
@@ -835,26 +829,22 @@ impl<T: IoVectoredBufMut, C: IoBufMut, S: AsRawFd> OpCode for RecvMsg<T, C, S> {
             })?;
 
         let this = self.get_unchecked_mut();
-        let mut slices = this.buffer.io_slices_mut();
-        let mut msg = WSAMSG {
-            name: &mut this.addr as *mut _ as _,
-            namelen: this.addr_len,
-            lpBuffers: slices.as_mut_ptr() as _,
-            dwBufferCount: slices.len() as _,
-            Control: std::mem::transmute::<IoSliceMut, WSABUF>(this.control.as_io_slice_mut()),
-            dwFlags: 0,
-        };
-        this.control_len = 0;
+
+        this.slices = this.buffer.io_slices_mut();
+        this.msg.name = &mut this.addr as *mut _ as _;
+        this.msg.namelen = std::mem::size_of::<SOCKADDR_STORAGE>() as _;
+        this.msg.lpBuffers = this.slices.as_mut_ptr() as _;
+        this.msg.dwBufferCount = this.slices.len() as _;
+        this.msg.Control = std::mem::transmute::<IoSliceMut, WSABUF>(this.control.as_io_slice_mut());
 
         let mut received = 0;
         let res = recvmsg_fn(
             this.fd.as_raw_fd() as _,
-            &mut msg,
+            &mut this.msg,
             &mut received,
             optr,
             None,
         );
-        this.control_len = msg.Control.len;
         winsock_result(res, received)
     }
 
@@ -866,10 +856,12 @@ impl<T: IoVectoredBufMut, C: IoBufMut, S: AsRawFd> OpCode for RecvMsg<T, C, S> {
 /// Send data to specified address accompanied by ancillary data from vectored
 /// buffer.
 pub struct SendMsg<T: IoVectoredBuf, C: IoBuf, S> {
+    msg: WSAMSG,
     fd: SharedFd<S>,
     buffer: T,
     control: C,
     addr: SockAddr,
+    pub(crate) slices: Vec<IoSlice>,
     _p: PhantomPinned,
 }
 
@@ -885,10 +877,12 @@ impl<T: IoVectoredBuf, C: IoBuf, S> SendMsg<T, C, S> {
             "misaligned control message buffer"
         );
         Self {
+            msg: unsafe { std::mem::zeroed() },
             fd,
             buffer,
             control,
             addr,
+            slices: vec![],
             _p: PhantomPinned,
         }
     }
@@ -906,18 +900,15 @@ impl<T: IoVectoredBuf, C: IoBuf, S: AsRawFd> OpCode for SendMsg<T, C, S> {
     unsafe fn operate(self: Pin<&mut Self>, optr: *mut OVERLAPPED) -> Poll<io::Result<usize>> {
         let this = self.get_unchecked_mut();
 
-        let slices = this.buffer.io_slices();
-        let msg = WSAMSG {
-            name: this.addr.as_ptr() as _,
-            namelen: this.addr.len(),
-            lpBuffers: slices.as_ptr() as _,
-            dwBufferCount: slices.len() as _,
-            Control: std::mem::transmute::<IoSlice, WSABUF>(this.control.as_io_slice()),
-            dwFlags: 0,
-        };
+        this.slices = this.buffer.io_slices();
+        this.msg.name = this.addr.as_ptr() as _;
+        this.msg.namelen = this.addr.len();
+        this.msg.lpBuffers = this.slices.as_ptr() as _;
+        this.msg.dwBufferCount = this.slices.len() as _;
+        this.msg.Control = std::mem::transmute::<IoSlice, WSABUF>(this.control.as_io_slice());
 
         let mut sent = 0;
-        let res = WSASendMsg(this.fd.as_raw_fd() as _, &msg, 0, &mut sent, optr, None);
+        let res = WSASendMsg(this.fd.as_raw_fd() as _, &this.msg, 0, &mut sent, optr, None);
         winsock_result(res, sent)
     }
 

compio-net/src/socket.rs

@@ -1,4 +1,8 @@
-use std::{future::Future, io, mem::ManuallyDrop};
+use std::{
+    future::Future,
+    io,
+    mem::{ManuallyDrop, MaybeUninit},
+};
 
 use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut};
 #[cfg(unix)]
@@ -319,6 +323,45 @@ impl Socket {
         compio_runtime::submit(op).await.into_inner()
     }
 
+    #[cfg(unix)]
+    pub fn get_socket_option<T>(&self, level: i32, name: i32) -> io::Result<T> {
+        let mut value: MaybeUninit<T> = MaybeUninit::uninit();
+        let mut len = size_of::<T>() as libc::socklen_t;
+        syscall!(libc::getsockopt(
+            self.socket.as_raw_fd(),
+            level,
+            name,
+            value.as_mut_ptr() as _,
+            &mut len
+        ))
+        .map(|_| {
+            debug_assert_eq!(len as usize, size_of::<T>());
+            // SAFETY: The value is initialized by `getsockopt`.
+            unsafe { value.assume_init() }
+        })
+    }
+
+    #[cfg(windows)]
+    pub fn get_socket_option<T>(&self, level: i32, name: i32) -> io::Result<T> {
+        let mut value: MaybeUninit<T> = MaybeUninit::uninit();
+        let mut len = size_of::<T>() as i32;
+        syscall!(
+            SOCKET,
+            windows_sys::Win32::Networking::WinSock::getsockopt(
+                self.socket.as_raw_fd() as _,
+                level,
+                name,
+                value.as_mut_ptr() as _,
+                &mut len
+            )
+        )
+        .map(|_| {
+            debug_assert_eq!(len as usize, size_of::<T>());
+            // SAFETY: The value is initialized by `getsockopt`.
+            unsafe { value.assume_init() }
+        })
+    }
+
     #[cfg(unix)]
     pub fn set_socket_option<T>(&self, level: i32, name: i32, value: &T) -> io::Result<()> {
         syscall!(libc::setsockopt(

compio-net/src/udp.rs

@@ -316,6 +316,11 @@ impl UdpSocket {
         .await
     }
 
+    /// Gets a socket option.
+    pub fn get_socket_option<T>(&self, level: i32, name: i32) -> io::Result<T> {
+        self.inner.get_socket_option(level, name)
+    }
+
     /// Sets a socket option.
     pub fn set_socket_option<T>(&self, level: i32, name: i32, value: &T) -> io::Result<()> {
         self.inner.set_socket_option(level, name, value)

compio-quic/Cargo.toml

@@ -0,0 +1,69 @@
+[package]
+name = "compio-quic"
+version = "0.1.0"
+description = "QUIC for compio"
+categories = ["asynchronous", "network-programming"]
+keywords = ["async", "net", "quic"]
+edition = { workspace = true }
+authors = { workspace = true }
+readme = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[dependencies]
+# Workspace dependencies
+compio-io = { workspace = true }
+compio-buf = { workspace = true }
+compio-log = { workspace = true }
+compio-net = { workspace = true }
+compio-runtime = { workspace = true, features = ["time"] }
+
+quinn-proto = "0.11.3"
+rustls = { workspace = true, features = ["ring"] }
+rustls-platform-verifier = { version = "0.3.3", optional = true }
+rustls-native-certs = { version = "0.7.1", optional = true }
+webpki-roots = { version = "0.26.3", optional = true }
+
+# Utils
+bytes = "1.7.1"
+event-listener = "5.3.1"
+flume = { workspace = true }
+futures-util = { workspace = true }
+thiserror = "1.0.63"
+
+# Windows specific dependencies
+[target.'cfg(windows)'.dependencies]
+windows-sys = { workspace = true, features = ["Win32_Networking_WinSock"] }
+
+[target.'cfg(unix)'.dependencies]
+libc = { workspace = true }
+
+[dev-dependencies]
+compio-dispatcher = { workspace = true }
+compio-driver = { workspace = true }
+compio-macros = { workspace = true }
+compio-runtime = { workspace = true, features = ["criterion"] }
+
+rand = "0.8.5"
+rcgen = "0.13.1"
+socket2 = { workspace = true, features = ["all"] }
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+
+criterion = { workspace = true, features = ["async_tokio"] }
+quinn = "0.11.3"
+tokio = { workspace = true, features = ["rt", "macros"] }
+
+[features]
+default = []
+io-compat = ["futures-util/io"]
+platform-verifier = ["dep:rustls-platform-verifier"]
+native-certs = ["dep:rustls-native-certs"]
+webpki-roots = ["dep:webpki-roots"]
+
+[[bench]]
+name = "quic"
+harness = false