Implement UDP send

Cargo.lock

@@ -87,9 +87,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
 [[package]]
 name = "defmt"
-version = "1.1.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1"
+checksum = "e4a0795ac740cb54d7b3775c9c17345f6aaecf884c01733b39d6017738e0e3e1"
 dependencies = [
  "bitflags 1.3.2",
  "defmt-macros",
@@ -97,21 +97,21 @@ dependencies = [
 
 [[package]]
 name = "defmt-macros"
-version = "1.1.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8"
+checksum = "840bb056bd99c495c9188198a258ab787c79f306d864db3e6e8451442d6d8366"
 dependencies = [
  "defmt-parser",
  "proc-macro2",
  "quote",
- "syn 2.0.119",
+ "syn 3.0.9",
 ]
 
 [[package]]
 name = "defmt-parser"
-version = "1.0.0"
+version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e"
+checksum = "71958f69fb2de856c5dd51ac9377e986ca1aa5919f931bf5395a807277f889a0"
 dependencies = [
  "thiserror",
 ]
@@ -158,9 +158,9 @@ checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
 
 [[package]]
 name = "jiff"
-version = "0.2.37"
+version = "0.2.38"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ab1baf72f08796de0260609515130699b890ac25f30e610ad894bc5856cafdb"
+checksum = "b2b005715dcbeb0089a3c0dab99f2ff1cc3b2525323552703d648585d342a383"
 dependencies = [
  "defmt",
  "jiff-core",
@@ -183,9 +183,9 @@ dependencies = [
 
 [[package]]
 name = "jiff-static"
-version = "0.2.37"
+version = "0.2.38"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "378268a1116ad67ae6228701118ac9f491d78fda38a40a1f1a9e1348de6f7212"
+checksum = "2cc9817253cf7c7ee4684451bd327e88d6f3658014e54a29198625590650695c"
 dependencies = [
  "jiff-core",
  "proc-macro2",
@@ -195,9 +195,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.189"
+version = "0.2.191"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a"
 
 [[package]]
 name = "log"
@@ -322,7 +322,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.6",
+ "syn 3.0.9",
 ]
 
 [[package]]
@@ -353,9 +353,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "3.0.6"
+version = "3.0.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
+checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -379,7 +379,7 @@ checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 3.0.6",
+ "syn 3.0.9",
 ]
 
 [[package]]
@@ -390,9 +390,9 @@ checksum = "05442d1d94026b1d3e8ca10750c1461ce60731aee72c698526574e0e91c6e7a2"
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.26"
+version = "1.0.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
+checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563"
 
 [[package]]
 name = "uring_async"

src/lib.rs

@@ -11,7 +11,7 @@ use std::{
 
 use io_uring::{
     IoUring,
-    opcode::{Close, Nop, Socket},
+    opcode::{Close, Nop, RecvMsg, SendMsg, Socket},
     register::Restriction,
     types::{SubmitArgs, Timespec},
 };
@@ -76,6 +76,8 @@ impl Runtime {
                 Restriction::sqe_op(Nop::CODE),
                 Restriction::sqe_op(Socket::CODE),
                 Restriction::sqe_op(Close::CODE),
+                Restriction::sqe_op(SendMsg::CODE),
+                Restriction::sqe_op(RecvMsg::CODE),
             ])
             .expect("failed to register restrictions on ring");
         submitter

src/ops/socket.rs

@@ -1,36 +1,43 @@
 use std::{io, mem};
 
 use io_uring::{
-    opcode::{Close, Socket as SocketOpcode},
+    opcode::{Close, RecvMsg, SendMsg, Socket as SocketOpcode},
     types::Fd,
 };
+use libc::msghdr;
 
 use crate::ring_op::{ring_op_result, submit_and_forget};
 
 pub async fn socket(domain: i32, socket_type: i32, protocol: i32) -> io::Result<Socket> {
     let fd = ring_op_result(|| SocketOpcode::new(domain, socket_type, protocol).build()).await?;
-    Ok(Socket { fd: Some(fd) })
+    Ok(Socket {
+        fd: Fd(fd.try_into().unwrap()),
+    })
 }
 
 pub struct Socket {
-    fd: Option<i32>,
+    fd: Fd,
 }
 
 impl Socket {
-    pub async fn close(mut self) -> io::Result<()> {
+    pub async fn close(self) -> io::Result<()> {
-        let _ = ring_op_result(|| {
+        let res = ring_op_result(|| Close::new(self.fd).build()).await;
-            let fd = mem::take(&mut self.fd).unwrap();
+        mem::forget(self); // Prevent close on drop
-            Close::new(Fd(fd)).build()
+        res?;
-        })
-        .await?;
         Ok(())
     }
+
+    pub async fn send_msg(&self, msghdr: &msghdr, flags: u32) -> io::Result<u32> {
+        ring_op_result(|| SendMsg::new(self.fd, msghdr).flags(flags).build()).await
+    }
+
+    pub async fn recv_msg(&self, msghdr: &mut msghdr, flags: u32) -> io::Result<u32> {
+        ring_op_result(|| RecvMsg::new(self.fd, msghdr).flags(flags).build()).await
+    }
 }
 
 impl Drop for Socket {
     fn drop(&mut self) {
-        if let Some(fd) = self.fd {
+        submit_and_forget(Close::new(self.fd).build());
-            submit_and_forget(Close::new(Fd(fd)).build());
-        }
     }
 }

src/ring_op.rs

@@ -161,11 +161,9 @@ pub async fn ring_op(with_entry: impl FnMut() -> Entry) -> Complete {
     .await
 }
 
-pub async fn ring_op_result(with_entry: impl FnMut() -> Entry) -> io::Result<i32> {
+pub async fn ring_op_result(with_entry: impl FnMut() -> Entry) -> io::Result<u32> {
     let Complete { result, .. } = ring_op(with_entry).await;
-    if result < 0 {
+    result
-        Err(io::Error::from_raw_os_error(-result))
+        .try_into()
-    } else {
+        .map_err(|_| io::Error::from_raw_os_error(-result))
-        Ok(result)
-    }
 }

src/tests.rs

@@ -1,9 +1,10 @@
 use std::{
     io::{Write as _, stdout},
+    ptr,
     time::Duration,
 };
 
-use libc::{AF_INET, SOCK_DGRAM};
+use libc::{AF_INET, SOCK_DGRAM, in_addr, iovec, msghdr, sockaddr_in};
 
 use crate::{
     ops::{no_op::no_op, socket::socket},
@@ -69,3 +70,46 @@ fn socket_lifecycle() {
     }));
     runtime.run();
 }
+
+#[test]
+fn udp() {
+    init();
+    let mut runtime = Runtime::new();
+    runtime.spawn(Box::pin(async move {
+        let socket = socket(AF_INET, SOCK_DGRAM, 0).await.unwrap();
+
+        let mut dest = sockaddr_in {
+            sin_family: AF_INET.try_into().unwrap(),
+            sin_port: 9999u16.to_be(),
+            sin_addr: in_addr {
+                s_addr: u32::from_be_bytes([127, 0, 0, 1]).to_be(),
+            },
+            sin_zero: [0; 8],
+        };
+        let mut buf = vec![];
+        writeln!(&mut buf, "Hello world!").unwrap();
+        let mut iovec = iovec {
+            iov_base: (&mut buf[0] as *mut u8) as *mut libc::c_void,
+            iov_len: buf.len(),
+        };
+        let bytes_sent = socket
+            .send_msg(
+                &msghdr {
+                    msg_name: (&mut dest as *mut sockaddr_in) as *mut libc::c_void,
+                    msg_namelen: size_of::<sockaddr_in>().try_into().unwrap(),
+                    msg_iov: &mut iovec,
+                    msg_iovlen: 1,
+                    msg_control: ptr::null_mut(),
+                    msg_controllen: 0,
+                    msg_flags: 0,
+                },
+                0,
+            )
+            .await
+            .unwrap();
+        println!("Sent {} bytes of {}", bytes_sent, buf.len());
+
+        socket.close().await.unwrap();
+    }));
+    runtime.run();
+}