src/io/seq_packet.rs

   1 use std::io::{self, IoSlice, IoSliceMut};
   2 use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd};
   3 use std::ptr;
   4
   5 use anyhow::Error;
   6 use nix::sys::socket::{self, AddressFamily, SockFlag, SockType, SockaddrLike};
   7 use tokio::io::unix::AsyncFd;
   8
   9 use crate::tools::AssertSendSync;
  10
  11 fn seq_packet_socket(flags: SockFlag) -> nix::Result<OwnedFd> {
  12     let fd = socket::socket(
  13         AddressFamily::Unix,
  14         SockType::SeqPacket,
  15         flags | SockFlag::SOCK_CLOEXEC,
  16         None,
  17     )?;
  18     Ok(unsafe { OwnedFd::from_raw_fd(fd) })
  19 }
  20
  21 pub struct SeqPacketListener {
  22     fd: AsyncFd<OwnedFd>,
  23 }
  24
  25 impl AsRawFd for SeqPacketListener {
  26     #[inline]
  27     fn as_raw_fd(&self) -> RawFd {
  28         self.fd.as_raw_fd()
  29     }
  30 }
  31
  32 impl SeqPacketListener {
  33     pub fn bind(address: &dyn SockaddrLike) -> Result<Self, Error> {
  34         let fd = seq_packet_socket(SockFlag::empty())?;
  35         socket::bind(fd.as_raw_fd(), address)?;
  36         socket::listen(fd.as_raw_fd(), 16)?;
  37
  38         let fd = AsyncFd::new(fd)?;
  39
  40         Ok(Self { fd })
  41     }
  42
  43     pub async fn accept(&mut self) -> io::Result<SeqPacketSocket> {
  44         let fd = super::wrap_read(&self.fd, |fd| {
  45             c_result!(unsafe {
  46                 libc::accept4(
  47                     fd,
  48                     ptr::null_mut(),
  49                     ptr::null_mut(),
  50                     libc::SOCK_CLOEXEC | libc::SOCK_NONBLOCK,
  51                 )
  52             })
  53         })
  54         .await?;
  55
  56         let fd = unsafe { OwnedFd::from_raw_fd(fd as RawFd) };
  57         SeqPacketSocket::new(fd)
  58     }
  59 }
  60
  61 pub struct SeqPacketSocket {
  62     fd: AsyncFd<OwnedFd>,
  63 }
  64
  65 impl AsRawFd for SeqPacketSocket {
  66     #[inline]
  67     fn as_raw_fd(&self) -> RawFd {
  68         self.fd.as_raw_fd()
  69     }
  70 }
  71
  72 impl SeqPacketSocket {
  73     pub fn new(fd: OwnedFd) -> io::Result<Self> {
  74         Ok(Self {
  75             fd: AsyncFd::new(fd)?,
  76         })
  77     }
  78
  79     async fn sendmsg(&self, msg: &AssertSendSync<libc::msghdr>) -> io::Result<usize> {
  80         let rc = super::wrap_write(&self.fd, |fd| {
  81             c_result!(unsafe { libc::sendmsg(fd, &msg.0 as *const libc::msghdr, 0) })
  82         })
  83         .await?;
  84         Ok(rc as usize)
  85     }
  86
  87     pub async fn sendmsg_vectored(&self, iov: &[IoSlice<'_>]) -> io::Result<usize> {
  88         let msg = AssertSendSync(libc::msghdr {
  89             msg_name: ptr::null_mut(),
  90             msg_namelen: 0,
  91             msg_iov: iov.as_ptr() as _,
  92             msg_iovlen: iov.len(),
  93             msg_control: ptr::null_mut(),
  94             msg_controllen: 0,
  95             msg_flags: 0,
  96         });
  97
  98         self.sendmsg(&msg).await
  99     }
 100
 101     async fn recvmsg(&self, msg: &mut AssertSendSync<libc::msghdr>) -> io::Result<usize> {
 102         let rc = super::wrap_read(&self.fd, move |fd| {
 103             c_result!(unsafe { libc::recvmsg(fd, &mut msg.0 as *mut libc::msghdr, 0) })
 104         })
 105         .await?;
 106         Ok(rc as usize)
 107     }
 108
 109     // clippy is wrong about this one
 110     #[allow(clippy::needless_lifetimes)]
 111     pub async fn recvmsg_vectored(
 112         &self,
 113         iov: &mut [IoSliceMut<'_>],
 114         cmsg_buf: &mut [u8],
 115     ) -> io::Result<(usize, usize)> {
 116         let mut msg = AssertSendSync(libc::msghdr {
 117             msg_name: ptr::null_mut(),
 118             msg_namelen: 0,
 119             msg_iov: iov.as_mut_ptr() as _,
 120             msg_iovlen: iov.len(),
 121             msg_control: cmsg_buf.as_mut_ptr() as *mut std::ffi::c_void,
 122             msg_controllen: cmsg_buf.len(),
 123             msg_flags: libc::MSG_CMSG_CLOEXEC,
 124         });
 125
 126         let data_size = self.recvmsg(&mut msg).await?;
 127         Ok((data_size, msg.0.msg_controllen))
 128     }
 129
 130     #[inline]
 131     pub fn shutdown(&self, how: socket::Shutdown) -> nix::Result<()> {
 132         socket::shutdown(self.as_raw_fd(), how)
 133     }
 134 }