send_to and recv_from methods utilizing associated SENDMSG and RE…

examples/udp_echo.rs

@@ -0,0 +1,22 @@
+use std::{
+    io::self,
+    net::{UdpSocket},
+};
+
+fn main() -> io::Result<()> {
+    let ring = rio::new();
+    let socket = UdpSocket::bind("0.0.0.0:34254")?;
+
+    extreme::run(async {
+        let buffer = &mut [0u8; 1024];
+        loop {
+            let (amt, peer) = ring.recv_from(&socket, &buffer).await?;
+            let peer_bstr = &buffer[..amt];
+            println!("Got bytes: {} with bytestring {:?} from peer {:?}",
+                     amt, &peer_bstr, peer);
+            let sent = ring.send_to(&socket, &peer_bstr, &peer).await?;
+            println!("Sent bytes: {} to peer", sent);
+            assert_eq!(amt, sent);
+        }
+    })
+}

src/completion.rs

@@ -8,13 +8,13 @@ use std::{
 };
 
 use super::{
-    io_uring::io_uring_cqe, FromCqe, Measure, Uring, M,
+    CqeData, FromCqeData, Measure, Uring, M,
 };
 
 #[derive(Debug)]
 struct CompletionState {
     done: bool,
-    item: Option<io::Result<io_uring_cqe>>,
+    item: Option<io::Result<CqeData>>,
     waker: Option<Waker>,
 }
 
