summaryrefslogtreecommitdiffstats
path: root/third_party/rust/tokio/src/net/unix
diff options
context:
space:
mode:
Diffstat (limited to 'third_party/rust/tokio/src/net/unix')
-rw-r--r--third_party/rust/tokio/src/net/unix/datagram.rs242
-rw-r--r--third_party/rust/tokio/src/net/unix/incoming.rs42
-rw-r--r--third_party/rust/tokio/src/net/unix/listener.rs229
-rw-r--r--third_party/rust/tokio/src/net/unix/mod.rs18
-rw-r--r--third_party/rust/tokio/src/net/unix/split.rs74
-rw-r--r--third_party/rust/tokio/src/net/unix/stream.rs233
-rw-r--r--third_party/rust/tokio/src/net/unix/ucred.rs151
7 files changed, 989 insertions, 0 deletions
diff --git a/third_party/rust/tokio/src/net/unix/datagram.rs b/third_party/rust/tokio/src/net/unix/datagram.rs
new file mode 100644
index 0000000000..ff0f4241d5
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/datagram.rs
@@ -0,0 +1,242 @@
+use crate::future::poll_fn;
+use crate::io::PollEvented;
+
+use std::convert::TryFrom;
+use std::fmt;
+use std::io;
+use std::net::Shutdown;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::task::{Context, Poll};
+
+cfg_uds! {
+ /// An I/O object representing a Unix datagram socket.
+ pub struct UnixDatagram {
+ io: PollEvented<mio_uds::UnixDatagram>,
+ }
+}
+
+impl UnixDatagram {
+ /// Creates a new `UnixDatagram` bound to the specified path.
+ pub fn bind<P>(path: P) -> io::Result<UnixDatagram>
+ where
+ P: AsRef<Path>,
+ {
+ let socket = mio_uds::UnixDatagram::bind(path)?;
+ UnixDatagram::new(socket)
+ }
+
+ /// Creates an unnamed pair of connected sockets.
+ ///
+ /// This function will create a pair of interconnected Unix sockets for
+ /// communicating back and forth between one another. Each socket will
+ /// be associated with the default event loop's handle.
+ pub fn pair() -> io::Result<(UnixDatagram, UnixDatagram)> {
+ let (a, b) = mio_uds::UnixDatagram::pair()?;
+ let a = UnixDatagram::new(a)?;
+ let b = UnixDatagram::new(b)?;
+
+ Ok((a, b))
+ }
+
+ /// Consumes a `UnixDatagram` in the standard library and returns a
+ /// nonblocking `UnixDatagram` from this crate.
+ ///
+ /// The returned datagram will be associated with the given event loop
+ /// specified by `handle` and is ready to perform I/O.
+ ///
+ /// # Panics
+ ///
+ /// This function panics if thread-local runtime is not set.
+ ///
+ /// The runtime is usually set implicitly when this function is called
+ /// from a future driven by a tokio runtime, otherwise runtime can be set
+ /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+ pub fn from_std(datagram: net::UnixDatagram) -> io::Result<UnixDatagram> {
+ let socket = mio_uds::UnixDatagram::from_datagram(datagram)?;
+ let io = PollEvented::new(socket)?;
+ Ok(UnixDatagram { io })
+ }
+
+ fn new(socket: mio_uds::UnixDatagram) -> io::Result<UnixDatagram> {
+ let io = PollEvented::new(socket)?;
+ Ok(UnixDatagram { io })
+ }
+
+ /// Creates a new `UnixDatagram` which is not bound to any address.
+ pub fn unbound() -> io::Result<UnixDatagram> {
+ let socket = mio_uds::UnixDatagram::unbound()?;
+ UnixDatagram::new(socket)
+ }
+
+ /// Connects the socket to the specified address.
+ ///
+ /// The `send` method may be used to send data to the specified address.
+ /// `recv` and `recv_from` will only receive data from that address.
+ pub fn connect<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
+ self.io.get_ref().connect(path)
+ }
+
+ /// Sends data on the socket to the socket's peer.
+ pub async fn send(&mut self, buf: &[u8]) -> io::Result<usize> {
+ poll_fn(|cx| self.poll_send_priv(cx, buf)).await
+ }
+
+ // Poll IO functions that takes `&self` are provided for the split API.
+ //
+ // They are not public because (taken from the doc of `PollEvented`):
+ //
+ // While `PollEvented` is `Sync` (if the underlying I/O type is `Sync`), the
+ // caller must ensure that there are at most two tasks that use a
+ // `PollEvented` instance concurrently. One for reading and one for writing.
+ // While violating this requirement is "safe" from a Rust memory model point
+ // of view, it will result in unexpected behavior in the form of lost
+ // notifications and tasks hanging.
+ pub(crate) fn poll_send_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<io::Result<usize>> {
+ ready!(self.io.poll_write_ready(cx))?;
+
+ match self.io.get_ref().send(buf) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_write_ready(cx)?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+
+ /// Receives data from the socket.
+ pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+ poll_fn(|cx| self.poll_recv_priv(cx, buf)).await
+ }
+
+ pub(crate) fn poll_recv_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<io::Result<usize>> {
+ ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+ match self.io.get_ref().recv(buf) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_read_ready(cx, mio::Ready::readable())?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+
+ /// Sends data on the socket to the specified address.
+ pub async fn send_to<P>(&mut self, buf: &[u8], target: P) -> io::Result<usize>
+ where
+ P: AsRef<Path> + Unpin,
+ {
+ poll_fn(|cx| self.poll_send_to_priv(cx, buf, target.as_ref())).await
+ }
+
+ pub(crate) fn poll_send_to_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ target: &Path,
+ ) -> Poll<io::Result<usize>> {
+ ready!(self.io.poll_write_ready(cx))?;
+
+ match self.io.get_ref().send_to(buf, target) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_write_ready(cx)?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+
+ /// Receives data from the socket.
+ pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
+ poll_fn(|cx| self.poll_recv_from_priv(cx, buf)).await
+ }
+
+ pub(crate) fn poll_recv_from_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<(usize, SocketAddr), io::Error>> {
+ ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+ match self.io.get_ref().recv_from(buf) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_read_ready(cx, mio::Ready::readable())?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+
+ /// Returns the local address that this socket is bound to.
+ pub fn local_addr(&self) -> io::Result<SocketAddr> {
+ self.io.get_ref().local_addr()
+ }
+
+ /// Returns the address of this socket's peer.
+ ///
+ /// The `connect` method will connect the socket to a peer.
+ pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+ self.io.get_ref().peer_addr()
+ }
+
+ /// Returns the value of the `SO_ERROR` option.
+ pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+ self.io.get_ref().take_error()
+ }
+
+ /// Shuts down the read, write, or both halves of this connection.
+ ///
+ /// This function will cause all pending and future I/O calls on the
+ /// specified portions to immediately return with an appropriate value
+ /// (see the documentation of `Shutdown`).
+ pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+ self.io.get_ref().shutdown(how)
+ }
+}
+
+impl TryFrom<UnixDatagram> for mio_uds::UnixDatagram {
+ type Error = io::Error;
+
+ /// Consumes value, returning the mio I/O object.
+ ///
+ /// See [`PollEvented::into_inner`] for more details about
+ /// resource deregistration that happens during the call.
+ ///
+ /// [`PollEvented::into_inner`]: crate::io::PollEvented::into_inner
+ fn try_from(value: UnixDatagram) -> Result<Self, Self::Error> {
+ value.io.into_inner()
+ }
+}
+
+impl TryFrom<net::UnixDatagram> for UnixDatagram {
+ type Error = io::Error;
+
+ /// Consumes stream, returning the tokio I/O object.
+ ///
+ /// This is equivalent to
+ /// [`UnixDatagram::from_std(stream)`](UnixDatagram::from_std).
+ fn try_from(stream: net::UnixDatagram) -> Result<Self, Self::Error> {
+ Self::from_std(stream)
+ }
+}
+
+impl fmt::Debug for UnixDatagram {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ self.io.get_ref().fmt(f)
+ }
+}
+
+impl AsRawFd for UnixDatagram {
+ fn as_raw_fd(&self) -> RawFd {
+ self.io.get_ref().as_raw_fd()
+ }
+}
diff --git a/third_party/rust/tokio/src/net/unix/incoming.rs b/third_party/rust/tokio/src/net/unix/incoming.rs
new file mode 100644
index 0000000000..af49360435
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/incoming.rs
@@ -0,0 +1,42 @@
+use crate::net::unix::{UnixListener, UnixStream};
+
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Stream of listeners
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Incoming<'a> {
+ inner: &'a mut UnixListener,
+}
+
+impl Incoming<'_> {
+ pub(crate) fn new(listener: &mut UnixListener) -> Incoming<'_> {
+ Incoming { inner: listener }
+ }
+
+ /// Attempts to poll `UnixStream` by polling inner `UnixListener` to accept
+ /// connection.
+ ///
+ /// If `UnixListener` isn't ready yet, `Poll::Pending` is returned and
+ /// current task will be notified by a waker. Otherwise `Poll::Ready` with
+ /// `Result` containing `UnixStream` will be returned.
+ pub fn poll_accept(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<io::Result<UnixStream>> {
+ let (socket, _) = ready!(self.inner.poll_accept(cx))?;
+ Poll::Ready(Ok(socket))
+ }
+}
+
+#[cfg(feature = "stream")]
+impl crate::stream::Stream for Incoming<'_> {
+ type Item = io::Result<UnixStream>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let (socket, _) = ready!(self.inner.poll_accept(cx))?;
+ Poll::Ready(Some(Ok(socket)))
+ }
+}
diff --git a/third_party/rust/tokio/src/net/unix/listener.rs b/third_party/rust/tokio/src/net/unix/listener.rs
new file mode 100644
index 0000000000..5acc1b7e82
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/listener.rs
@@ -0,0 +1,229 @@
+use crate::future::poll_fn;
+use crate::io::PollEvented;
+use crate::net::unix::{Incoming, UnixStream};
+
+use mio::Ready;
+use mio_uds;
+use std::convert::TryFrom;
+use std::fmt;
+use std::io;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::task::{Context, Poll};
+
+cfg_uds! {
+ /// A Unix socket which can accept connections from other Unix sockets.
+ ///
+ /// You can accept a new connection by using the [`accept`](`UnixListener::accept`) method. Alternatively `UnixListener`
+ /// implements the [`Stream`](`crate::stream::Stream`) trait, which allows you to use the listener in places that want a
+ /// stream. The stream will never return `None` and will also not yield the peer's `SocketAddr` structure. Iterating over
+ /// it is equivalent to calling accept in a loop.
+ ///
+ /// # Errors
+ ///
+ /// Note that accepting a connection can lead to various errors and not all
+ /// of them are necessarily fatal ‒ for example having too many open file
+ /// descriptors or the other side closing the connection while it waits in
+ /// an accept queue. These would terminate the stream if not handled in any
+ /// way.
+ ///
+ /// # Examples
+ ///
+ /// ```no_run
+ /// use tokio::net::UnixListener;
+ /// use tokio::stream::StreamExt;
+ ///
+ /// #[tokio::main]
+ /// async fn main() {
+ /// let mut listener = UnixListener::bind("/path/to/the/socket").unwrap();
+ /// while let Some(stream) = listener.next().await {
+ /// match stream {
+ /// Ok(stream) => {
+ /// println!("new client!");
+ /// }
+ /// Err(e) => { /* connection failed */ }
+ /// }
+ /// }
+ /// }
+ /// ```
+ pub struct UnixListener {
+ io: PollEvented<mio_uds::UnixListener>,
+ }
+}
+
+impl UnixListener {
+ /// Creates a new `UnixListener` bound to the specified path.
+ ///
+ /// # Panics
+ ///
+ /// This function panics if thread-local runtime is not set.
+ ///
+ /// The runtime is usually set implicitly when this function is called
+ /// from a future driven by a tokio runtime, otherwise runtime can be set
+ /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+ pub fn bind<P>(path: P) -> io::Result<UnixListener>
+ where
+ P: AsRef<Path>,
+ {
+ let listener = mio_uds::UnixListener::bind(path)?;
+ let io = PollEvented::new(listener)?;
+ Ok(UnixListener { io })
+ }
+
+ /// Consumes a `UnixListener` in the standard library and returns a
+ /// nonblocking `UnixListener` from this crate.
+ ///
+ /// The returned listener will be associated with the given event loop
+ /// specified by `handle` and is ready to perform I/O.
+ ///
+ /// # Panics
+ ///
+ /// This function panics if thread-local runtime is not set.
+ ///
+ /// The runtime is usually set implicitly when this function is called
+ /// from a future driven by a tokio runtime, otherwise runtime can be set
+ /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+ pub fn from_std(listener: net::UnixListener) -> io::Result<UnixListener> {
+ let listener = mio_uds::UnixListener::from_listener(listener)?;
+ let io = PollEvented::new(listener)?;
+ Ok(UnixListener { io })
+ }
+
+ /// Returns the local socket address of this listener.
+ pub fn local_addr(&self) -> io::Result<SocketAddr> {
+ self.io.get_ref().local_addr()
+ }
+
+ /// Returns the value of the `SO_ERROR` option.
+ pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+ self.io.get_ref().take_error()
+ }
+
+ /// Accepts a new incoming connection to this listener.
+ pub async fn accept(&mut self) -> io::Result<(UnixStream, SocketAddr)> {
+ poll_fn(|cx| self.poll_accept(cx)).await
+ }
+
+ pub(crate) fn poll_accept(
+ &mut self,
+ cx: &mut Context<'_>,
+ ) -> Poll<io::Result<(UnixStream, SocketAddr)>> {
+ let (io, addr) = ready!(self.poll_accept_std(cx))?;
+
+ let io = mio_uds::UnixStream::from_stream(io)?;
+ Ok((UnixStream::new(io)?, addr)).into()
+ }
+
+ fn poll_accept_std(
+ &mut self,
+ cx: &mut Context<'_>,
+ ) -> Poll<io::Result<(net::UnixStream, SocketAddr)>> {
+ ready!(self.io.poll_read_ready(cx, Ready::readable()))?;
+
+ match self.io.get_ref().accept_std() {
+ Ok(None) => {
+ self.io.clear_read_ready(cx, Ready::readable())?;
+ Poll::Pending
+ }
+ Ok(Some((sock, addr))) => Ok((sock, addr)).into(),
+ Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_read_ready(cx, Ready::readable())?;
+ Poll::Pending
+ }
+ Err(err) => Err(err).into(),
+ }
+ }
+
+ /// Returns a stream over the connections being received on this listener.
+ ///
+ /// Note that `UnixListener` also directly implements `Stream`.
+ ///
+ /// The returned stream will never return `None` and will also not yield the
+ /// peer's `SocketAddr` structure. Iterating over it is equivalent to
+ /// calling accept in a loop.
+ ///
+ /// # Errors
+ ///
+ /// Note that accepting a connection can lead to various errors and not all
+ /// of them are necessarily fatal ‒ for example having too many open file
+ /// descriptors or the other side closing the connection while it waits in
+ /// an accept queue. These would terminate the stream if not handled in any
+ /// way.
+ ///
+ /// # Examples
+ ///
+ /// ```no_run
+ /// use tokio::net::UnixListener;
+ /// use tokio::stream::StreamExt;
+ ///
+ /// #[tokio::main]
+ /// async fn main() {
+ /// let mut listener = UnixListener::bind("/path/to/the/socket").unwrap();
+ /// let mut incoming = listener.incoming();
+ ///
+ /// while let Some(stream) = incoming.next().await {
+ /// match stream {
+ /// Ok(stream) => {
+ /// println!("new client!");
+ /// }
+ /// Err(e) => { /* connection failed */ }
+ /// }
+ /// }
+ /// }
+ /// ```
+ pub fn incoming(&mut self) -> Incoming<'_> {
+ Incoming::new(self)
+ }
+}
+
+#[cfg(feature = "stream")]
+impl crate::stream::Stream for UnixListener {
+ type Item = io::Result<UnixStream>;
+
+ fn poll_next(
+ mut self: std::pin::Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Self::Item>> {
+ let (socket, _) = ready!(self.poll_accept(cx))?;
+ Poll::Ready(Some(Ok(socket)))
+ }
+}
+
+impl TryFrom<UnixListener> for mio_uds::UnixListener {
+ type Error = io::Error;
+
+ /// Consumes value, returning the mio I/O object.
+ ///
+ /// See [`PollEvented::into_inner`] for more details about
+ /// resource deregistration that happens during the call.
+ ///
+ /// [`PollEvented::into_inner`]: crate::io::PollEvented::into_inner
+ fn try_from(value: UnixListener) -> Result<Self, Self::Error> {
+ value.io.into_inner()
+ }
+}
+
+impl TryFrom<net::UnixListener> for UnixListener {
+ type Error = io::Error;
+
+ /// Consumes stream, returning the tokio I/O object.
+ ///
+ /// This is equivalent to
+ /// [`UnixListener::from_std(stream)`](UnixListener::from_std).
+ fn try_from(stream: net::UnixListener) -> io::Result<Self> {
+ Self::from_std(stream)
+ }
+}
+
+impl fmt::Debug for UnixListener {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ self.io.get_ref().fmt(f)
+ }
+}
+
+impl AsRawFd for UnixListener {
+ fn as_raw_fd(&self) -> RawFd {
+ self.io.get_ref().as_raw_fd()
+ }
+}
diff --git a/third_party/rust/tokio/src/net/unix/mod.rs b/third_party/rust/tokio/src/net/unix/mod.rs
new file mode 100644
index 0000000000..ddba60d10a
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/mod.rs
@@ -0,0 +1,18 @@
+//! Unix domain socket utility types
+
+pub(crate) mod datagram;
+
+mod incoming;
+pub use incoming::Incoming;
+
+pub(crate) mod listener;
+pub(crate) use listener::UnixListener;
+
+mod split;
+pub use split::{ReadHalf, WriteHalf};
+
+pub(crate) mod stream;
+pub(crate) use stream::UnixStream;
+
+mod ucred;
+pub use ucred::UCred;
diff --git a/third_party/rust/tokio/src/net/unix/split.rs b/third_party/rust/tokio/src/net/unix/split.rs
new file mode 100644
index 0000000000..9b9fa5ee1d
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/split.rs
@@ -0,0 +1,74 @@
+//! `UnixStream` split support.
+//!
+//! A `UnixStream` can be split into a read half and a write half with
+//! `UnixStream::split`. The read half implements `AsyncRead` while the write
+//! half implements `AsyncWrite`.
+//!
+//! Compared to the generic split of `AsyncRead + AsyncWrite`, this specialized
+//! split has no associated overhead and enforces all invariants at the type
+//! level.
+
+use crate::io::{AsyncRead, AsyncWrite};
+use crate::net::UnixStream;
+
+use std::io;
+use std::mem::MaybeUninit;
+use std::net::Shutdown;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Read half of a `UnixStream`.
+#[derive(Debug)]
+pub struct ReadHalf<'a>(&'a UnixStream);
+
+/// Write half of a `UnixStream`.
+#[derive(Debug)]
+pub struct WriteHalf<'a>(&'a UnixStream);
+
+pub(crate) fn split(stream: &mut UnixStream) -> (ReadHalf<'_>, WriteHalf<'_>) {
+ (ReadHalf(stream), WriteHalf(stream))
+}
+
+impl AsyncRead for ReadHalf<'_> {
+ unsafe fn prepare_uninitialized_buffer(&self, _: &mut [MaybeUninit<u8>]) -> bool {
+ false
+ }
+
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<io::Result<usize>> {
+ self.0.poll_read_priv(cx, buf)
+ }
+}
+
+impl AsyncWrite for WriteHalf<'_> {
+ fn poll_write(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<io::Result<usize>> {
+ self.0.poll_write_priv(cx, buf)
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+
+ fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+ self.0.shutdown(Shutdown::Write).into()
+ }
+}
+
+impl AsRef<UnixStream> for ReadHalf<'_> {
+ fn as_ref(&self) -> &UnixStream {
+ self.0
+ }
+}
+
+impl AsRef<UnixStream> for WriteHalf<'_> {
+ fn as_ref(&self) -> &UnixStream {
+ self.0
+ }
+}
diff --git a/third_party/rust/tokio/src/net/unix/stream.rs b/third_party/rust/tokio/src/net/unix/stream.rs
new file mode 100644
index 0000000000..beae699962
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/stream.rs
@@ -0,0 +1,233 @@
+use crate::future::poll_fn;
+use crate::io::{AsyncRead, AsyncWrite, PollEvented};
+use crate::net::unix::split::{split, ReadHalf, WriteHalf};
+use crate::net::unix::ucred::{self, UCred};
+
+use std::convert::TryFrom;
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::mem::MaybeUninit;
+use std::net::Shutdown;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+cfg_uds! {
+ /// A structure representing a connected Unix socket.
+ ///
+ /// This socket can be connected directly with `UnixStream::connect` or accepted
+ /// from a listener with `UnixListener::incoming`. Additionally, a pair of
+ /// anonymous Unix sockets can be created with `UnixStream::pair`.
+ pub struct UnixStream {
+ io: PollEvented<mio_uds::UnixStream>,
+ }
+}
+
+impl UnixStream {
+ /// Connects to the socket named by `path`.
+ ///
+ /// This function will create a new Unix socket and connect to the path
+ /// specified, associating the returned stream with the default event loop's
+ /// handle.
+ pub async fn connect<P>(path: P) -> io::Result<UnixStream>
+ where
+ P: AsRef<Path>,
+ {
+ let stream = mio_uds::UnixStream::connect(path)?;
+ let stream = UnixStream::new(stream)?;
+
+ poll_fn(|cx| stream.io.poll_write_ready(cx)).await?;
+ Ok(stream)
+ }
+
+ /// Consumes a `UnixStream` in the standard library and returns a
+ /// nonblocking `UnixStream` from this crate.
+ ///
+ /// The returned stream will be associated with the given event loop
+ /// specified by `handle` and is ready to perform I/O.
+ ///
+ /// # Panics
+ ///
+ /// This function panics if thread-local runtime is not set.
+ ///
+ /// The runtime is usually set implicitly when this function is called
+ /// from a future driven by a tokio runtime, otherwise runtime can be set
+ /// explicitly with [`Handle::enter`](crate::runtime::Handle::enter) function.
+ pub fn from_std(stream: net::UnixStream) -> io::Result<UnixStream> {
+ let stream = mio_uds::UnixStream::from_stream(stream)?;
+ let io = PollEvented::new(stream)?;
+
+ Ok(UnixStream { io })
+ }
+
+ /// Creates an unnamed pair of connected sockets.
+ ///
+ /// This function will create a pair of interconnected Unix sockets for
+ /// communicating back and forth between one another. Each socket will
+ /// be associated with the default event loop's handle.
+ pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
+ let (a, b) = mio_uds::UnixStream::pair()?;
+ let a = UnixStream::new(a)?;
+ let b = UnixStream::new(b)?;
+
+ Ok((a, b))
+ }
+
+ pub(crate) fn new(stream: mio_uds::UnixStream) -> io::Result<UnixStream> {
+ let io = PollEvented::new(stream)?;
+ Ok(UnixStream { io })
+ }
+
+ /// Returns the socket address of the local half of this connection.
+ pub fn local_addr(&self) -> io::Result<SocketAddr> {
+ self.io.get_ref().local_addr()
+ }
+
+ /// Returns the socket address of the remote half of this connection.
+ pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+ self.io.get_ref().peer_addr()
+ }
+
+ /// Returns effective credentials of the process which called `connect` or `pair`.
+ pub fn peer_cred(&self) -> io::Result<UCred> {
+ ucred::get_peer_cred(self)
+ }
+
+ /// Returns the value of the `SO_ERROR` option.
+ pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+ self.io.get_ref().take_error()
+ }
+
+ /// Shuts down the read, write, or both halves of this connection.
+ ///
+ /// This function will cause all pending and future I/O calls on the
+ /// specified portions to immediately return with an appropriate value
+ /// (see the documentation of `Shutdown`).
+ pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+ self.io.get_ref().shutdown(how)
+ }
+
+ /// Split a `UnixStream` into a read half and a write half, which can be used
+ /// to read and write the stream concurrently.
+ pub fn split(&mut self) -> (ReadHalf<'_>, WriteHalf<'_>) {
+ split(self)
+ }
+}
+
+impl TryFrom<UnixStream> for mio_uds::UnixStream {
+ type Error = io::Error;
+
+ /// Consumes value, returning the mio I/O object.
+ ///
+ /// See [`PollEvented::into_inner`] for more details about
+ /// resource deregistration that happens during the call.
+ ///
+ /// [`PollEvented::into_inner`]: crate::io::PollEvented::into_inner
+ fn try_from(value: UnixStream) -> Result<Self, Self::Error> {
+ value.io.into_inner()
+ }
+}
+
+impl TryFrom<net::UnixStream> for UnixStream {
+ type Error = io::Error;
+
+ /// Consumes stream, returning the tokio I/O object.
+ ///
+ /// This is equivalent to
+ /// [`UnixStream::from_std(stream)`](UnixStream::from_std).
+ fn try_from(stream: net::UnixStream) -> io::Result<Self> {
+ Self::from_std(stream)
+ }
+}
+
+impl AsyncRead for UnixStream {
+ unsafe fn prepare_uninitialized_buffer(&self, _: &mut [MaybeUninit<u8>]) -> bool {
+ false
+ }
+
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<io::Result<usize>> {
+ self.poll_read_priv(cx, buf)
+ }
+}
+
+impl AsyncWrite for UnixStream {
+ fn poll_write(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<io::Result<usize>> {
+ self.poll_write_priv(cx, buf)
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+
+ fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+ self.shutdown(std::net::Shutdown::Write)?;
+ Poll::Ready(Ok(()))
+ }
+}
+
+impl UnixStream {
+ // == Poll IO functions that takes `&self` ==
+ //
+ // They are not public because (taken from the doc of `PollEvented`):
+ //
+ // While `PollEvented` is `Sync` (if the underlying I/O type is `Sync`), the
+ // caller must ensure that there are at most two tasks that use a
+ // `PollEvented` instance concurrently. One for reading and one for writing.
+ // While violating this requirement is "safe" from a Rust memory model point
+ // of view, it will result in unexpected behavior in the form of lost
+ // notifications and tasks hanging.
+
+ pub(crate) fn poll_read_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<io::Result<usize>> {
+ ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
+
+ match self.io.get_ref().read(buf) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_read_ready(cx, mio::Ready::readable())?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+
+ pub(crate) fn poll_write_priv(
+ &self,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<io::Result<usize>> {
+ ready!(self.io.poll_write_ready(cx))?;
+
+ match self.io.get_ref().write(buf) {
+ Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+ self.io.clear_write_ready(cx)?;
+ Poll::Pending
+ }
+ x => Poll::Ready(x),
+ }
+ }
+}
+
+impl fmt::Debug for UnixStream {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ self.io.get_ref().fmt(f)
+ }
+}
+
+impl AsRawFd for UnixStream {
+ fn as_raw_fd(&self) -> RawFd {
+ self.io.get_ref().as_raw_fd()
+ }
+}
diff --git a/third_party/rust/tokio/src/net/unix/ucred.rs b/third_party/rust/tokio/src/net/unix/ucred.rs
new file mode 100644
index 0000000000..cdd77ea414
--- /dev/null
+++ b/third_party/rust/tokio/src/net/unix/ucred.rs
@@ -0,0 +1,151 @@
+use libc::{gid_t, uid_t};
+
+/// Credentials of a process
+#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]
+pub struct UCred {
+ /// UID (user ID) of the process
+ pub uid: uid_t,
+ /// GID (group ID) of the process
+ pub gid: gid_t,
+}
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub(crate) use self::impl_linux::get_peer_cred;
+
+#[cfg(any(
+ target_os = "dragonfly",
+ target_os = "macos",
+ target_os = "ios",
+ target_os = "freebsd",
+ target_os = "netbsd",
+ target_os = "openbsd"
+))]
+pub(crate) use self::impl_macos::get_peer_cred;
+
+#[cfg(any(target_os = "solaris"))]
+pub(crate) use self::impl_solaris::get_peer_cred;
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+pub(crate) mod impl_linux {
+ use crate::net::unix::UnixStream;
+
+ use libc::{c_void, getsockopt, socklen_t, SOL_SOCKET, SO_PEERCRED};
+ use std::{io, mem};
+
+ use libc::ucred;
+
+ pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
+ use std::os::unix::io::AsRawFd;
+
+ unsafe {
+ let raw_fd = sock.as_raw_fd();
+
+ let mut ucred = ucred {
+ pid: 0,
+ uid: 0,
+ gid: 0,
+ };
+
+ let ucred_size = mem::size_of::<ucred>();
+
+ // These paranoid checks should be optimized-out
+ assert!(mem::size_of::<u32>() <= mem::size_of::<usize>());
+ assert!(ucred_size <= u32::max_value() as usize);
+
+ let mut ucred_size = ucred_size as socklen_t;
+
+ let ret = getsockopt(
+ raw_fd,
+ SOL_SOCKET,
+ SO_PEERCRED,
+ &mut ucred as *mut ucred as *mut c_void,
+ &mut ucred_size,
+ );
+ if ret == 0 && ucred_size as usize == mem::size_of::<ucred>() {
+ Ok(super::UCred {
+ uid: ucred.uid,
+ gid: ucred.gid,
+ })
+ } else {
+ Err(io::Error::last_os_error())
+ }
+ }
+ }
+}
+
+#[cfg(any(
+ target_os = "dragonfly",
+ target_os = "macos",
+ target_os = "ios",
+ target_os = "freebsd",
+ target_os = "netbsd",
+ target_os = "openbsd"
+))]
+pub(crate) mod impl_macos {
+ use crate::net::unix::UnixStream;
+
+ use libc::getpeereid;
+ use std::io;
+ use std::mem::MaybeUninit;
+ use std::os::unix::io::AsRawFd;
+
+ pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
+ unsafe {
+ let raw_fd = sock.as_raw_fd();
+
+ let mut uid = MaybeUninit::uninit();
+ let mut gid = MaybeUninit::uninit();
+
+ let ret = getpeereid(raw_fd, uid.as_mut_ptr(), gid.as_mut_ptr());
+
+ if ret == 0 {
+ Ok(super::UCred {
+ uid: uid.assume_init(),
+ gid: gid.assume_init(),
+ })
+ } else {
+ Err(io::Error::last_os_error())
+ }
+ }
+ }
+}
+
+#[cfg(any(target_os = "solaris"))]
+pub(crate) mod impl_solaris {
+ use crate::net::unix::UnixStream;
+ use std::io;
+ use std::os::unix::io::AsRawFd;
+ use std::ptr;
+
+ #[allow(non_camel_case_types)]
+ enum ucred_t {}
+
+ extern "C" {
+ fn ucred_free(cred: *mut ucred_t);
+ fn ucred_geteuid(cred: *const ucred_t) -> super::uid_t;
+ fn ucred_getegid(cred: *const ucred_t) -> super::gid_t;
+
+ fn getpeerucred(fd: std::os::raw::c_int, cred: *mut *mut ucred_t) -> std::os::raw::c_int;
+ }
+
+ pub(crate) fn get_peer_cred(sock: &UnixStream) -> io::Result<super::UCred> {
+ unsafe {
+ let raw_fd = sock.as_raw_fd();
+
+ let mut cred = ptr::null_mut::<*mut ucred_t>() as *mut ucred_t;
+
+ let ret = getpeerucred(raw_fd, &mut cred);
+
+ if ret == 0 {
+ let uid = ucred_geteuid(cred);
+ let gid = ucred_getegid(cred);
+
+ ucred_free(cred);
+
+ Ok(super::UCred { uid, gid })
+ } else {
+ Err(io::Error::last_os_error())
+ }
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-27 11:50:22 +0000