Make futures Send.

Author: dvc94ch

src/unix/linux.rs

@@ -95,9 +95,10 @@ impl NetlinkSocket {
                 rtm_flags: libc::RTM_F_NOTIFY,
             },
         };
-        let msg: *const _ = &msg;
-        let address: *const _ = &self.address;
+
         self.fd.write_with(|fd| {
+            let msg: *const _ = &msg;
+            let address: *const _ = &self.address;
             let status = unsafe {
                 libc::sendto(
                     fd.as_raw_fd(),
@@ -121,63 +122,58 @@ impl NetlinkSocket {
     }
 
     async fn recv(&self, buf: &mut Vec<u64>, flags: libc::c_int) -> Status<NetlinkIterator<'_>> {
-        let mut address = MaybeUninit::<libc::sockaddr_nl>::uninit();
-        let mut iovec = libc::iovec {
-            iov_base: buf.as_mut_ptr() as *mut _,
-            iov_len: buf.capacity() * size_of!(u64),
-        };
-
-        let mut msghdr = libc::msghdr {
-            msg_name: address.as_mut_ptr() as _,
-            msg_namelen: size_of!(libc::sockaddr_nl) as u32,
-            msg_iov: &mut iovec,
-            msg_iovlen: 1,
-            msg_control: std::ptr::null_mut(),
-            msg_controllen: 0,
-            msg_flags: 0,
-        };
-
         loop {
             let res = self.fd.read_with(|fd| {
+                let mut address = MaybeUninit::<libc::sockaddr_nl>::uninit();
+                let mut iovec = libc::iovec {
+                    iov_base: buf.as_mut_ptr() as *mut _,
+                    iov_len: buf.capacity() * size_of!(u64),
+                };
+                let mut msghdr = libc::msghdr {
+                    msg_name: address.as_mut_ptr() as _,
+                    msg_namelen: size_of!(libc::sockaddr_nl) as u32,
+                    msg_iov: &mut iovec,
+                    msg_iovlen: 1,
+                    msg_control: std::ptr::null_mut(),
+                    msg_controllen: 0,
+                    msg_flags: 0,
+                };
+
                 let status = unsafe { libc::recvmsg(fd.as_raw_fd(), &mut msghdr, flags) };
                 if status < 0 {
-                    Err(std::io::Error::last_os_error())
-                } else {
-                    Ok(status)
+                    return Ok(Some(match unsafe { *libc::__errno_location() } {
+                        libc::ENOBUFS => Status::Desync,
+                        libc::EAGAIN => return Err(std::io::Error::from_raw_os_error(libc::EAGAIN)),
+                        errno => Status::IO(std::io::Error::from_raw_os_error(errno)),
+                    }))
                 }
-            }).await;
-            let status = match res {
-                Ok(status) => status,
-                Err(err) => {
-                    return if err.raw_os_error() == Some(libc::ENOBUFS) {
-                        Status::Desync
-                    } else {
-                        Status::IO(err)
-                    }
+
+                if msghdr.msg_namelen as usize != size_of!(libc::sockaddr_nl)
+                    || msghdr.msg_flags & (libc::MSG_TRUNC | libc::MSG_CTRUNC) != 0
+                {
+                    return Ok(Some(Status::Desync))
                 }
-            };
-            if msghdr.msg_namelen as usize != size_of!(libc::sockaddr_nl)
-                || msghdr.msg_flags & (libc::MSG_TRUNC | libc::MSG_CTRUNC) != 0
-            {
-                return Status::Desync
-            }
-            // SAFETY: we just checked that the kernel filled in the right size
-            // of address
-            let address = unsafe { address.assume_init() };
-            if address.nl_family != libc::AF_NETLINK as u16 {
-                // wrong address family?
-                return Status::Desync
-            }
-            if address.nl_pid != 0 {
-                // message not from kernel
-                continue
+                // SAFETY: we just checked that the kernel filled in the right size
+                // of address
+                let address = unsafe { address.assume_init() };
+                if address.nl_family != libc::AF_NETLINK as u16 {
+                    // wrong address family?
+                    return Ok(Some(Status::Desync))
+                }
+                if address.nl_pid != 0 {
+                    // message not from kernel
+                    return Ok(None)
+                }
+                Ok(Some(Status::Data(unsafe {
+                    NetlinkIterator::new(core::slice::from_raw_parts(
+                        iovec.iov_base as _,
+                        status as _,
+                    ))
+                })))
+            }).await;
+            if let Ok(Some(status)) = res {
+                return status;
             }
-            break Status::Data(unsafe {
-                NetlinkIterator::new(core::slice::from_raw_parts(
-                    iovec.iov_base as _,
-                    status as _,
-                ))
-            })
         }
     }