mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-09-07 00:00:08 +02:00
uds: move into tokio-net (#1462)
This commit is contained in:
Carl Lerche
@@ -45,3 +45,7 @@ pub mod tcp;
|
||||
|
||||
#[cfg(feature = "udp")]
|
||||
pub mod udp;
|
||||
|
||||
#[cfg(feature = "uds")]
|
||||
#[cfg(unix)]
|
||||
pub mod uds;
|
||||
|
||||
@@ -0,0 +1,233 @@
|
||||
use crate::driver::Handle;
|
||||
use crate::util::PollEvented;
|
||||
|
||||
use futures_core::ready;
|
||||
use futures_util::future::poll_fn;
|
||||
use mio_uds;
|
||||
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};
|
||||
|
||||
/// 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)?;
|
||||
Ok(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.
|
||||
pub fn from_std(datagram: net::UnixDatagram, handle: &Handle) -> io::Result<UnixDatagram> {
|
||||
let socket = mio_uds::UnixDatagram::from_datagram(datagram)?;
|
||||
let io = PollEvented::new_with_handle(socket, handle)?;
|
||||
Ok(UnixDatagram { io })
|
||||
}
|
||||
|
||||
fn new(socket: mio_uds::UnixDatagram) -> UnixDatagram {
|
||||
let io = PollEvented::new(socket);
|
||||
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()?;
|
||||
Ok(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()
|
||||
}
|
||||
|
||||
/// Shut 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 [`tokio_net::util::PollEvented::into_inner`] for more details about
|
||||
/// resource deregistration that happens during the call.
|
||||
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, &Handle::default())`](UnixDatagram::from_std).
|
||||
fn try_from(stream: net::UnixDatagram) -> Result<Self, Self::Error> {
|
||||
Self::from_std(stream, &Handle::default())
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
use super::UnixDatagram;
|
||||
use bytes::{BufMut, BytesMut};
|
||||
use futures::{try_ready, Async, AsyncSink, Poll, Sink, StartSend, Stream};
|
||||
use log::trace;
|
||||
use std::io;
|
||||
use std::os::unix::net::SocketAddr;
|
||||
use std::path::Path;
|
||||
use tokio_codec::{Decoder, Encoder};
|
||||
|
||||
/// A unified `Stream` and `Sink` interface to an underlying `UnixDatagram`, using
|
||||
/// the `Encoder` and `Decoder` traits to encode and decode frames.
|
||||
///
|
||||
/// Unix datagram sockets work with datagrams, but higher-level code may wants to
|
||||
/// batch these into meaningful chunks, called "frames". This method layers
|
||||
/// framing on top of this socket by using the `Encoder` and `Decoder` traits to
|
||||
/// handle encoding and decoding of messages frames. Note that the incoming and
|
||||
/// outgoing frame types may be distinct.
|
||||
///
|
||||
/// This function returns a *single* object that is both `Stream` and `Sink`;
|
||||
/// grouping this into a single object is often useful for layering things which
|
||||
/// require both read and write access to the underlying object.
|
||||
///
|
||||
/// If you want to work more directly with the streams and sink, consider
|
||||
/// calling `split` on the `UnixDatagramFramed` returned by this method, which will break
|
||||
/// them into separate objects, allowing them to interact more easily.
|
||||
#[must_use = "sinks do nothing unless polled"]
|
||||
#[derive(Debug)]
|
||||
pub struct UnixDatagramFramed<A, C> {
|
||||
socket: UnixDatagram,
|
||||
codec: C,
|
||||
rd: BytesMut,
|
||||
wr: BytesMut,
|
||||
out_addr: Option<A>,
|
||||
flushed: bool,
|
||||
}
|
||||
|
||||
impl<A, C: Decoder> Stream for UnixDatagramFramed<A, C> {
|
||||
type Item = (C::Item, SocketAddr);
|
||||
type Error = C::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
|
||||
self.rd.reserve(INITIAL_RD_CAPACITY);
|
||||
|
||||
let (n, addr) = unsafe {
|
||||
let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut()));
|
||||
self.rd.advance_mut(n);
|
||||
(n, addr)
|
||||
};
|
||||
trace!("received {} bytes, decoding", n);
|
||||
let frame_res = self.codec.decode(&mut self.rd);
|
||||
self.rd.clear();
|
||||
let frame = frame_res?;
|
||||
let result = frame.map(|frame| (frame, addr));
|
||||
trace!("frame decoded from buffer");
|
||||
Ok(Async::Ready(result))
|
||||
}
|
||||
}
|
||||
|
||||
impl<A: AsRef<Path>, C: Encoder> Sink for UnixDatagramFramed<A, C> {
|
||||
type SinkItem = (C::Item, A);
|
||||
type SinkError = C::Error;
|
||||
|
||||
fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
|
||||
trace!("sending frame");
|
||||
|
||||
if !self.flushed {
|
||||
match self.poll_complete()? {
|
||||
Async::Ready(()) => {}
|
||||
Async::NotReady => return Ok(AsyncSink::NotReady(item)),
|
||||
}
|
||||
}
|
||||
|
||||
let (frame, out_addr) = item;
|
||||
self.codec.encode(frame, &mut self.wr)?;
|
||||
self.out_addr = Some(out_addr);
|
||||
self.flushed = false;
|
||||
trace!("frame encoded; length={}", self.wr.len());
|
||||
|
||||
Ok(AsyncSink::Ready)
|
||||
}
|
||||
|
||||
fn poll_complete(&mut self) -> Poll<(), C::Error> {
|
||||
if self.flushed {
|
||||
return Ok(Async::Ready(()));
|
||||
}
|
||||
|
||||
let n = {
|
||||
let out_path = match self.out_addr {
|
||||
Some(ref out_path) => out_path.as_ref(),
|
||||
None => {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"internal error: addr not available while data not flushed",
|
||||
)
|
||||
.into());
|
||||
}
|
||||
};
|
||||
|
||||
trace!("flushing frame; length={}", self.wr.len());
|
||||
try_ready!(self.socket.poll_send_to(&self.wr, out_path))
|
||||
};
|
||||
|
||||
trace!("written {}", n);
|
||||
|
||||
let wrote_all = n == self.wr.len();
|
||||
self.wr.clear();
|
||||
self.flushed = true;
|
||||
|
||||
if wrote_all {
|
||||
self.out_addr = None;
|
||||
Ok(Async::Ready(()))
|
||||
} else {
|
||||
Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"failed to write entire datagram to socket",
|
||||
)
|
||||
.into())
|
||||
}
|
||||
}
|
||||
|
||||
fn close(&mut self) -> Poll<(), C::Error> {
|
||||
self.poll_complete()
|
||||
}
|
||||
}
|
||||
|
||||
const INITIAL_RD_CAPACITY: usize = 64 * 1024;
|
||||
const INITIAL_WR_CAPACITY: usize = 8 * 1024;
|
||||
|
||||
impl<A, C> UnixDatagramFramed<A, C> {
|
||||
/// Create a new `UnixDatagramFramed` backed by the given socket and codec.
|
||||
///
|
||||
/// See struct level documentation for more details.
|
||||
pub fn new(socket: UnixDatagram, codec: C) -> UnixDatagramFramed<A, C> {
|
||||
UnixDatagramFramed {
|
||||
socket: socket,
|
||||
codec: codec,
|
||||
out_addr: None,
|
||||
rd: BytesMut::with_capacity(INITIAL_RD_CAPACITY),
|
||||
wr: BytesMut::with_capacity(INITIAL_WR_CAPACITY),
|
||||
flushed: true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a reference to the underlying I/O stream wrapped by `Framed`.
|
||||
///
|
||||
/// # Note
|
||||
///
|
||||
/// Care should be taken to not tamper with the underlying stream of data
|
||||
/// coming in as it may corrupt the stream of frames otherwise being worked
|
||||
/// with.
|
||||
pub fn get_ref(&self) -> &UnixDatagram {
|
||||
&self.socket
|
||||
}
|
||||
|
||||
/// Returns a mutable reference to the underlying I/O stream wrapped by
|
||||
/// `Framed`.
|
||||
///
|
||||
/// # Note
|
||||
///
|
||||
/// Care should be taken to not tamper with the underlying stream of data
|
||||
/// coming in as it may corrupt the stream of frames otherwise being worked
|
||||
/// with.
|
||||
pub fn get_mut(&mut self) -> &mut UnixDatagram {
|
||||
&mut self.socket
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
#![cfg(feature = "async-traits")]
|
||||
|
||||
use super::{UnixListener, UnixStream};
|
||||
|
||||
use futures_core::ready;
|
||||
use futures_core::stream::Stream;
|
||||
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 {
|
||||
inner: UnixListener,
|
||||
}
|
||||
|
||||
impl Incoming {
|
||||
pub(crate) fn new(listener: UnixListener) -> Incoming {
|
||||
Incoming { inner: listener }
|
||||
}
|
||||
}
|
||||
|
||||
impl 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!(Pin::new(&mut self.inner).poll_accept(cx))?;
|
||||
Poll::Ready(Some(Ok(socket)))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
use super::UnixStream;
|
||||
use crate::driver::Handle;
|
||||
use crate::util::PollEvented;
|
||||
|
||||
use futures_core::ready;
|
||||
use futures_util::future::poll_fn;
|
||||
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};
|
||||
|
||||
/// A Unix socket which can accept connections from other Unix sockets.
|
||||
pub struct UnixListener {
|
||||
io: PollEvented<mio_uds::UnixListener>,
|
||||
}
|
||||
|
||||
impl UnixListener {
|
||||
/// Creates a new `UnixListener` bound to the specified path.
|
||||
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.
|
||||
pub fn from_std(listener: net::UnixListener, handle: &Handle) -> io::Result<UnixListener> {
|
||||
let listener = mio_uds::UnixListener::from_listener(listener)?;
|
||||
let io = PollEvented::new_with_handle(listener, handle)?;
|
||||
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(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Consumes this listener, returning a stream of the sockets this listener
|
||||
/// accepts.
|
||||
///
|
||||
/// This method returns an implementation of the `Stream` trait which
|
||||
/// resolves to the sockets the are accepted on this listener.
|
||||
#[cfg(feature = "async-traits")]
|
||||
pub fn incoming(self) -> super::Incoming {
|
||||
super::Incoming::new(self)
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<UnixListener> for mio_uds::UnixListener {
|
||||
type Error = io::Error;
|
||||
|
||||
/// Consumes value, returning the mio I/O object.
|
||||
///
|
||||
/// See [`tokio_net::util::PollEvented::into_inner`] for more details about
|
||||
/// resource deregistration that happens during the call.
|
||||
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, &Handle::default())`](UnixListener::from_std).
|
||||
fn try_from(stream: net::UnixListener) -> io::Result<Self> {
|
||||
Self::from_std(stream, &Handle::default())
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
//! Unix Domain Sockets for Tokio.
|
||||
//!
|
||||
//! This crate provides APIs for using Unix Domain Sockets with Tokio.
|
||||
|
||||
mod datagram;
|
||||
// mod frame;
|
||||
mod incoming;
|
||||
mod listener;
|
||||
pub mod split;
|
||||
mod stream;
|
||||
mod ucred;
|
||||
|
||||
pub use self::datagram::UnixDatagram;
|
||||
#[cfg(feature = "async-traits")]
|
||||
pub use self::incoming::Incoming;
|
||||
pub use self::listener::UnixListener;
|
||||
pub use self::stream::UnixStream;
|
||||
pub use self::ucred::UCred;
|
||||
@@ -0,0 +1,182 @@
|
||||
//! `UnixStream` split support.
|
||||
//!
|
||||
//! A `UnixStream` can be split into a read half and a write half with `UnixStream::split`
|
||||
//! and `UnixStream::split_mut` methods. The read half implements `AsyncRead` while
|
||||
//! the write half implements `AsyncWrite`. The two halves can be used concurrently.
|
||||
//!
|
||||
//! Compared to the generic split of `AsyncRead + AsyncWrite`, this specialized
|
||||
//! split gives read and write halves that are faster and smaller, because they
|
||||
//! do not use locks. They also provide access to the underlying `UnixStream`
|
||||
//! after split, implementing `AsRef<UnixStream>`. This allows you to call
|
||||
//! `UnixStream` methods that takes `&self`, e.g., to get local and peer
|
||||
//! addresses, to get and set socket options, and to shutdown the sockets.
|
||||
|
||||
use super::UnixStream;
|
||||
|
||||
use tokio_io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use bytes::{Buf, BufMut};
|
||||
use std::io;
|
||||
use std::net::Shutdown;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
/// Read half of a `UnixStream`.
|
||||
#[derive(Debug)]
|
||||
pub struct UnixStreamReadHalf(Arc<UnixStream>);
|
||||
|
||||
/// Write half of a `UnixStream`.
|
||||
///
|
||||
/// Note that in the `AsyncWrite` implementation of `UnixStreamWriteHalf`,
|
||||
/// `poll_shutdown` actually shuts down the stream in the write direction.
|
||||
#[derive(Debug)]
|
||||
pub struct UnixStreamWriteHalf(Arc<UnixStream>);
|
||||
|
||||
/// Read half of a `UnixStream`.
|
||||
#[derive(Debug)]
|
||||
pub struct UnixStreamReadHalfMut<'a>(&'a UnixStream);
|
||||
|
||||
/// Write half of a `UnixStream`.
|
||||
///
|
||||
/// Note that in the `AsyncWrite` implementation of `UnixStreamWriteHalfMut`,
|
||||
/// `poll_shutdown` actually shuts down the stream in the write direction.
|
||||
#[derive(Debug)]
|
||||
pub struct UnixStreamWriteHalfMut<'a>(&'a UnixStream);
|
||||
|
||||
pub(crate) fn split(stream: UnixStream) -> (UnixStreamReadHalf, UnixStreamWriteHalf) {
|
||||
let shared = Arc::new(stream);
|
||||
(
|
||||
UnixStreamReadHalf(shared.clone()),
|
||||
UnixStreamWriteHalf(shared),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn split_mut(
|
||||
stream: &mut UnixStream,
|
||||
) -> (UnixStreamReadHalfMut<'_>, UnixStreamWriteHalfMut<'_>) {
|
||||
(
|
||||
UnixStreamReadHalfMut(stream),
|
||||
UnixStreamWriteHalfMut(stream),
|
||||
)
|
||||
}
|
||||
|
||||
impl AsRef<UnixStream> for UnixStreamReadHalf {
|
||||
fn as_ref(&self) -> &UnixStream {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRef<UnixStream> for UnixStreamWriteHalf {
|
||||
fn as_ref(&self) -> &UnixStream {
|
||||
&self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRef<UnixStream> for UnixStreamReadHalfMut<'_> {
|
||||
fn as_ref(&self) -> &UnixStream {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRef<UnixStream> for UnixStreamWriteHalfMut<'_> {
|
||||
fn as_ref(&self) -> &UnixStream {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for UnixStreamReadHalf {
|
||||
unsafe fn prepare_uninitialized_buffer(&self, _: &mut [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)
|
||||
}
|
||||
|
||||
fn poll_read_buf<B: BufMut>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.0.poll_read_buf_priv(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for UnixStreamReadHalfMut<'_> {
|
||||
unsafe fn prepare_uninitialized_buffer(&self, _: &mut [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)
|
||||
}
|
||||
|
||||
fn poll_read_buf<B: BufMut>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.0.poll_read_buf_priv(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncWrite for UnixStreamWriteHalf {
|
||||
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()
|
||||
}
|
||||
|
||||
fn poll_write_buf<B: Buf>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.0.poll_write_buf_priv(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncWrite for UnixStreamWriteHalfMut<'_> {
|
||||
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()
|
||||
}
|
||||
|
||||
fn poll_write_buf<B: Buf>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.0.poll_write_buf_priv(cx, buf)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,350 @@
|
||||
use super::split::{
|
||||
split, split_mut, UnixStreamReadHalf, UnixStreamReadHalfMut, UnixStreamWriteHalf,
|
||||
UnixStreamWriteHalfMut,
|
||||
};
|
||||
use super::ucred::{self, UCred};
|
||||
use crate::driver::Handle;
|
||||
use crate::util::PollEvented;
|
||||
|
||||
use tokio_io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use bytes::{Buf, BufMut};
|
||||
use futures_core::ready;
|
||||
use futures_util::future::poll_fn;
|
||||
use iovec::IoVec;
|
||||
use std::convert::TryFrom;
|
||||
use std::fmt;
|
||||
use std::io::{self, Read, Write};
|
||||
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};
|
||||
|
||||
/// 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.
|
||||
pub fn from_std(stream: net::UnixStream, handle: &Handle) -> io::Result<UnixStream> {
|
||||
let stream = mio_uds::UnixStream::from_stream(stream)?;
|
||||
let io = PollEvented::new_with_handle(stream, handle)?;
|
||||
|
||||
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) -> UnixStream {
|
||||
let io = PollEvented::new(stream);
|
||||
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.
|
||||
///
|
||||
/// See the module level documenation of [`split`](super::split) for more
|
||||
/// details.
|
||||
pub fn split(self) -> (UnixStreamReadHalf, UnixStreamWriteHalf) {
|
||||
split(self)
|
||||
}
|
||||
|
||||
/// Split a `UnixStream` into a read half and a write half, which can be used
|
||||
/// to read and write the stream concurrently.
|
||||
///
|
||||
/// See the module level documenation of [`split`](super::split) for more
|
||||
/// details.
|
||||
pub fn split_mut(&mut self) -> (UnixStreamReadHalfMut<'_>, UnixStreamWriteHalfMut<'_>) {
|
||||
split_mut(self)
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<UnixStream> for mio_uds::UnixStream {
|
||||
type Error = io::Error;
|
||||
|
||||
/// Consumes value, returning the mio I/O object.
|
||||
///
|
||||
/// See [`tokio_net::util::PollEvented::into_inner`] for more details about
|
||||
/// resource deregistration that happens during the call.
|
||||
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, &Handle::default())`](UnixStream::from_std).
|
||||
fn try_from(stream: net::UnixStream) -> io::Result<Self> {
|
||||
Self::from_std(stream, &Handle::default())
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for UnixStream {
|
||||
unsafe fn prepare_uninitialized_buffer(&self, _: &mut [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)
|
||||
}
|
||||
|
||||
fn poll_read_buf<B: BufMut>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.poll_read_buf_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<()>> {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
|
||||
fn poll_write_buf<B: Buf>(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
self.poll_write_buf_priv(cx, buf)
|
||||
}
|
||||
}
|
||||
|
||||
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_read_buf_priv<B: BufMut>(
|
||||
&self,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
|
||||
|
||||
let r = unsafe {
|
||||
// The `IoVec` type can't have a 0-length size, so we create a bunch
|
||||
// of dummy versions on the stack with 1 length which we'll quickly
|
||||
// overwrite.
|
||||
let b1: &mut [u8] = &mut [0];
|
||||
let b2: &mut [u8] = &mut [0];
|
||||
let b3: &mut [u8] = &mut [0];
|
||||
let b4: &mut [u8] = &mut [0];
|
||||
let b5: &mut [u8] = &mut [0];
|
||||
let b6: &mut [u8] = &mut [0];
|
||||
let b7: &mut [u8] = &mut [0];
|
||||
let b8: &mut [u8] = &mut [0];
|
||||
let b9: &mut [u8] = &mut [0];
|
||||
let b10: &mut [u8] = &mut [0];
|
||||
let b11: &mut [u8] = &mut [0];
|
||||
let b12: &mut [u8] = &mut [0];
|
||||
let b13: &mut [u8] = &mut [0];
|
||||
let b14: &mut [u8] = &mut [0];
|
||||
let b15: &mut [u8] = &mut [0];
|
||||
let b16: &mut [u8] = &mut [0];
|
||||
let mut bufs: [&mut IoVec; 16] = [
|
||||
b1.into(),
|
||||
b2.into(),
|
||||
b3.into(),
|
||||
b4.into(),
|
||||
b5.into(),
|
||||
b6.into(),
|
||||
b7.into(),
|
||||
b8.into(),
|
||||
b9.into(),
|
||||
b10.into(),
|
||||
b11.into(),
|
||||
b12.into(),
|
||||
b13.into(),
|
||||
b14.into(),
|
||||
b15.into(),
|
||||
b16.into(),
|
||||
];
|
||||
let n = buf.bytes_vec_mut(&mut bufs);
|
||||
self.io.get_ref().read_bufs(&mut bufs[..n])
|
||||
};
|
||||
|
||||
match r {
|
||||
Ok(n) => {
|
||||
unsafe {
|
||||
buf.advance_mut(n);
|
||||
}
|
||||
Poll::Ready(Ok(n))
|
||||
}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
|
||||
self.io.clear_read_ready(cx, mio::Ready::readable())?;
|
||||
Poll::Pending
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
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),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn poll_write_buf_priv<B: Buf>(
|
||||
&self,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut B,
|
||||
) -> Poll<io::Result<usize>> {
|
||||
ready!(self.io.poll_write_ready(cx))?;
|
||||
|
||||
let r = {
|
||||
// The `IoVec` type can't have a zero-length size, so create a dummy
|
||||
// version from a 1-length slice which we'll overwrite with the
|
||||
// `bytes_vec` method.
|
||||
static DUMMY: &[u8] = &[0];
|
||||
let iovec = <&IoVec>::from(DUMMY);
|
||||
let mut bufs = [iovec; 64];
|
||||
let n = buf.bytes_vec(&mut bufs);
|
||||
self.io.get_ref().write_bufs(&bufs[..n])
|
||||
};
|
||||
match r {
|
||||
Ok(n) => {
|
||||
buf.advance(n);
|
||||
Poll::Ready(Ok(n))
|
||||
}
|
||||
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
|
||||
self.io.clear_write_ready(cx)?;
|
||||
Poll::Pending
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
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::uds::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::uds::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 std::io;
|
||||
use std::os::unix::io::AsRawFd;
|
||||
use std::ptr;
|
||||
use UnixStream;
|
||||
|
||||
#[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())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Note that LOCAL_PEERCRED is not supported on DragonFly (yet). So do not run tests.
|
||||
#[cfg(not(target_os = "dragonfly"))]
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use crate::uds::UnixStream;
|
||||
|
||||
use libc::getegid;
|
||||
use libc::geteuid;
|
||||
|
||||
#[test]
|
||||
#[cfg_attr(
|
||||
target_os = "freebsd",
|
||||
ignore = "Requires FreeBSD 12.0 or later. https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=176419"
|
||||
)]
|
||||
#[cfg_attr(
|
||||
target_os = "netbsd",
|
||||
ignore = "NetBSD does not support getpeereid() for sockets created by socketpair()"
|
||||
)]
|
||||
fn test_socket_pair() {
|
||||
let (a, b) = UnixStream::pair().unwrap();
|
||||
let cred_a = a.peer_cred().unwrap();
|
||||
let cred_b = b.peer_cred().unwrap();
|
||||
assert_eq!(cred_a, cred_b);
|
||||
|
||||
let uid = unsafe { geteuid() };
|
||||
let gid = unsafe { getegid() };
|
||||
|
||||
assert_eq!(cred_a.uid, uid);
|
||||
assert_eq!(cred_a.gid, gid);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user