@@ -38,7 +38,7 @@ impl Default for CompletionState {
 /// happen with `std::mem::forget`, cycles in
 /// `Arc` or `Rc`, and in other ways.
 #[derive(Debug)]
-pub struct Completion<'a, C: FromCqe> {
+pub struct Completion<'a, C: FromCqeData> {
     lifetime: PhantomData<&'a C>,
     mu: Arc<Mutex<CompletionState>>,
     cv: Arc<Condvar>,
@@ -55,7 +55,7 @@ pub struct Filler {
 
 /// Create a new `Filler` and the `Completion`
 /// that will be filled by its completion.
-pub fn pair<'a, C: FromCqe>(
+pub fn pair<'a, C: FromCqeData>(
     uring: &'a Uring,
 ) -> (Completion<'a, C>, Filler) {
     let mu =
@@ -73,19 +73,19 @@ pub fn pair<'a, C: FromCqe>(
     (future, filler)
 }
 
-impl<'a, C: FromCqe> Completion<'a, C> {
+impl<'a, C: FromCqeData> Completion<'a, C> {
     /// Block on the `Completion`'s completion
     /// or dropping of the `Filler`
     pub fn wait(self) -> io::Result<C>
     where
-        C: FromCqe,
+        C: FromCqeData,
     {
         self.wait_inner().unwrap()
     }
 
     fn wait_inner(&self) -> Option<io::Result<C>>
     where
-        C: FromCqe,
+        C: FromCqeData,
     {
         debug_assert_ne!(
             self.sqe_id,
@@ -106,18 +106,18 @@ impl<'a, C: FromCqe> Completion<'a, C> {
         }
 
         inner.item.take().map(|io_result| {
-            io_result.map(FromCqe::from_cqe)
+            io_result.map(FromCqeData::from_cqe_data)
         })
     }
 }
 
-impl<'a, C: FromCqe> Drop for Completion<'a, C> {
+impl<'a, C: FromCqeData> Drop for Completion<'a, C> {
     fn drop(&mut self) {
         self.wait_inner();
     }
 }
 
-impl<'a, C: FromCqe> Future for Completion<'a, C> {
+impl<'a, C: FromCqeData> Future for Completion<'a, C> {
     type Output = io::Result<C>;
 
     fn poll(
@@ -135,7 +135,7 @@ impl<'a, C: FromCqe> Future for Completion<'a, C> {
                     .item
                     .take()
                     .unwrap()
-                    .map(FromCqe::from_cqe),
+                    .map(FromCqeData::from_cqe_data),
             )
         } else {
             if !state.done {
@@ -148,7 +148,7 @@ impl<'a, C: FromCqe> Future for Completion<'a, C> {
 
 impl Filler {
     /// Complete the `Completion`
-    pub fn fill(self, inner: io::Result<io_uring_cqe>) {
+    pub fn fill(self, inner: io::Result<CqeData>) {
         let mut state = self.mu.lock().unwrap();
 
         if let Some(waker) = state.waker.take() {

src/io_uring/cq.rs

@@ -151,7 +151,11 @@ impl Cq {
             let result = if res < 0 {
                 Err(io::Error::from_raw_os_error(res.neg()))
             } else {
-                Ok(*cqe)
+                let address = cq.in_flight.take_address(ticket as usize);
+                Ok(CqeData{
+                    cqe: *cqe,
+                    address: address,
+                })
             };
 
             completion_filler.fill(result);

src/io_uring/in_flight.rs

@@ -1,11 +1,13 @@
 use std::ptr::null_mut;
+use std::net::{SocketAddr, IpAddr, Ipv4Addr};
 
 use super::*;
 
 pub(crate) struct InFlight {
     iovecs: UnsafeCell<Vec<libc::iovec>>,
     msghdrs: UnsafeCell<Vec<libc::msghdr>>,
     fillers: UnsafeCell<Vec<Option<Filler>>>,
+    addresses: UnsafeCell<Vec<Option<SocketAddr>>>,
 }
 
 impl std::fmt::Debug for InFlight {
@@ -33,28 +35,34 @@ impl InFlight {
         ]);
 
         let mut filler_vec = Vec::with_capacity(size);
+        let mut addresses_vec = Vec::with_capacity(size);
         for _ in 0..size {
             filler_vec.push(None);
+            addresses_vec.push(None);
         }
         let fillers = UnsafeCell::new(filler_vec);
+        let addresses = UnsafeCell::new(addresses_vec);
         InFlight {
             iovecs,
             msghdrs,
             fillers,
+            addresses,
         }
     }
 
     pub(crate) fn insert(
         &self,
         ticket: usize,
         iovec: Option<libc::iovec>,
+        address: Option<(*const libc::sockaddr, libc::socklen_t)>,
         msghdr: bool,
         filler: Filler,
     ) -> u64 {
         #[allow(unsafe_code)]
         unsafe {
             let iovec_ptr = self.iovecs.get();
             let msghdr_ptr = self.msghdrs.get();
+            let addresses_ptr = self.addresses.get();
             if let Some(iovec) = iovec {
                 (*iovec_ptr)[ticket] = iovec;
 
@@ -64,6 +72,17 @@ impl InFlight {
                             .as_mut_ptr()
                             .add(ticket);
                     (*msghdr_ptr)[ticket].msg_iovlen = 1;
+                    if let Some((sname, slen)) = address {
+                        (*addresses_ptr)[ticket] = None;
+                        (*msghdr_ptr)[ticket].msg_name = sname as *mut libc::c_void;
+                        (*msghdr_ptr)[ticket].msg_namelen = slen;
+                    } else {
+                        (*addresses_ptr)[ticket] =
+                            Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0));
+                        let (sname, slen) = addr2raw((*addresses_ptr)[ticket].as_ref().unwrap());
+                        (*msghdr_ptr)[ticket].msg_name = sname as *mut libc::c_void;
+                        (*msghdr_ptr)[ticket].msg_namelen = slen;
+                    }
                 }
             }
             (*self.fillers.get())[ticket] = Some(filler);
@@ -90,4 +109,14 @@ impl InFlight {
             (*self.fillers.get())[ticket].take().unwrap()
         }
     }
+
+    pub(crate) fn take_address(
+        &self,
+        ticket: usize,
+    ) -> Option<SocketAddr> {
+        #[allow(unsafe_code)]
+        unsafe {
+            (*self.addresses.get())[ticket].take()
+        }
+    }
 }

src/io_uring/mod.rs

@@ -17,7 +17,7 @@ use std::{
 };
 
 use super::{
-    pair, AsIoVec, AsIoVecMut, Completion, Filler, FromCqe,
+    pair, AsIoVec, AsIoVecMut, Completion, CqeData, Filler, FromCqeData,
     Measure, M,
 };
 
@@ -107,11 +107,39 @@ fn uring_mmap(
     Ok(ptr)
 }
 
-impl FromCqe for TcpStream {
-    fn from_cqe(cqe: io_uring_cqe) -> TcpStream {
+impl FromCqeData for TcpStream {
+    fn from_cqe_data(data: CqeData) -> TcpStream {
         #[allow(unsafe_code)]
         unsafe {
-            TcpStream::from_raw_fd(cqe.res)
+            TcpStream::from_raw_fd(data.cqe.res)
+        }
+    }
+}
+
+impl FromCqeData for (usize, ::std::net::SocketAddr) {
+    fn from_cqe_data(data: CqeData) -> (usize, ::std::net::SocketAddr) {
+        let bytes = usize::try_from(data.cqe.res).unwrap();
+        (bytes, data.address.unwrap())
+    }
+}
+
+fn addr2raw(
+    addr: &std::net::SocketAddr,
+) -> (*const libc::sockaddr, libc::socklen_t) {
+    match *addr {
+        std::net::SocketAddr::V4(ref a) => {
+            let b: *const std::net::SocketAddrV4 = a;
+            (
+                b as *const _,
+                std::mem::size_of_val(a) as libc::socklen_t,
+            )
+        }
+        std::net::SocketAddr::V6(ref a) => {
+            let b: *const std::net::SocketAddrV6 = a;
+            (
+                b as *const _,
+                std::mem::size_of_val(a) as libc::socklen_t,
+            )
         }
     }
 }