Skip to main content

tokio/net/unix/datagram/
socket.rs

1use crate::io::{Interest, PollEvented, ReadBuf, Ready};
2use crate::net::unix::SocketAddr;
3use crate::util::check_socket_for_blocking;
4
5use std::fmt;
6use std::io;
7use std::net::Shutdown;
8use std::os::unix::io::{AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, RawFd};
9use std::os::unix::net;
10use std::path::Path;
11use std::task::{ready, Context, Poll};
12
13cfg_io_util! {
14    use bytes::BufMut;
15}
16
17cfg_net_unix! {
18    /// An I/O object representing a Unix datagram socket.
19    ///
20    /// A socket can be either named (associated with a filesystem path) or
21    /// unnamed.
22    ///
23    /// This type does not provide a `split` method, because this functionality
24    /// can be achieved by wrapping the socket in an [`Arc`]. Note that you do
25    /// not need a `Mutex` to share the `UnixDatagram` — an `Arc<UnixDatagram>`
26    /// is enough. This is because all of the methods take `&self` instead of
27    /// `&mut self`.
28    ///
29    /// **Note:** named sockets are persisted even after the object is dropped
30    /// and the program has exited, and cannot be reconnected. It is advised
31    /// that you either check for and unlink the existing socket if it exists,
32    /// or use a temporary file that is guaranteed to not already exist.
33    ///
34    /// [`Arc`]: std::sync::Arc
35    ///
36    /// # Examples
37    /// Using named sockets, associated with a filesystem path:
38    /// ```
39    /// # use std::error::Error;
40    /// # #[tokio::main]
41    /// # async fn main() -> Result<(), Box<dyn Error>> {
42    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
43    /// use tokio::net::UnixDatagram;
44    /// use tempfile::tempdir;
45    ///
46    /// // We use a temporary directory so that the socket
47    /// // files left by the bound sockets will get cleaned up.
48    /// let tmp = tempdir()?;
49    ///
50    /// // Bind each socket to a filesystem path
51    /// let tx_path = tmp.path().join("tx");
52    /// let tx = UnixDatagram::bind(&tx_path)?;
53    /// let rx_path = tmp.path().join("rx");
54    /// let rx = UnixDatagram::bind(&rx_path)?;
55    ///
56    /// let bytes = b"hello world";
57    /// tx.send_to(bytes, &rx_path).await?;
58    ///
59    /// let mut buf = vec![0u8; 24];
60    /// let (size, addr) = rx.recv_from(&mut buf).await?;
61    ///
62    /// let dgram = &buf[..size];
63    /// assert_eq!(dgram, bytes);
64    /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
65    ///
66    /// # Ok(())
67    /// # }
68    /// ```
69    ///
70    /// Using unnamed sockets, created as a pair
71    /// ```
72    /// # use std::error::Error;
73    /// # #[tokio::main]
74    /// # async fn main() -> Result<(), Box<dyn Error>> {
75    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
76    /// use tokio::net::UnixDatagram;
77    ///
78    /// // Create the pair of sockets
79    /// let (sock1, sock2) = UnixDatagram::pair()?;
80    ///
81    /// // Since the sockets are paired, the paired send/recv
82    /// // functions can be used
83    /// let bytes = b"hello world";
84    /// sock1.send(bytes).await?;
85    ///
86    /// let mut buff = vec![0u8; 24];
87    /// let size = sock2.recv(&mut buff).await?;
88    ///
89    /// let dgram = &buff[..size];
90    /// assert_eq!(dgram, bytes);
91    ///
92    /// # Ok(())
93    /// # }
94    /// ```
95    #[cfg_attr(docsrs, doc(alias = "uds"))]
96    pub struct UnixDatagram {
97        io: PollEvented<mio::net::UnixDatagram>,
98    }
99}
100
101impl UnixDatagram {
102    pub(crate) fn from_mio(sys: mio::net::UnixDatagram) -> io::Result<UnixDatagram> {
103        let datagram = UnixDatagram::new(sys)?;
104
105        if let Some(e) = datagram.io.take_error()? {
106            return Err(e);
107        }
108
109        Ok(datagram)
110    }
111
112    /// Waits for any of the requested ready states.
113    ///
114    /// This function is usually paired with `try_recv()` or `try_send()`. It
115    /// can be used to concurrently `recv` / `send` to the same socket on a single
116    /// task without splitting the socket.
117    ///
118    /// The function may complete without the socket being ready. This is a
119    /// false-positive and attempting an operation will return with
120    /// `io::ErrorKind::WouldBlock`. The function can also return with an empty
121    /// [`Ready`] set, so you should always check the returned value and possibly
122    /// wait again if the requested states are not set.
123    ///
124    /// # Cancel safety
125    ///
126    /// This method is cancel safe. Once a readiness event occurs, the method
127    /// will continue to return immediately until the readiness event is
128    /// consumed by an attempt to read or write that fails with `WouldBlock` or
129    /// `Poll::Pending`.
130    ///
131    /// # Examples
132    ///
133    /// Concurrently receive from and send to the socket on the same task
134    /// without splitting.
135    ///
136    /// ```no_run
137    /// use tokio::io::Interest;
138    /// use tokio::net::UnixDatagram;
139    /// use std::io;
140    ///
141    /// #[tokio::main]
142    /// async fn main() -> io::Result<()> {
143    ///     let dir = tempfile::tempdir().unwrap();
144    ///     let client_path = dir.path().join("client.sock");
145    ///     let server_path = dir.path().join("server.sock");
146    ///     let socket = UnixDatagram::bind(&client_path)?;
147    ///     socket.connect(&server_path)?;
148    ///
149    ///     loop {
150    ///         let ready = socket.ready(Interest::READABLE | Interest::WRITABLE).await?;
151    ///
152    ///         if ready.is_readable() {
153    ///             let mut data = [0; 1024];
154    ///             match socket.try_recv(&mut data[..]) {
155    ///                 Ok(n) => {
156    ///                     println!("received {:?}", &data[..n]);
157    ///                 }
158    ///                 // False-positive, continue
159    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
160    ///                 Err(e) => {
161    ///                     return Err(e);
162    ///                 }
163    ///             }
164    ///         }
165    ///
166    ///         if ready.is_writable() {
167    ///             // Write some data
168    ///             match socket.try_send(b"hello world") {
169    ///                 Ok(n) => {
170    ///                     println!("sent {} bytes", n);
171    ///                 }
172    ///                 // False-positive, continue
173    ///                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
174    ///                 Err(e) => {
175    ///                     return Err(e);
176    ///                 }
177    ///             }
178    ///         }
179    ///     }
180    /// }
181    /// ```
182    pub async fn ready(&self, interest: Interest) -> io::Result<Ready> {
183        let event = self.io.registration().readiness(interest).await?;
184        Ok(event.ready)
185    }
186
187    /// Waits for the socket to become writable.
188    ///
189    /// This function is equivalent to `ready(Interest::WRITABLE)` and is
190    /// usually paired with `try_send()` or `try_send_to()`.
191    ///
192    /// The function may complete without the socket being writable. This is a
193    /// false-positive and attempting a `try_send()` will return with
194    /// `io::ErrorKind::WouldBlock`.
195    ///
196    /// # Cancel safety
197    ///
198    /// This method is cancel safe. Once a readiness event occurs, the method
199    /// will continue to return immediately until the readiness event is
200    /// consumed by an attempt to write that fails with `WouldBlock` or
201    /// `Poll::Pending`.
202    ///
203    /// # Examples
204    ///
205    /// ```no_run
206    /// use tokio::net::UnixDatagram;
207    /// use std::io;
208    ///
209    /// #[tokio::main]
210    /// async fn main() -> io::Result<()> {
211    ///     let dir = tempfile::tempdir().unwrap();
212    ///     let client_path = dir.path().join("client.sock");
213    ///     let server_path = dir.path().join("server.sock");
214    ///     let socket = UnixDatagram::bind(&client_path)?;
215    ///     socket.connect(&server_path)?;
216    ///
217    ///     loop {
218    ///         // Wait for the socket to be writable
219    ///         socket.writable().await?;
220    ///
221    ///         // Try to send data, this may still fail with `WouldBlock`
222    ///         // if the readiness event is a false positive.
223    ///         match socket.try_send(b"hello world") {
224    ///             Ok(n) => {
225    ///                 break;
226    ///             }
227    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
228    ///                 continue;
229    ///             }
230    ///             Err(e) => {
231    ///                 return Err(e);
232    ///             }
233    ///         }
234    ///     }
235    ///
236    ///     Ok(())
237    /// }
238    /// ```
239    pub async fn writable(&self) -> io::Result<()> {
240        self.ready(Interest::WRITABLE).await?;
241        Ok(())
242    }
243
244    /// Polls for write/send readiness.
245    ///
246    /// If the socket is not currently ready for sending, this method will
247    /// store a clone of the `Waker` from the provided `Context`. When the socket
248    /// becomes ready for sending, `Waker::wake` will be called on the
249    /// waker.
250    ///
251    /// Note that on multiple calls to `poll_send_ready` or `poll_send`, only
252    /// the `Waker` from the `Context` passed to the most recent call is
253    /// scheduled to receive a wakeup. (However, `poll_recv_ready` retains a
254    /// second, independent waker.)
255    ///
256    /// This function is intended for cases where creating and pinning a future
257    /// via [`writable`] is not feasible. Where possible, using [`writable`] is
258    /// preferred, as this supports polling from multiple tasks at once.
259    ///
260    /// # Return value
261    ///
262    /// The function returns:
263    ///
264    /// * `Poll::Pending` if the socket is not ready for writing.
265    /// * `Poll::Ready(Ok(()))` if the socket is ready for writing.
266    /// * `Poll::Ready(Err(e))` if an error is encountered.
267    ///
268    /// # Errors
269    ///
270    /// This function may encounter any standard I/O error except `WouldBlock`.
271    ///
272    /// [`writable`]: method@Self::writable
273    pub fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
274        self.io.registration().poll_write_ready(cx).map_ok(|_| ())
275    }
276
277    /// Waits for the socket to become readable.
278    ///
279    /// This function is equivalent to `ready(Interest::READABLE)` and is usually
280    /// paired with `try_recv()`.
281    ///
282    /// The function may complete without the socket being readable. This is a
283    /// false-positive and attempting a `try_recv()` will return with
284    /// `io::ErrorKind::WouldBlock`.
285    ///
286    /// # Cancel safety
287    ///
288    /// This method is cancel safe. Once a readiness event occurs, the method
289    /// will continue to return immediately until the readiness event is
290    /// consumed by an attempt to read that fails with `WouldBlock` or
291    /// `Poll::Pending`.
292    ///
293    /// # Examples
294    ///
295    /// ```no_run
296    /// use tokio::net::UnixDatagram;
297    /// use std::io;
298    ///
299    /// #[tokio::main]
300    /// async fn main() -> io::Result<()> {
301    ///     // Connect to a peer
302    ///     let dir = tempfile::tempdir().unwrap();
303    ///     let client_path = dir.path().join("client.sock");
304    ///     let server_path = dir.path().join("server.sock");
305    ///     let socket = UnixDatagram::bind(&client_path)?;
306    ///     socket.connect(&server_path)?;
307    ///
308    ///     loop {
309    ///         // Wait for the socket to be readable
310    ///         socket.readable().await?;
311    ///
312    ///         // The buffer is **not** included in the async task and will
313    ///         // only exist on the stack.
314    ///         let mut buf = [0; 1024];
315    ///
316    ///         // Try to recv data, this may still fail with `WouldBlock`
317    ///         // if the readiness event is a false positive.
318    ///         match socket.try_recv(&mut buf) {
319    ///             Ok(n) => {
320    ///                 println!("GOT {:?}", &buf[..n]);
321    ///                 break;
322    ///             }
323    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
324    ///                 continue;
325    ///             }
326    ///             Err(e) => {
327    ///                 return Err(e);
328    ///             }
329    ///         }
330    ///     }
331    ///
332    ///     Ok(())
333    /// }
334    /// ```
335    pub async fn readable(&self) -> io::Result<()> {
336        self.ready(Interest::READABLE).await?;
337        Ok(())
338    }
339
340    /// Polls for read/receive readiness.
341    ///
342    /// If the socket is not currently ready for receiving, this method will
343    /// store a clone of the `Waker` from the provided `Context`. When the
344    /// socket becomes ready for reading, `Waker::wake` will be called on the
345    /// waker.
346    ///
347    /// Note that on multiple calls to `poll_recv_ready`, `poll_recv` or
348    /// `poll_peek`, only the `Waker` from the `Context` passed to the most
349    /// recent call is scheduled to receive a wakeup. (However,
350    /// `poll_send_ready` retains a second, independent waker.)
351    ///
352    /// This function is intended for cases where creating and pinning a future
353    /// via [`readable`] is not feasible. Where possible, using [`readable`] is
354    /// preferred, as this supports polling from multiple tasks at once.
355    ///
356    /// # Return value
357    ///
358    /// The function returns:
359    ///
360    /// * `Poll::Pending` if the socket is not ready for reading.
361    /// * `Poll::Ready(Ok(()))` if the socket is ready for reading.
362    /// * `Poll::Ready(Err(e))` if an error is encountered.
363    ///
364    /// # Errors
365    ///
366    /// This function may encounter any standard I/O error except `WouldBlock`.
367    ///
368    /// [`readable`]: method@Self::readable
369    pub fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
370        self.io.registration().poll_read_ready(cx).map_ok(|_| ())
371    }
372
373    /// Creates a new `UnixDatagram` bound to the specified path.
374    ///
375    /// # Examples
376    /// ```
377    /// # use std::error::Error;
378    /// # #[tokio::main]
379    /// # async fn main() -> Result<(), Box<dyn Error>> {
380    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
381    /// use tokio::net::UnixDatagram;
382    /// use tempfile::tempdir;
383    ///
384    /// // We use a temporary directory so that the socket
385    /// // files left by the bound sockets will get cleaned up.
386    /// let tmp = tempdir()?;
387    ///
388    /// // Bind the socket to a filesystem path
389    /// let socket_path = tmp.path().join("socket");
390    /// let socket = UnixDatagram::bind(&socket_path)?;
391    ///
392    /// # Ok(())
393    /// # }
394    /// ```
395    pub fn bind<P>(path: P) -> io::Result<UnixDatagram>
396    where
397        P: AsRef<Path>,
398    {
399        let socket = mio::net::UnixDatagram::bind(path)?;
400        UnixDatagram::new(socket)
401    }
402
403    /// Creates an unnamed pair of connected sockets.
404    ///
405    /// This function will create a pair of interconnected Unix sockets for
406    /// communicating back and forth between one another.
407    ///
408    /// # Examples
409    /// ```
410    /// # use std::error::Error;
411    /// # #[tokio::main]
412    /// # async fn main() -> Result<(), Box<dyn Error>> {
413    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
414    /// use tokio::net::UnixDatagram;
415    ///
416    /// // Create the pair of sockets
417    /// let (sock1, sock2) = UnixDatagram::pair()?;
418    ///
419    /// // Since the sockets are paired, the paired send/recv
420    /// // functions can be used
421    /// let bytes = b"hail eris";
422    /// sock1.send(bytes).await?;
423    ///
424    /// let mut buff = vec![0u8; 24];
425    /// let size = sock2.recv(&mut buff).await?;
426    ///
427    /// let dgram = &buff[..size];
428    /// assert_eq!(dgram, bytes);
429    ///
430    /// # Ok(())
431    /// # }
432    /// ```
433    pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
434        let (a, b) = mio::net::UnixDatagram::pair()?;
435        let a = UnixDatagram::new(a)?;
436        let b = UnixDatagram::new(b)?;
437        // A fresh pair has empty buffers: writable now, readable only once
438        // the peer sends.
439        a.io.registration().assume_ready(Ready::WRITABLE);
440        b.io.registration().assume_ready(Ready::WRITABLE);
441
442        Ok((a, b))
443    }
444
445    /// Creates new [`UnixDatagram`] from a [`std::os::unix::net::UnixDatagram`].
446    ///
447    /// This function is intended to be used to wrap a `UnixDatagram` from the
448    /// standard library in the Tokio equivalent.
449    ///
450    /// # Notes
451    ///
452    /// The caller is responsible for ensuring that the socket is in
453    /// non-blocking mode. Otherwise all I/O operations on the socket
454    /// will block the thread, which will cause unexpected behavior.
455    /// Non-blocking mode can be set using [`set_nonblocking`].
456    ///
457    /// Passing a listener in blocking mode is always erroneous,
458    /// and the behavior in that case may change in the future.
459    /// For example, it could panic.
460    ///
461    /// [`set_nonblocking`]: std::os::unix::net::UnixDatagram::set_nonblocking
462    ///
463    /// # Panics
464    ///
465    /// This function panics if it is not called from within a runtime with
466    /// IO enabled.
467    ///
468    /// The runtime is usually set implicitly when this function is called
469    /// from a future driven by a Tokio runtime, otherwise runtime can be set
470    /// explicitly with [`Runtime::enter`](crate::runtime::Runtime::enter) function.
471    /// # Examples
472    /// ```
473    /// # use std::error::Error;
474    /// # #[tokio::main]
475    /// # async fn main() -> Result<(), Box<dyn Error>> {
476    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
477    /// use tokio::net::UnixDatagram;
478    /// use std::os::unix::net::UnixDatagram as StdUDS;
479    /// use tempfile::tempdir;
480    ///
481    /// // We use a temporary directory so that the socket
482    /// // files left by the bound sockets will get cleaned up.
483    /// let tmp = tempdir()?;
484    ///
485    /// // Bind the socket to a filesystem path
486    /// let socket_path = tmp.path().join("socket");
487    /// let std_socket = StdUDS::bind(&socket_path)?;
488    /// std_socket.set_nonblocking(true)?;
489    /// let tokio_socket = UnixDatagram::from_std(std_socket)?;
490    ///
491    /// # Ok(())
492    /// # }
493    /// ```
494    #[track_caller]
495    pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
496        check_socket_for_blocking(&datagram)?;
497
498        let socket = mio::net::UnixDatagram::from_std(datagram);
499        let io = PollEvented::new(socket)?;
500        Ok(UnixDatagram { io })
501    }
502
503    /// Turns a [`tokio::net::UnixDatagram`] into a [`std::os::unix::net::UnixDatagram`].
504    ///
505    /// The returned [`std::os::unix::net::UnixDatagram`] will have nonblocking
506    /// mode set as `true`. Use [`set_nonblocking`] to change the blocking mode
507    /// if needed.
508    ///
509    /// # Examples
510    ///
511    /// ```rust,no_run
512    /// # use std::error::Error;
513    /// # async fn dox() -> Result<(), Box<dyn Error>> {
514    /// let tokio_socket = tokio::net::UnixDatagram::bind("/path/to/the/socket")?;
515    /// let std_socket = tokio_socket.into_std()?;
516    /// std_socket.set_nonblocking(false)?;
517    /// # Ok(())
518    /// # }
519    /// ```
520    ///
521    /// [`tokio::net::UnixDatagram`]: UnixDatagram
522    /// [`std::os::unix::net::UnixDatagram`]: std::os::unix::net::UnixDatagram
523    /// [`set_nonblocking`]: fn@std::os::unix::net::UnixDatagram::set_nonblocking
524    pub fn into_std(self) -> io::Result<std::os::unix::net::UnixDatagram> {
525        self.io
526            .into_inner()
527            .map(IntoRawFd::into_raw_fd)
528            .map(|raw_fd| unsafe { std::os::unix::net::UnixDatagram::from_raw_fd(raw_fd) })
529    }
530
531    fn new(socket: mio::net::UnixDatagram) -> io::Result<UnixDatagram> {
532        let io = PollEvented::new(socket)?;
533        Ok(UnixDatagram { io })
534    }
535
536    /// Creates a new `UnixDatagram` which is not bound to any address.
537    ///
538    /// # Examples
539    /// ```
540    /// # use std::error::Error;
541    /// # #[tokio::main]
542    /// # async fn main() -> Result<(), Box<dyn Error>> {
543    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
544    /// use tokio::net::UnixDatagram;
545    /// use tempfile::tempdir;
546    ///
547    /// // Create an unbound socket
548    /// let tx = UnixDatagram::unbound()?;
549    ///
550    /// // Create another, bound socket
551    /// let tmp = tempdir()?;
552    /// let rx_path = tmp.path().join("rx");
553    /// let rx = UnixDatagram::bind(&rx_path)?;
554    ///
555    /// // Send to the bound socket
556    /// let bytes = b"hello world";
557    /// tx.send_to(bytes, &rx_path).await?;
558    ///
559    /// let mut buf = vec![0u8; 24];
560    /// let (size, addr) = rx.recv_from(&mut buf).await?;
561    ///
562    /// let dgram = &buf[..size];
563    /// assert_eq!(dgram, bytes);
564    ///
565    /// # Ok(())
566    /// # }
567    /// ```
568    pub fn unbound() -> io::Result<UnixDatagram> {
569        let socket = mio::net::UnixDatagram::unbound()?;
570        UnixDatagram::new(socket)
571    }
572
573    /// Connects the socket to the specified address.
574    ///
575    /// The `send` method may be used to send data to the specified address.
576    /// `recv` and `recv_from` will only receive data from that address.
577    ///
578    /// # Examples
579    /// ```
580    /// # use std::error::Error;
581    /// # #[tokio::main]
582    /// # async fn main() -> Result<(), Box<dyn Error>> {
583    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
584    /// use tokio::net::UnixDatagram;
585    /// use tempfile::tempdir;
586    ///
587    /// // Create an unbound socket
588    /// let tx = UnixDatagram::unbound()?;
589    ///
590    /// // Create another, bound socket
591    /// let tmp = tempdir()?;
592    /// let rx_path = tmp.path().join("rx");
593    /// let rx = UnixDatagram::bind(&rx_path)?;
594    ///
595    /// // Connect to the bound socket
596    /// tx.connect(&rx_path)?;
597    ///
598    /// // Send to the bound socket
599    /// let bytes = b"hello world";
600    /// tx.send(bytes).await?;
601    ///
602    /// let mut buf = vec![0u8; 24];
603    /// let (size, addr) = rx.recv_from(&mut buf).await?;
604    ///
605    /// let dgram = &buf[..size];
606    /// assert_eq!(dgram, bytes);
607    ///
608    /// # Ok(())
609    /// # }
610    /// ```
611    pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
612        self.io.connect(path)
613    }
614
615    /// Sends data on the socket to the socket's peer.
616    ///
617    /// # Cancel safety
618    ///
619    /// This method is cancel safe. If `send` is used as a branch in
620    /// [`tokio::select!`](crate::select) and another branch
621    /// completes first, then it is guaranteed that the message was not sent.
622    ///
623    /// # Examples
624    /// ```
625    /// # use std::error::Error;
626    /// # #[tokio::main]
627    /// # async fn main() -> Result<(), Box<dyn Error>> {
628    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
629    /// use tokio::net::UnixDatagram;
630    ///
631    /// // Create the pair of sockets
632    /// let (sock1, sock2) = UnixDatagram::pair()?;
633    ///
634    /// // Since the sockets are paired, the paired send/recv
635    /// // functions can be used
636    /// let bytes = b"hello world";
637    /// sock1.send(bytes).await?;
638    ///
639    /// let mut buff = vec![0u8; 24];
640    /// let size = sock2.recv(&mut buff).await?;
641    ///
642    /// let dgram = &buff[..size];
643    /// assert_eq!(dgram, bytes);
644    ///
645    /// # Ok(())
646    /// # }
647    /// ```
648    pub async fn send(&self, buf: &[u8]) -> io::Result<usize> {
649        self.io
650            .registration()
651            .async_io(Interest::WRITABLE, || self.io.send(buf))
652            .await
653    }
654
655    /// Tries to send a datagram to the peer without waiting.
656    ///
657    /// # Examples
658    ///
659    /// ```no_run
660    /// use tokio::net::UnixDatagram;
661    /// use std::io;
662    ///
663    /// #[tokio::main]
664    /// async fn main() -> io::Result<()> {
665    ///     let dir = tempfile::tempdir().unwrap();
666    ///     let client_path = dir.path().join("client.sock");
667    ///     let server_path = dir.path().join("server.sock");
668    ///     let socket = UnixDatagram::bind(&client_path)?;
669    ///     socket.connect(&server_path)?;
670    ///
671    ///     loop {
672    ///         // Wait for the socket to be writable
673    ///         socket.writable().await?;
674    ///
675    ///         // Try to send data, this may still fail with `WouldBlock`
676    ///         // if the readiness event is a false positive.
677    ///         match socket.try_send(b"hello world") {
678    ///             Ok(n) => {
679    ///                 break;
680    ///             }
681    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
682    ///                 continue;
683    ///             }
684    ///             Err(e) => {
685    ///                 return Err(e);
686    ///             }
687    ///         }
688    ///     }
689    ///
690    ///     Ok(())
691    /// }
692    /// ```
693    pub fn try_send(&self, buf: &[u8]) -> io::Result<usize> {
694        self.io
695            .registration()
696            .try_io(Interest::WRITABLE, || self.io.send(buf))
697    }
698
699    /// Tries to send a datagram to the peer without waiting.
700    ///
701    /// # Examples
702    ///
703    /// ```no_run
704    /// use tokio::net::UnixDatagram;
705    /// use std::io;
706    ///
707    /// #[tokio::main]
708    /// async fn main() -> io::Result<()> {
709    ///     let dir = tempfile::tempdir().unwrap();
710    ///     let client_path = dir.path().join("client.sock");
711    ///     let server_path = dir.path().join("server.sock");
712    ///     let socket = UnixDatagram::bind(&client_path)?;
713    ///
714    ///     loop {
715    ///         // Wait for the socket to be writable
716    ///         socket.writable().await?;
717    ///
718    ///         // Try to send data, this may still fail with `WouldBlock`
719    ///         // if the readiness event is a false positive.
720    ///         match socket.try_send_to(b"hello world", &server_path) {
721    ///             Ok(n) => {
722    ///                 break;
723    ///             }
724    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
725    ///                 continue;
726    ///             }
727    ///             Err(e) => {
728    ///                 return Err(e);
729    ///             }
730    ///         }
731    ///     }
732    ///
733    ///     Ok(())
734    /// }
735    /// ```
736    pub fn try_send_to<P>(&self, buf: &[u8], target: P) -> io::Result<usize>
737    where
738        P: AsRef<Path>,
739    {
740        self.io
741            .registration()
742            .try_io(Interest::WRITABLE, || self.io.send_to(buf, target))
743    }
744
745    /// Receives data from the socket.
746    ///
747    /// # Cancel safety
748    ///
749    /// This method is cancel safe. If `recv` is used as a branch in
750    /// [`tokio::select!`](crate::select) and another branch
751    /// completes first, it is guaranteed that no messages were received on this
752    /// socket.
753    ///
754    /// # Examples
755    /// ```
756    /// # use std::error::Error;
757    /// # #[tokio::main]
758    /// # async fn main() -> Result<(), Box<dyn Error>> {
759    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
760    /// use tokio::net::UnixDatagram;
761    ///
762    /// // Create the pair of sockets
763    /// let (sock1, sock2) = UnixDatagram::pair()?;
764    ///
765    /// // Since the sockets are paired, the paired send/recv
766    /// // functions can be used
767    /// let bytes = b"hello world";
768    /// sock1.send(bytes).await?;
769    ///
770    /// let mut buff = vec![0u8; 24];
771    /// let size = sock2.recv(&mut buff).await?;
772    ///
773    /// let dgram = &buff[..size];
774    /// assert_eq!(dgram, bytes);
775    ///
776    /// # Ok(())
777    /// # }
778    /// ```
779    pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
780        self.io
781            .registration()
782            .async_io(Interest::READABLE, || self.io.recv(buf))
783            .await
784    }
785
786    /// Tries to receive a datagram from the peer without waiting.
787    ///
788    /// # Examples
789    ///
790    /// ```no_run
791    /// use tokio::net::UnixDatagram;
792    /// use std::io;
793    ///
794    /// #[tokio::main]
795    /// async fn main() -> io::Result<()> {
796    ///     // Connect to a peer
797    ///     let dir = tempfile::tempdir().unwrap();
798    ///     let client_path = dir.path().join("client.sock");
799    ///     let server_path = dir.path().join("server.sock");
800    ///     let socket = UnixDatagram::bind(&client_path)?;
801    ///     socket.connect(&server_path)?;
802    ///
803    ///     loop {
804    ///         // Wait for the socket to be readable
805    ///         socket.readable().await?;
806    ///
807    ///         // The buffer is **not** included in the async task and will
808    ///         // only exist on the stack.
809    ///         let mut buf = [0; 1024];
810    ///
811    ///         // Try to recv data, this may still fail with `WouldBlock`
812    ///         // if the readiness event is a false positive.
813    ///         match socket.try_recv(&mut buf) {
814    ///             Ok(n) => {
815    ///                 println!("GOT {:?}", &buf[..n]);
816    ///                 break;
817    ///             }
818    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
819    ///                 continue;
820    ///             }
821    ///             Err(e) => {
822    ///                 return Err(e);
823    ///             }
824    ///         }
825    ///     }
826    ///
827    ///     Ok(())
828    /// }
829    /// ```
830    pub fn try_recv(&self, buf: &mut [u8]) -> io::Result<usize> {
831        self.io
832            .registration()
833            .try_io(Interest::READABLE, || self.io.recv(buf))
834    }
835
836    cfg_io_util! {
837        /// Tries to receive data from the socket without waiting.
838        ///
839        /// This method can be used even if `buf` is uninitialized.
840        ///
841        /// # Examples
842        ///
843        /// ```no_run
844        /// use tokio::net::UnixDatagram;
845        /// use std::io;
846        ///
847        /// #[tokio::main]
848        /// async fn main() -> io::Result<()> {
849        ///     // Connect to a peer
850        ///     let dir = tempfile::tempdir().unwrap();
851        ///     let client_path = dir.path().join("client.sock");
852        ///     let server_path = dir.path().join("server.sock");
853        ///     let socket = UnixDatagram::bind(&client_path)?;
854        ///
855        ///     loop {
856        ///         // Wait for the socket to be readable
857        ///         socket.readable().await?;
858        ///
859        ///         let mut buf = Vec::with_capacity(1024);
860        ///
861        ///         // Try to recv data, this may still fail with `WouldBlock`
862        ///         // if the readiness event is a false positive.
863        ///         match socket.try_recv_buf_from(&mut buf) {
864        ///             Ok((n, _addr)) => {
865        ///                 println!("GOT {:?}", &buf[..n]);
866        ///                 break;
867        ///             }
868        ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
869        ///                 continue;
870        ///             }
871        ///             Err(e) => {
872        ///                 return Err(e);
873        ///             }
874        ///         }
875        ///     }
876        ///
877        ///     Ok(())
878        /// }
879        /// ```
880        pub fn try_recv_buf_from<B: BufMut>(&self, buf: &mut B) -> io::Result<(usize, SocketAddr)> {
881            let (n, addr) = self.io.registration().try_io(Interest::READABLE, || {
882                let dst = buf.chunk_mut();
883                let dst =
884                    unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
885
886                // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
887                // buffer.
888                let (n, addr) = (*self.io).recv_from(dst)?;
889
890                unsafe {
891                    buf.advance_mut(n);
892                }
893
894                Ok((n, addr))
895            })?;
896
897            Ok((n, SocketAddr(addr)))
898        }
899
900        /// Receives from the socket, advances the
901        /// buffer's internal cursor and returns how many bytes were read and the origin.
902        ///
903        /// This method can be used even if `buf` is uninitialized.
904        ///
905        /// # Examples
906        /// ```
907        /// # use std::error::Error;
908        /// # #[tokio::main]
909        /// # async fn main() -> Result<(), Box<dyn Error>> {
910        /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
911        /// use tokio::net::UnixDatagram;
912        /// use tempfile::tempdir;
913        ///
914        /// // We use a temporary directory so that the socket
915        /// // files left by the bound sockets will get cleaned up.
916        /// let tmp = tempdir()?;
917        ///
918        /// // Bind each socket to a filesystem path
919        /// let tx_path = tmp.path().join("tx");
920        /// let tx = UnixDatagram::bind(&tx_path)?;
921        /// let rx_path = tmp.path().join("rx");
922        /// let rx = UnixDatagram::bind(&rx_path)?;
923        ///
924        /// let bytes = b"hello world";
925        /// tx.send_to(bytes, &rx_path).await?;
926        ///
927        /// let mut buf = Vec::with_capacity(24);
928        /// let (size, addr) = rx.recv_buf_from(&mut buf).await?;
929        ///
930        /// let dgram = &buf[..size];
931        /// assert_eq!(dgram, bytes);
932        /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
933        ///
934        /// # Ok(())
935        /// # }
936        /// ```
937        pub async fn recv_buf_from<B: BufMut>(&self, buf: &mut B) -> io::Result<(usize, SocketAddr)> {
938            self.io.registration().async_io(Interest::READABLE, || {
939                let dst = buf.chunk_mut();
940                let dst =
941                    unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
942
943                // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
944                // buffer.
945                let (n, addr) = (*self.io).recv_from(dst)?;
946
947                unsafe {
948                    buf.advance_mut(n);
949                }
950                Ok((n,SocketAddr(addr)))
951            }).await
952        }
953
954        /// Tries to read data from the stream into the provided buffer, advancing the
955        /// buffer's internal cursor, returning how many bytes were read.
956        ///
957        /// This method can be used even if `buf` is uninitialized.
958        ///
959        /// # Examples
960        ///
961        /// ```no_run
962        /// use tokio::net::UnixDatagram;
963        /// use std::io;
964        ///
965        /// #[tokio::main]
966        /// async fn main() -> io::Result<()> {
967        ///     // Connect to a peer
968        ///     let dir = tempfile::tempdir().unwrap();
969        ///     let client_path = dir.path().join("client.sock");
970        ///     let server_path = dir.path().join("server.sock");
971        ///     let socket = UnixDatagram::bind(&client_path)?;
972        ///     socket.connect(&server_path)?;
973        ///
974        ///     loop {
975        ///         // Wait for the socket to be readable
976        ///         socket.readable().await?;
977        ///
978        ///         let mut buf = Vec::with_capacity(1024);
979        ///
980        ///         // Try to recv data, this may still fail with `WouldBlock`
981        ///         // if the readiness event is a false positive.
982        ///         match socket.try_recv_buf(&mut buf) {
983        ///             Ok(n) => {
984        ///                 println!("GOT {:?}", &buf[..n]);
985        ///                 break;
986        ///             }
987        ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
988        ///                 continue;
989        ///             }
990        ///             Err(e) => {
991        ///                 return Err(e);
992        ///             }
993        ///         }
994        ///     }
995        ///
996        ///     Ok(())
997        /// }
998        /// ```
999        pub fn try_recv_buf<B: BufMut>(&self, buf: &mut B) -> io::Result<usize> {
1000            self.io.registration().try_io(Interest::READABLE, || {
1001                let dst = buf.chunk_mut();
1002                let dst =
1003                    unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
1004
1005                // Safety: We trust `UnixDatagram::recv` to have filled up `n` bytes in the
1006                // buffer.
1007                let n = (*self.io).recv(dst)?;
1008
1009                unsafe {
1010                    buf.advance_mut(n);
1011                }
1012
1013                Ok(n)
1014            })
1015        }
1016
1017        /// Receives data from the socket from the address to which it is connected,
1018        /// advancing the buffer's internal cursor, returning how many bytes were read.
1019        ///
1020        /// This method can be used even if `buf` is uninitialized.
1021        ///
1022        /// # Examples
1023        /// ```
1024        /// # use std::error::Error;
1025        /// # #[tokio::main]
1026        /// # async fn main() -> Result<(), Box<dyn Error>> {
1027        /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1028        /// use tokio::net::UnixDatagram;
1029        ///
1030        /// // Create the pair of sockets
1031        /// let (sock1, sock2) = UnixDatagram::pair()?;
1032        ///
1033        /// // Since the sockets are paired, the paired send/recv
1034        /// // functions can be used
1035        /// let bytes = b"hello world";
1036        /// sock1.send(bytes).await?;
1037        ///
1038        /// let mut buff = Vec::with_capacity(24);
1039        /// let size = sock2.recv_buf(&mut buff).await?;
1040        ///
1041        /// let dgram = &buff[..size];
1042        /// assert_eq!(dgram, bytes);
1043        ///
1044        /// # Ok(())
1045        /// # }
1046        /// ```
1047        pub async fn recv_buf<B: BufMut>(&self, buf: &mut B) -> io::Result<usize> {
1048            self.io.registration().async_io(Interest::READABLE, || {
1049                let dst = buf.chunk_mut();
1050                let dst =
1051                    unsafe { &mut *(dst as *mut _ as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]) };
1052
1053                // Safety: We trust `UnixDatagram::recv_from` to have filled up `n` bytes in the
1054                // buffer.
1055                let n = (*self.io).recv(dst)?;
1056
1057                unsafe {
1058                    buf.advance_mut(n);
1059                }
1060                Ok(n)
1061            }).await
1062        }
1063    }
1064
1065    /// Sends data on the socket to the specified address.
1066    ///
1067    /// # Cancel safety
1068    ///
1069    /// This method is cancel safe. If `send_to` is used as a branch in
1070    /// [`tokio::select!`](crate::select) and another branch
1071    /// completes first, then it is guaranteed that the message was not sent.
1072    ///
1073    /// # Examples
1074    /// ```
1075    /// # use std::error::Error;
1076    /// # #[tokio::main]
1077    /// # async fn main() -> Result<(), Box<dyn Error>> {
1078    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1079    /// use tokio::net::UnixDatagram;
1080    /// use tempfile::tempdir;
1081    ///
1082    /// // We use a temporary directory so that the socket
1083    /// // files left by the bound sockets will get cleaned up.
1084    /// let tmp = tempdir()?;
1085    ///
1086    /// // Bind each socket to a filesystem path
1087    /// let tx_path = tmp.path().join("tx");
1088    /// let tx = UnixDatagram::bind(&tx_path)?;
1089    /// let rx_path = tmp.path().join("rx");
1090    /// let rx = UnixDatagram::bind(&rx_path)?;
1091    ///
1092    /// let bytes = b"hello world";
1093    /// tx.send_to(bytes, &rx_path).await?;
1094    ///
1095    /// let mut buf = vec![0u8; 24];
1096    /// let (size, addr) = rx.recv_from(&mut buf).await?;
1097    ///
1098    /// let dgram = &buf[..size];
1099    /// assert_eq!(dgram, bytes);
1100    /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
1101    ///
1102    /// # Ok(())
1103    /// # }
1104    /// ```
1105    pub async fn send_to<P>(&self, buf: &[u8], target: P) -> io::Result<usize>
1106    where
1107        P: AsRef<Path>,
1108    {
1109        self.io
1110            .registration()
1111            .async_io(Interest::WRITABLE, || self.io.send_to(buf, target.as_ref()))
1112            .await
1113    }
1114
1115    /// Receives data from the socket.
1116    ///
1117    /// # Cancel safety
1118    ///
1119    /// This method is cancel safe. If `recv_from` is used as a branch in
1120    /// [`tokio::select!`](crate::select) and another branch
1121    /// completes first, it is guaranteed that no messages were received on this
1122    /// socket.
1123    ///
1124    /// # Examples
1125    /// ```
1126    /// # use std::error::Error;
1127    /// # #[tokio::main]
1128    /// # async fn main() -> Result<(), Box<dyn Error>> {
1129    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1130    /// use tokio::net::UnixDatagram;
1131    /// use tempfile::tempdir;
1132    ///
1133    /// // We use a temporary directory so that the socket
1134    /// // files left by the bound sockets will get cleaned up.
1135    /// let tmp = tempdir()?;
1136    ///
1137    /// // Bind each socket to a filesystem path
1138    /// let tx_path = tmp.path().join("tx");
1139    /// let tx = UnixDatagram::bind(&tx_path)?;
1140    /// let rx_path = tmp.path().join("rx");
1141    /// let rx = UnixDatagram::bind(&rx_path)?;
1142    ///
1143    /// let bytes = b"hello world";
1144    /// tx.send_to(bytes, &rx_path).await?;
1145    ///
1146    /// let mut buf = vec![0u8; 24];
1147    /// let (size, addr) = rx.recv_from(&mut buf).await?;
1148    ///
1149    /// let dgram = &buf[..size];
1150    /// assert_eq!(dgram, bytes);
1151    /// assert_eq!(addr.as_pathname().unwrap(), &tx_path);
1152    ///
1153    /// # Ok(())
1154    /// # }
1155    /// ```
1156    pub async fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
1157        let (n, addr) = self
1158            .io
1159            .registration()
1160            .async_io(Interest::READABLE, || self.io.recv_from(buf))
1161            .await?;
1162
1163        Ok((n, SocketAddr(addr)))
1164    }
1165
1166    /// Attempts to receive a single datagram on the specified address.
1167    ///
1168    /// Note that on multiple calls to a `poll_*` method in the `recv` direction, only the
1169    /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1170    /// receive a wakeup.
1171    ///
1172    /// # Return value
1173    ///
1174    /// The function returns:
1175    ///
1176    /// * `Poll::Pending` if the socket is not ready to read
1177    /// * `Poll::Ready(Ok(addr))` reads data from `addr` into `ReadBuf` if the socket is ready
1178    /// * `Poll::Ready(Err(e))` if an error is encountered.
1179    ///
1180    /// # Errors
1181    ///
1182    /// This function may encounter any standard I/O error except `WouldBlock`.
1183    pub fn poll_recv_from(
1184        &self,
1185        cx: &mut Context<'_>,
1186        buf: &mut ReadBuf<'_>,
1187    ) -> Poll<io::Result<SocketAddr>> {
1188        #[allow(clippy::blocks_in_conditions)]
1189        let (n, addr) = ready!(self.io.registration().poll_read_io(cx, || {
1190            // Safety: will not read the maybe uninitialized bytes.
1191            let b = unsafe {
1192                &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
1193            };
1194
1195            self.io.recv_from(b)
1196        }))?;
1197
1198        // Safety: We trust `recv` to have filled up `n` bytes in the buffer.
1199        unsafe {
1200            buf.assume_init(n);
1201        }
1202        buf.advance(n);
1203        Poll::Ready(Ok(SocketAddr(addr)))
1204    }
1205
1206    /// Attempts to send data to the specified address.
1207    ///
1208    /// Note that on multiple calls to a `poll_*` method in the send direction, only the
1209    /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1210    /// receive a wakeup.
1211    ///
1212    /// # Return value
1213    ///
1214    /// The function returns:
1215    ///
1216    /// * `Poll::Pending` if the socket is not ready to write
1217    /// * `Poll::Ready(Ok(n))` `n` is the number of bytes sent.
1218    /// * `Poll::Ready(Err(e))` if an error is encountered.
1219    ///
1220    /// # Errors
1221    ///
1222    /// This function may encounter any standard I/O error except `WouldBlock`.
1223    pub fn poll_send_to<P>(
1224        &self,
1225        cx: &mut Context<'_>,
1226        buf: &[u8],
1227        target: P,
1228    ) -> Poll<io::Result<usize>>
1229    where
1230        P: AsRef<Path>,
1231    {
1232        self.io
1233            .registration()
1234            .poll_write_io(cx, || self.io.send_to(buf, target.as_ref()))
1235    }
1236
1237    /// Attempts to send data on the socket to the remote address to which it
1238    /// was previously `connect`ed.
1239    ///
1240    /// The [`connect`] method will connect this socket to a remote address.
1241    /// This method will fail if the socket is not connected.
1242    ///
1243    /// Note that on multiple calls to a `poll_*` method in the send direction,
1244    /// only the `Waker` from the `Context` passed to the most recent call will
1245    /// be scheduled to receive a wakeup.
1246    ///
1247    /// # Return value
1248    ///
1249    /// The function returns:
1250    ///
1251    /// * `Poll::Pending` if the socket is not available to write
1252    /// * `Poll::Ready(Ok(n))` `n` is the number of bytes sent
1253    /// * `Poll::Ready(Err(e))` if an error is encountered.
1254    ///
1255    /// # Errors
1256    ///
1257    /// This function may encounter any standard I/O error except `WouldBlock`.
1258    ///
1259    /// [`connect`]: method@Self::connect
1260    pub fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
1261        self.io
1262            .registration()
1263            .poll_write_io(cx, || self.io.send(buf))
1264    }
1265
1266    /// Attempts to receive a single datagram message on the socket from the remote
1267    /// address to which it is `connect`ed.
1268    ///
1269    /// The [`connect`] method will connect this socket to a remote address. This method
1270    /// resolves to an error if the socket is not connected.
1271    ///
1272    /// Note that on multiple calls to a `poll_*` method in the `recv` direction, only the
1273    /// `Waker` from the `Context` passed to the most recent call will be scheduled to
1274    /// receive a wakeup.
1275    ///
1276    /// # Return value
1277    ///
1278    /// The function returns:
1279    ///
1280    /// * `Poll::Pending` if the socket is not ready to read
1281    /// * `Poll::Ready(Ok(()))` reads data `ReadBuf` if the socket is ready
1282    /// * `Poll::Ready(Err(e))` if an error is encountered.
1283    ///
1284    /// # Errors
1285    ///
1286    /// This function may encounter any standard I/O error except `WouldBlock`.
1287    ///
1288    /// [`connect`]: method@Self::connect
1289    pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
1290        #[allow(clippy::blocks_in_conditions)]
1291        let n = ready!(self.io.registration().poll_read_io(cx, || {
1292            // Safety: will not read the maybe uninitialized bytes.
1293            let b = unsafe {
1294                &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8])
1295            };
1296
1297            self.io.recv(b)
1298        }))?;
1299
1300        // Safety: We trust `recv` to have filled up `n` bytes in the buffer.
1301        unsafe {
1302            buf.assume_init(n);
1303        }
1304        buf.advance(n);
1305        Poll::Ready(Ok(()))
1306    }
1307
1308    /// Tries to receive data from the socket without waiting.
1309    ///
1310    /// # Examples
1311    ///
1312    /// ```no_run
1313    /// use tokio::net::UnixDatagram;
1314    /// use std::io;
1315    ///
1316    /// #[tokio::main]
1317    /// async fn main() -> io::Result<()> {
1318    ///     // Connect to a peer
1319    ///     let dir = tempfile::tempdir().unwrap();
1320    ///     let client_path = dir.path().join("client.sock");
1321    ///     let server_path = dir.path().join("server.sock");
1322    ///     let socket = UnixDatagram::bind(&client_path)?;
1323    ///
1324    ///     loop {
1325    ///         // Wait for the socket to be readable
1326    ///         socket.readable().await?;
1327    ///
1328    ///         // The buffer is **not** included in the async task and will
1329    ///         // only exist on the stack.
1330    ///         let mut buf = [0; 1024];
1331    ///
1332    ///         // Try to recv data, this may still fail with `WouldBlock`
1333    ///         // if the readiness event is a false positive.
1334    ///         match socket.try_recv_from(&mut buf) {
1335    ///             Ok((n, _addr)) => {
1336    ///                 println!("GOT {:?}", &buf[..n]);
1337    ///                 break;
1338    ///             }
1339    ///             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
1340    ///                 continue;
1341    ///             }
1342    ///             Err(e) => {
1343    ///                 return Err(e);
1344    ///             }
1345    ///         }
1346    ///     }
1347    ///
1348    ///     Ok(())
1349    /// }
1350    /// ```
1351    pub fn try_recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
1352        let (n, addr) = self
1353            .io
1354            .registration()
1355            .try_io(Interest::READABLE, || self.io.recv_from(buf))?;
1356
1357        Ok((n, SocketAddr(addr)))
1358    }
1359
1360    /// Tries to read or write from the socket using a user-provided IO operation.
1361    ///
1362    /// If the socket is ready, the provided closure is called. The closure
1363    /// should attempt to perform IO operation on the socket by manually
1364    /// calling the appropriate syscall. If the operation fails because the
1365    /// socket is not actually ready, then the closure should return a
1366    /// `WouldBlock` error and the readiness flag is cleared. The return value
1367    /// of the closure is then returned by `try_io`.
1368    ///
1369    /// If the socket is not ready, then the closure is not called
1370    /// and a `WouldBlock` error is returned.
1371    ///
1372    /// The closure should only return a `WouldBlock` error if it has performed
1373    /// an IO operation on the socket that failed due to the socket not being
1374    /// ready. Returning a `WouldBlock` error in any other situation will
1375    /// incorrectly clear the readiness flag, which can cause the socket to
1376    /// behave incorrectly.
1377    ///
1378    /// The closure should not perform the IO operation using any of the methods
1379    /// defined on the Tokio `UnixDatagram` type, as this will mess with the
1380    /// readiness flag and can cause the socket to behave incorrectly.
1381    ///
1382    /// This method is not intended to be used with combined interests.
1383    /// The closure should perform only one type of IO operation, so it should not
1384    /// require more than one ready state. This method may panic or sleep forever
1385    /// if it is called with a combined interest.
1386    ///
1387    /// Usually, [`readable()`], [`writable()`] or [`ready()`] is used with this function.
1388    ///
1389    /// [`readable()`]: UnixDatagram::readable()
1390    /// [`writable()`]: UnixDatagram::writable()
1391    /// [`ready()`]: UnixDatagram::ready()
1392    pub fn try_io<R>(
1393        &self,
1394        interest: Interest,
1395        f: impl FnOnce() -> io::Result<R>,
1396    ) -> io::Result<R> {
1397        self.io
1398            .registration()
1399            .try_io(interest, || self.io.try_io(f))
1400    }
1401
1402    /// Reads or writes from the socket using a user-provided IO operation.
1403    ///
1404    /// The readiness of the socket is awaited and when the socket is ready,
1405    /// the provided closure is called. The closure should attempt to perform
1406    /// IO operation on the socket by manually calling the appropriate syscall.
1407    /// If the operation fails because the socket is not actually ready,
1408    /// then the closure should return a `WouldBlock` error. In such case the
1409    /// readiness flag is cleared and the socket readiness is awaited again.
1410    /// This loop is repeated until the closure returns an `Ok` or an error
1411    /// other than `WouldBlock`.
1412    ///
1413    /// The closure should only return a `WouldBlock` error if it has performed
1414    /// an IO operation on the socket that failed due to the socket not being
1415    /// ready. Returning a `WouldBlock` error in any other situation will
1416    /// incorrectly clear the readiness flag, which can cause the socket to
1417    /// behave incorrectly.
1418    ///
1419    /// The closure should not perform the IO operation using any of the methods
1420    /// defined on the Tokio `UnixDatagram` type, as this will mess with the
1421    /// readiness flag and can cause the socket to behave incorrectly.
1422    ///
1423    /// This method is not intended to be used with combined interests.
1424    /// The closure should perform only one type of IO operation, so it should not
1425    /// require more than one ready state. This method may panic or sleep forever
1426    /// if it is called with a combined interest.
1427    pub async fn async_io<R>(
1428        &self,
1429        interest: Interest,
1430        mut f: impl FnMut() -> io::Result<R>,
1431    ) -> io::Result<R> {
1432        self.io
1433            .registration()
1434            .async_io(interest, || self.io.try_io(&mut f))
1435            .await
1436    }
1437
1438    /// Returns the local address that this socket is bound to.
1439    ///
1440    /// # Examples
1441    /// For a socket bound to a local path
1442    /// ```
1443    /// # use std::error::Error;
1444    /// # #[tokio::main]
1445    /// # async fn main() -> Result<(), Box<dyn Error>> {
1446    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1447    /// use tokio::net::UnixDatagram;
1448    /// use tempfile::tempdir;
1449    ///
1450    /// // We use a temporary directory so that the socket
1451    /// // files left by the bound sockets will get cleaned up.
1452    /// let tmp = tempdir()?;
1453    ///
1454    /// // Bind socket to a filesystem path
1455    /// let socket_path = tmp.path().join("socket");
1456    /// let socket = UnixDatagram::bind(&socket_path)?;
1457    ///
1458    /// assert_eq!(socket.local_addr()?.as_pathname().unwrap(), &socket_path);
1459    ///
1460    /// # Ok(())
1461    /// # }
1462    /// ```
1463    ///
1464    /// For an unbound socket
1465    /// ```
1466    /// # use std::error::Error;
1467    /// # #[tokio::main]
1468    /// # async fn main() -> Result<(), Box<dyn Error>> {
1469    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1470    /// use tokio::net::UnixDatagram;
1471    ///
1472    /// // Create an unbound socket
1473    /// let socket = UnixDatagram::unbound()?;
1474    ///
1475    /// assert!(socket.local_addr()?.is_unnamed());
1476    ///
1477    /// # Ok(())
1478    /// # }
1479    /// ```
1480    pub fn local_addr(&self) -> io::Result<SocketAddr> {
1481        self.io.local_addr().map(SocketAddr)
1482    }
1483
1484    /// Returns the address of this socket's peer.
1485    ///
1486    /// The `connect` method will connect the socket to a peer.
1487    ///
1488    /// # Examples
1489    /// For a peer with a local path
1490    /// ```
1491    /// # use std::error::Error;
1492    /// # #[tokio::main]
1493    /// # async fn main() -> Result<(), Box<dyn Error>> {
1494    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1495    /// use tokio::net::UnixDatagram;
1496    /// use tempfile::tempdir;
1497    ///
1498    /// // Create an unbound socket
1499    /// let tx = UnixDatagram::unbound()?;
1500    ///
1501    /// // Create another, bound socket
1502    /// let tmp = tempdir()?;
1503    /// let rx_path = tmp.path().join("rx");
1504    /// let rx = UnixDatagram::bind(&rx_path)?;
1505    ///
1506    /// // Connect to the bound socket
1507    /// tx.connect(&rx_path)?;
1508    ///
1509    /// assert_eq!(tx.peer_addr()?.as_pathname().unwrap(), &rx_path);
1510    ///
1511    /// # Ok(())
1512    /// # }
1513    /// ```
1514    ///
1515    /// For an unbound peer
1516    /// ```
1517    /// # use std::error::Error;
1518    /// # #[tokio::main]
1519    /// # async fn main() -> Result<(), Box<dyn Error>> {
1520    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1521    /// use tokio::net::UnixDatagram;
1522    ///
1523    /// // Create the pair of sockets
1524    /// let (sock1, sock2) = UnixDatagram::pair()?;
1525    ///
1526    /// assert!(sock1.peer_addr()?.is_unnamed());
1527    ///
1528    /// # Ok(())
1529    /// # }
1530    /// ```
1531    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
1532        self.io.peer_addr().map(SocketAddr)
1533    }
1534
1535    /// Returns the value of the `SO_ERROR` option.
1536    ///
1537    /// # Examples
1538    /// ```
1539    /// # use std::error::Error;
1540    /// # #[tokio::main]
1541    /// # async fn main() -> Result<(), Box<dyn Error>> {
1542    /// # if cfg!(miri) { return Ok(()); } // No Unix domain sockets in miri.
1543    /// use tokio::net::UnixDatagram;
1544    ///
1545    /// // Create an unbound socket
1546    /// let socket = UnixDatagram::unbound()?;
1547    ///
1548    /// if let Ok(Some(err)) = socket.take_error() {
1549    ///     println!("Got error: {:?}", err);
1550    /// }
1551    ///
1552    /// # Ok(())
1553    /// # }
1554    /// ```
1555    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
1556        self.io.take_error()
1557    }
1558
1559    /// Shuts down the read, write, or both halves of this connection.
1560    ///
1561    /// This function will cause all pending and future I/O calls on the
1562    /// specified portions to immediately return with an appropriate value
1563    /// (see the documentation of `Shutdown`).
1564    ///
1565    /// # Examples
1566    /// ```
1567    /// # use std::error::Error;
1568    /// # #[tokio::main]
1569    /// # async fn main() -> Result<(), Box<dyn Error>> {
1570    /// # if cfg!(miri) { return Ok(()); } // No SOCK_DGRAM for `socketpair` in miri.
1571    /// use tokio::net::UnixDatagram;
1572    /// use std::net::Shutdown;
1573    ///
1574    /// // Create an unbound socket
1575    /// let (socket, other) = UnixDatagram::pair()?;
1576    ///
1577    /// socket.shutdown(Shutdown::Both)?;
1578    ///
1579    /// // NOTE: the following commented out code does NOT work as expected.
1580    /// // Due to an underlying issue, the recv call will block indefinitely.
1581    /// // See: https://github.com/tokio-rs/tokio/issues/1679
1582    /// //let mut buff = vec![0u8; 24];
1583    /// //let size = socket.recv(&mut buff).await?;
1584    /// //assert_eq!(size, 0);
1585    ///
1586    /// let send_result = socket.send(b"hello world").await;
1587    /// assert!(send_result.is_err());
1588    ///
1589    /// # Ok(())
1590    /// # }
1591    /// ```
1592    pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
1593        self.io.shutdown(how)
1594    }
1595}
1596
1597impl TryFrom<std::os::unix::net::UnixDatagram> for UnixDatagram {
1598    type Error = io::Error;
1599
1600    /// Consumes stream, returning the Tokio I/O object.
1601    ///
1602    /// This is equivalent to
1603    /// [`UnixDatagram::from_std(stream)`](UnixDatagram::from_std).
1604    fn try_from(stream: std::os::unix::net::UnixDatagram) -> Result<Self, Self::Error> {
1605        Self::from_std(stream)
1606    }
1607}
1608
1609impl fmt::Debug for UnixDatagram {
1610    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1611        (*self.io).fmt(f)
1612    }
1613}
1614
1615impl AsRawFd for UnixDatagram {
1616    fn as_raw_fd(&self) -> RawFd {
1617        self.io.as_raw_fd()
1618    }
1619}
1620
1621impl AsFd for UnixDatagram {
1622    fn as_fd(&self) -> BorrowedFd<'_> {
1623        unsafe { BorrowedFd::borrow_raw(self.as_raw_fd()) }
1624    }
1625}