diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 331dbaca7..18df1638f 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -12,6 +12,24 @@ path = "src/helloworld/server.rs" name = "helloworld-client" path = "src/helloworld/client.rs" +[[bin]] +name = "socket-activation-tcp-server" +path = "src/socket_activation/tcp_server.rs" + +[[bin]] +name = "socket-activation-tcp-client" +path = "src/socket_activation/tcp_client.rs" + +[[bin]] +name = "socket-activation-uds-server" +path = "src/socket_activation/uds_server.rs" +required-features = ["uds"] + +[[bin]] +name = "socket-activation-uds-client" +path = "src/socket_activation/uds_client.rs" +required-features = ["uds"] + [[bin]] name = "blocking-server" path = "src/blocking/server.rs" diff --git a/examples/README.md b/examples/README.md index a1efcc9dc..688a074f2 100644 --- a/examples/README.md +++ b/examples/README.md @@ -233,6 +233,23 @@ The autoload example requires the following crates installed globally: * [systemfd](https://crates.io/crates/systemfd) * [cargo-watch](https://crates.io/crates/cargo-watch) +## Socket Activation (systemd) + +The servers adopt a listening socket passed by a socket-activation manager (via +`LISTEN_FDS`/`LISTEN_PID`), or bind directly otherwise. + +### TCP + +```bash +$ ./src/socket_activation/test_tcp.sh +``` + +### Unix domain socket + +```bash +$ ./src/socket_activation/test_uds.sh +``` + ## Richer Error Both clients and both servers do the same thing, but using the two different diff --git a/examples/src/socket_activation/systemd/socketact-tcp.service b/examples/src/socket_activation/systemd/socketact-tcp.service new file mode 100644 index 000000000..666394089 --- /dev/null +++ b/examples/src/socket_activation/systemd/socketact-tcp.service @@ -0,0 +1,11 @@ +# Service unit for the tonic TCP socket-activation example. +# +# Replace ExecStart with the absolute path to your built binary, e.g. +# /path/to/target/debug/socket-activation-tcp-server +[Unit] +Description=tonic socket-activation TCP example service +Requires=socketact-tcp.socket +After=socketact-tcp.socket + +[Service] +ExecStart=/path/to/target/debug/socket-activation-tcp-server diff --git a/examples/src/socket_activation/systemd/socketact-tcp.socket b/examples/src/socket_activation/systemd/socketact-tcp.socket new file mode 100644 index 000000000..a6bcdb0d2 --- /dev/null +++ b/examples/src/socket_activation/systemd/socketact-tcp.socket @@ -0,0 +1,9 @@ +# Socket unit for the tonic TCP socket-activation example. +[Unit] +Description=tonic socket-activation TCP example socket + +[Socket] +ListenStream=[::1]:50051 + +[Install] +WantedBy=sockets.target diff --git a/examples/src/socket_activation/systemd/socketact-uds.service b/examples/src/socket_activation/systemd/socketact-uds.service new file mode 100644 index 000000000..91f2b8cf5 --- /dev/null +++ b/examples/src/socket_activation/systemd/socketact-uds.service @@ -0,0 +1,13 @@ +# Service unit for the tonic socket-activation UDS example. +# +# Replace ExecStart with the absolute path to your built binary, e.g. +# /path/to/target/debug/socket-activation-uds-server +[Unit] +Description=tonic socket-activation UDS example service +Requires=socketact-uds.socket +After=socketact-uds.socket + +[Service] +ExecStart=/path/to/target/debug/socket-activation-uds-server +# Must match ListenStream in socketact-uds.socket. +Environment=TONIC_UDS_PATH=%t/tonic-sockact.sock diff --git a/examples/src/socket_activation/systemd/socketact-uds.socket b/examples/src/socket_activation/systemd/socketact-uds.socket new file mode 100644 index 000000000..d5cd3bc94 --- /dev/null +++ b/examples/src/socket_activation/systemd/socketact-uds.socket @@ -0,0 +1,9 @@ +# Socket unit for the tonic socket-activation UDS example. +[Unit] +Description=tonic socket-activation UDS example socket + +[Socket] +ListenStream=%t/tonic-sockact.sock + +[Install] +WantedBy=sockets.target diff --git a/examples/src/socket_activation/tcp_client.rs b/examples/src/socket_activation/tcp_client.rs new file mode 100644 index 000000000..d57708ea8 --- /dev/null +++ b/examples/src/socket_activation/tcp_client.rs @@ -0,0 +1,23 @@ +//! Client for the socket-activation example, connecting over a TCP socket. + +use hello_world::HelloRequest; +use hello_world::greeter_client::GreeterClient; + +pub mod hello_world { + tonic::include_proto!("helloworld"); +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let mut client = GreeterClient::connect("http://[::1]:50051").await?; + + let request = tonic::Request::new(HelloRequest { + name: "Tonic".into(), + }); + + let response = client.say_hello(request).await?; + + println!("RESPONSE={response:?}"); + + Ok(()) +} diff --git a/examples/src/socket_activation/tcp_server.rs b/examples/src/socket_activation/tcp_server.rs new file mode 100644 index 000000000..205f7b37c --- /dev/null +++ b/examples/src/socket_activation/tcp_server.rs @@ -0,0 +1,49 @@ +//! Greeter server demonstrating systemd socket activation over a TCP socket. + +use tonic::{Request, Response, Status, transport::Server}; + +use hello_world::greeter_server::{Greeter, GreeterServer}; +use hello_world::{HelloReply, HelloRequest}; + +pub mod hello_world { + tonic::include_proto!("helloworld"); +} + +#[derive(Default)] +pub struct MyGreeter {} + +#[tonic::async_trait] +impl Greeter for MyGreeter { + async fn say_hello( + &self, + request: Request, + ) -> Result, Status> { + println!("Got a request from {:?}", request.remote_addr()); + + let reply = hello_world::HelloReply { + message: format!("Hello {}!", request.into_inner().name), + }; + Ok(Response::new(reply)) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let addr = "[::1]:50051".parse().unwrap(); + let greeter = MyGreeter::default(); + + if std::env::var_os("LISTEN_FDS").is_some() { + println!( + "socket activation detected: adopting socket-activated descriptor for {addr} if available" + ); + } else { + println!("no socket activation: binding {addr} directly"); + } + + Server::builder() + .add_service(GreeterServer::new(greeter)) + .serve(addr) + .await?; + + Ok(()) +} diff --git a/examples/src/socket_activation/test_tcp.sh b/examples/src/socket_activation/test_tcp.sh new file mode 100755 index 000000000..bc26a7ca4 --- /dev/null +++ b/examples/src/socket_activation/test_tcp.sh @@ -0,0 +1,90 @@ +#!/usr/bin/env bash +# Smoke test for the tonic systemd socket-activation example over TCP. + +set -u + +ADDR="[::1]:50051" +UNIT_DIR="${XDG_CONFIG_HOME:-${HOME}/.config}/systemd/user" +SOCKET_UNIT="socketact-tcp.socket" +SERVICE_UNIT="socketact-tcp.service" + +clean() { + echo "Cleaning..." + systemctl --user stop "${SOCKET_UNIT}" 2>/dev/null + systemctl --user stop "${SERVICE_UNIT}" 2>/dev/null + rm -f "${UNIT_DIR}/${SOCKET_UNIT}" "${UNIT_DIR}/${SERVICE_UNIT}" + systemctl --user daemon-reload 2>/dev/null +} + +fail() { + clean + echo "FAIL: $*" >&2 + exit 1 +} + +pass() { + echo "SUCCESS: $1" +} + +command -v systemctl >/dev/null 2>&1 \ + || fail "systemctl not found (install systemd)" +systemctl --user show-environment >/dev/null 2>&1 \ + || fail "no systemd user instance available (needs a user session bus)" + +echo "Building example binaries..." +cargo build --bin socket-activation-tcp-server --bin socket-activation-tcp-client \ + || fail "Failed to build example binaries" + +SERVER_BIN="$(cargo metadata --format-version 1 --no-deps \ + | grep -o '"target_directory":"[^"]*"' | head -n1 | cut -d'"' -f4)/debug/socket-activation-tcp-server" +[[ -x "${SERVER_BIN}" ]] || fail "Server binary not found at ${SERVER_BIN}" + +# Install user units. systemd owns the listening socket; the service inherits it +# as fd 3 when it is activated on the first client connection. +echo "Installing user units into ${UNIT_DIR}..." +mkdir -p "${UNIT_DIR}" + +cat > "${UNIT_DIR}/${SOCKET_UNIT}" < "${UNIT_DIR}/${SERVICE_UNIT}" </dev/null + systemctl --user stop "${SERVICE_UNIT}" 2>/dev/null + rm -f "${UNIT_DIR}/${SOCKET_UNIT}" "${UNIT_DIR}/${SERVICE_UNIT}" + systemctl --user daemon-reload 2>/dev/null + rm -f "${SOCK}" +} + +fail() { + clean + echo "FAIL: $*" >&2 + exit 1 +} + +pass() { + echo "SUCCESS: $1" +} + +command -v systemctl >/dev/null 2>&1 \ + || fail "systemctl not found (install systemd)" +systemctl --user show-environment >/dev/null 2>&1 \ + || fail "no systemd user instance available (needs a user session bus)" + +echo "Building example binaries..." +cargo build --bin socket-activation-uds-server --bin socket-activation-uds-client \ + || fail "Failed to build example binaries" + +SERVER_BIN="$(cargo metadata --format-version 1 --no-deps \ + | grep -o '"target_directory":"[^"]*"' | head -n1 | cut -d'"' -f4)/debug/socket-activation-uds-server" +[[ -x "${SERVER_BIN}" ]] || fail "Server binary not found at ${SERVER_BIN}" + +# Install user units. systemd owns the listening socket; the service inherits it +# as fd 3 when it is activated on the first client connection. TONIC_UDS_PATH +# tells the server which path the descriptor corresponds to. +echo "Installing user units into ${UNIT_DIR}..." +mkdir -p "${UNIT_DIR}" +rm -f "${SOCK}" + +cat > "${UNIT_DIR}/${SOCKET_UNIT}" < "${UNIT_DIR}/${SERVICE_UNIT}" < Result<(), Box> { + let path = std::env::var("TONIC_UDS_PATH") + .unwrap_or_else(|_| "/tmp/tonic/socket-activation.sock".to_string()); + + // The URI is ignored by the custom connector; UDS uses the path below. + let channel = Endpoint::try_from("http://[::]:50051")? + .connect_with_connector(service_fn(move |_: Uri| { + let path = path.clone(); + async move { Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path).await?)) } + })) + .await?; + + let mut client = GreeterClient::new(channel); + + let request = tonic::Request::new(HelloRequest { + name: "Tonic".into(), + }); + + let response = client.say_hello(request).await?; + + println!("RESPONSE={response:?}"); + + Ok(()) +} + +#[cfg(not(unix))] +fn main() { + panic!("The socket-activation UDS example only works on unix systems!"); +} diff --git a/examples/src/socket_activation/uds_server.rs b/examples/src/socket_activation/uds_server.rs new file mode 100644 index 000000000..e78474160 --- /dev/null +++ b/examples/src/socket_activation/uds_server.rs @@ -0,0 +1,67 @@ +//! Greeter server demonstrating systemd socket activation over a Unix domain +#![cfg_attr(not(unix), allow(unused_imports))] + +use tonic::{Request, Response, Status, transport::Server}; + +#[cfg(unix)] +use tonic::transport::server::UnixIncoming; + +use hello_world::greeter_server::{Greeter, GreeterServer}; +use hello_world::{HelloReply, HelloRequest}; + +pub mod hello_world { + tonic::include_proto!("helloworld"); +} + +#[derive(Default)] +pub struct MyGreeter {} + +#[tonic::async_trait] +impl Greeter for MyGreeter { + async fn say_hello( + &self, + request: Request, + ) -> Result, Status> { + println!("Got a request: {request:?}"); + + let reply = hello_world::HelloReply { + message: format!("Hello {}!", request.into_inner().name), + }; + Ok(Response::new(reply)) + } +} + +#[cfg(unix)] +#[tokio::main] +async fn main() -> Result<(), Box> { + let path = std::env::var("TONIC_UDS_PATH") + .unwrap_or_else(|_| "/tmp/tonic/socket-activation.sock".to_string()); + let greeter = MyGreeter::default(); + + if std::env::var_os("LISTEN_FDS").is_some() { + println!( + "socket activation detected: adopting socket-activated descriptor for {path} if available" + ); + } else { + println!("no socket activation: binding {path} directly"); + if let Some(parent) = std::path::Path::new(&path).parent() { + std::fs::create_dir_all(parent)?; + } + // Remove a stale socket file from a previous direct run. + let _ = std::fs::remove_file(&path); + } + + let incoming = UnixIncoming::bind(&path)?; + + Server::builder() + .add_service(GreeterServer::new(greeter)) + .serve_with_incoming(incoming) + .await?; + + Ok(()) +} + +#[cfg(not(unix))] +fn main() { + panic!("The socket-activation UDS example only works on unix systems!"); +} diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 69aef608c..52d521e1e 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -50,6 +50,9 @@ channel = [ ] transport = ["server", "channel"] +# Adopt listening sockets passed in by systemd. +socket-activation = [] + # [[bench]] # name = "bench_main" # harness = false diff --git a/tonic/src/transport/server/incoming.rs b/tonic/src/transport/server/incoming.rs index 3eb06d76e..2a455fb5d 100644 --- a/tonic/src/transport/server/incoming.rs +++ b/tonic/src/transport/server/incoming.rs @@ -29,6 +29,12 @@ impl TcpIncoming { /// /// Returns a TcpIncoming if the socket address was successfully bound. /// + /// If the process was launched under a socket-activation manager + /// that passed a listening socket matching `addr` via the + /// `LISTEN_FDS` / `LISTEN_PID` environment variables, that inherited + /// descriptor is adopted instead of opening a new socket. This behavior + /// requires the `socket-activation` feature (Unix only). + /// /// # Examples /// ```no_run /// # use tower_service::Service; @@ -57,7 +63,11 @@ impl TcpIncoming { /// # Ok(()) /// # } pub fn bind(addr: SocketAddr) -> std::io::Result { - let std_listener = StdTcpListener::bind(addr)?; + let std_listener = match find_preallocated_fd(addr) { + Some(listener) => listener, + None => StdTcpListener::bind(addr)?, + }; + std_listener.set_nonblocking(true)?; Ok(TcpListener::from_std(std_listener)?.into()) @@ -224,9 +234,133 @@ fn make_keepalive( dirty.then_some(keepalive) } +// Adopts a socket-activation fd whose address matches `addr`, if one was passed in. +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +fn find_preallocated_fd(addr: SocketAddr) -> Option { + use std::os::unix::io::FromRawFd; + + let fd = super::socket_activation::find_preallocated_fd(|fd| tcp_fd_matches(fd, addr))?; + + // SAFETY: `fd` is a validated, open activation descriptor. Ownership is taken + // once here, the returned listener becomes its sole owner. + Some(unsafe { StdTcpListener::from_raw_fd(fd) }) +} + +// Returns true if the listening socket at `fd` is bound to the requested address. +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +fn tcp_fd_matches(fd: std::os::unix::io::RawFd, requested: SocketAddr) -> bool { + use std::mem::ManuallyDrop; + use std::os::unix::io::FromRawFd; + + // SAFETY: `fd` is a valid, open activation descriptor. `ManuallyDrop` keeps + // ownership with the caller so it is not closed here, it is only borrowed to + // read the bound address. + let listener = ManuallyDrop::new(unsafe { StdTcpListener::from_raw_fd(fd) }); + matches!(listener.local_addr(), Ok(local) if socket_addr_matches(local, requested)) +} + +// Compares two socket addresses, treating IPv4-mapped and wildcard binds as equal. +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +fn socket_addr_matches(inherited: SocketAddr, requested: SocketAddr) -> bool { + use std::net::IpAddr; + + if inherited.port() != requested.port() { + return false; + } + + // Normalize IPv4-mapped IPv6 addresses to plain IPv4. + fn normalize(ip: IpAddr) -> IpAddr { + match ip { + IpAddr::V6(v6) => match v6.to_ipv4_mapped() { + Some(v4) => IpAddr::V4(v4), + None => IpAddr::V6(v6), + }, + v4 => v4, + } + } + + let inherited_ip = normalize(inherited.ip()); + let requested_ip = normalize(requested.ip()); + + if inherited_ip == requested_ip { + return true; + } + + requested_ip.is_unspecified() && inherited_ip.is_unspecified() +} + +#[cfg(not(all(target_os = "linux", feature = "socket-activation")))] +fn find_preallocated_fd(_addr: SocketAddr) -> Option { + None +} + #[cfg(test)] mod tests { use crate::transport::server::TcpIncoming; + + #[cfg(all(target_os = "linux", feature = "socket-activation"))] + #[test] + fn socket_addr_matches_cases() { + use super::socket_addr_matches; + + let parse = |s: &str| -> std::net::SocketAddr { s.parse().unwrap() }; + + assert!(socket_addr_matches( + parse("127.0.0.1:50051"), + parse("127.0.0.1:50051") + )); + + assert!(!socket_addr_matches( + parse("127.0.0.1:50051"), + parse("127.0.0.1:1234") + )); + + assert!(!socket_addr_matches( + parse("127.0.0.1:50051"), + parse("192.168.0.1:50051") + )); + + assert!(socket_addr_matches( + parse("[::]:50051"), + parse("0.0.0.0:50051") + )); + assert!(socket_addr_matches( + parse("0.0.0.0:50051"), + parse("[::]:50051") + )); + + assert!(socket_addr_matches( + parse("[::ffff:127.0.0.1]:50051"), + parse("127.0.0.1:50051") + )); + + assert!(!socket_addr_matches( + parse("127.0.0.1:50051"), + parse("0.0.0.0:50051") + )); + } + + #[cfg(all(target_os = "linux", feature = "socket-activation"))] + #[test] + fn is_listening_stream_socket_cases() { + use crate::transport::server::socket_activation::is_listening_stream_socket; + use std::os::unix::io::AsRawFd; + + let tcp = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + assert!(is_listening_stream_socket(tcp.as_raw_fd())); + + let udp = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + assert!(!is_listening_stream_socket(udp.as_raw_fd())); + + let raw = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::STREAM, + Some(socket2::Protocol::TCP), + ) + .unwrap(); + assert!(!is_listening_stream_socket(raw.as_raw_fd())); + } + #[tokio::test] async fn one_tcpincoming_at_a_time() { let addr = "127.0.0.1:1322".parse().unwrap(); @@ -236,4 +370,24 @@ mod tests { } let _t3 = TcpIncoming::bind(addr).unwrap(); } + + #[cfg(all(target_os = "linux", feature = "socket-activation"))] + #[test] + fn scan_adopts_matching_tcp_fd() { + use super::tcp_fd_matches; + use crate::transport::server::socket_activation::scan_preallocated_fds; + use std::net::TcpListener as StdTcpListener; + use std::os::unix::io::AsRawFd; + + let listener = StdTcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let fd = listener.as_raw_fd(); + + let n_fds = fd - 2; + let found = scan_preallocated_fds(std::process::id(), n_fds, |candidate| { + tcp_fd_matches(candidate, addr) + }); + + assert_eq!(found, Some(fd)); + } } diff --git a/tonic/src/transport/server/mod.rs b/tonic/src/transport/server/mod.rs index af89c2f9d..5298fc1f9 100644 --- a/tonic/src/transport/server/mod.rs +++ b/tonic/src/transport/server/mod.rs @@ -5,6 +5,8 @@ mod display_error_stack; mod incoming; mod io_stream; mod service; +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +mod socket_activation; #[cfg(feature = "_tls-any")] mod tls; #[cfg(unix)] @@ -35,7 +37,7 @@ pub use conn::TlsConnectInfo; use self::service::TlsAcceptor; #[cfg(unix)] -pub use unix::UdsConnectInfo; +pub use unix::{UdsConnectInfo, UnixIncoming}; pub use incoming::TcpIncoming; diff --git a/tonic/src/transport/server/socket_activation.rs b/tonic/src/transport/server/socket_activation.rs new file mode 100644 index 000000000..2f0118b51 --- /dev/null +++ b/tonic/src/transport/server/socket_activation.rs @@ -0,0 +1,41 @@ +// Shared helpers for adopting sockets passed in by a socket-activation manager + +use std::os::unix::io::{BorrowedFd, RawFd}; + +const SD_LISTEN_FDS_START: RawFd = 3; + +// Reads the socket-activation env vars (LISTEN_PID/LISTEN_FDS) and returns the +// first inherited fd accepted by `matches`, or None if none were passed to us. +pub(super) fn find_preallocated_fd(matches: F) -> Option +where + F: Fn(RawFd) -> bool, +{ + let listen_pid: u32 = std::env::var("LISTEN_PID").ok()?.parse().ok()?; + let n_fds: i32 = std::env::var("LISTEN_FDS").ok()?.parse().ok()?; + + scan_preallocated_fds(listen_pid, n_fds, matches) +} + +pub(super) fn scan_preallocated_fds(listen_pid: u32, n_fds: i32, matches: F) -> Option +where + F: Fn(RawFd) -> bool, +{ + if listen_pid != std::process::id() { + return None; + } + + let end = SD_LISTEN_FDS_START.checked_add(n_fds)?; + + (SD_LISTEN_FDS_START..end).find(|&fd| is_listening_stream_socket(fd) && matches(fd)) +} + +/// Returns `true` if `fd` refers to a stream socket in the listening state. +pub(super) fn is_listening_stream_socket(fd: RawFd) -> bool { + // SAFETY: `fd` is a valid, open descriptor from the activation range. It is + // only borrowed, not owned: the `BorrowedFd` does not close it on drop and + // does not outlive this function. + let borrowed = unsafe { BorrowedFd::borrow_raw(fd) }; + let sock = socket2::SockRef::from(&borrowed); + + matches!(sock.r#type(), Ok(socket2::Type::STREAM)) && matches!(sock.is_listener(), Ok(true)) +} diff --git a/tonic/src/transport/server/unix.rs b/tonic/src/transport/server/unix.rs index 78b408c95..998dc7e26 100644 --- a/tonic/src/transport/server/unix.rs +++ b/tonic/src/transport/server/unix.rs @@ -1,5 +1,120 @@ use super::Connected; use std::sync::Arc; +use std::{ + os::unix::net::UnixListener as StdUnixListener, + path::Path, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::net::{UnixListener, UnixStream}; +use tokio_stream::{Stream, wrappers::UnixListenerStream}; + +/// Binds a Unix domain socket for a [Router](super::Router). +/// +/// An incoming stream, usable with +/// [Router::serve_with_incoming](super::Router::serve_with_incoming), of +/// `AsyncRead + AsyncWrite` that communicate with clients that connect to a +/// Unix domain socket path. +#[derive(Debug)] +pub struct UnixIncoming { + inner: UnixListenerStream, +} + +impl UnixIncoming { + /// Creates an instance by binding (opening) the specified socket path. + /// + /// Returns a `UnixIncoming` if the socket path was successfully bound. + /// + /// If the process was launched under a socket-activation manager + /// that passed a listening Unix socket matching `path` via the + /// `LISTEN_FDS` / `LISTEN_PID` environment variables, that inherited + /// descriptor is adopted instead of opening a new socket. This behavior + /// requires the `socket-activation` feature. + /// + /// # Examples + /// ```no_run + /// # use tower_service::Service; + /// # use http::{request::Request, response::Response}; + /// # use tonic::{body::Body, server::NamedService, transport::{Server, server::UnixIncoming}}; + /// # use core::convert::Infallible; + /// # use std::error::Error; + /// # fn main() { } + /// # fn run(some_service: S) -> Result<(), Box> + /// # where + /// # S: Service, Response = Response, Error = Infallible> + NamedService + Clone + Send + Sync + 'static, + /// # S::Future: Send + 'static, + /// # { + /// let uinc = UnixIncoming::bind("/tmp/tonic/helloworld")?; + /// Server::builder() + /// .add_service(some_service) + /// .serve_with_incoming(uinc); + /// # Ok(()) + /// # } + /// ``` + pub fn bind(path: impl AsRef) -> std::io::Result { + let path = path.as_ref(); + let std_listener = match find_preallocated_fd(path) { + Some(listener) => listener, + None => StdUnixListener::bind(path)?, + }; + + std_listener.set_nonblocking(true)?; + + Ok(UnixListener::from_std(std_listener)?.into()) + } + + /// Returns the local address that this Unix incoming is bound to. + pub fn local_addr(&self) -> std::io::Result { + self.inner.as_ref().local_addr() + } +} + +impl From for UnixIncoming { + fn from(listener: UnixListener) -> Self { + Self { + inner: UnixListenerStream::new(listener), + } + } +} + +impl Stream for UnixIncoming { + type Item = std::io::Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_next(cx) + } +} + +// Adopts a socket-activation fd bound to `path`, if one was passed in. +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +fn find_preallocated_fd(path: &Path) -> Option { + use std::os::unix::io::FromRawFd; + + let fd = super::socket_activation::find_preallocated_fd(|fd| unix_fd_matches(fd, path))?; + + // SAFETY: `fd` is a validated, open activation descriptor. Ownership is taken + // once here, the returned listener becomes its sole owner. + Some(unsafe { StdUnixListener::from_raw_fd(fd) }) +} + +#[cfg(not(all(target_os = "linux", feature = "socket-activation")))] +fn find_preallocated_fd(_path: &Path) -> Option { + None +} + +// Returns true if the listening socket at `fd` is bound to the requested path. +#[cfg(all(target_os = "linux", feature = "socket-activation"))] +fn unix_fd_matches(fd: std::os::unix::io::RawFd, requested: &Path) -> bool { + use std::mem::ManuallyDrop; + use std::os::unix::io::FromRawFd; + + // SAFETY: `fd` is a valid, open activation descriptor. `ManuallyDrop` keeps + // ownership with the caller so it is not closed here; it is only borrowed to + // read the bound address. + let listener = ManuallyDrop::new(unsafe { StdUnixListener::from_raw_fd(fd) }); + matches!(listener.local_addr(), Ok(addr) if addr.as_pathname() == Some(requested)) +} /// Connection info for Unix domain socket streams. /// @@ -27,3 +142,79 @@ impl Connected for tokio::net::UnixStream { } } } + +#[cfg(all(test, target_os = "linux", feature = "socket-activation"))] +mod tests { + use std::os::unix::net::UnixListener as StdUnixListener; + use std::path::PathBuf; + + fn temp_socket_path(tag: &str) -> PathBuf { + let mut path = std::env::temp_dir(); + path.push(format!( + "tonic-uds-test-{}-{}.sock", + tag, + std::process::id() + )); + let _ = std::fs::remove_file(&path); + path + } + + #[test] + fn unix_fd_matches_cases() { + use super::unix_fd_matches; + use std::os::unix::io::AsRawFd; + + let path = temp_socket_path("matches"); + let other = temp_socket_path("matches-other"); + + let listener = StdUnixListener::bind(&path).unwrap(); + let fd = listener.as_raw_fd(); + + assert!(unix_fd_matches(fd, &path)); + assert!(!unix_fd_matches(fd, &other)); + + drop(listener); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn is_listening_stream_socket_cases() { + use crate::transport::server::socket_activation::is_listening_stream_socket; + use std::os::unix::io::AsRawFd; + + let path = temp_socket_path("listening"); + + let listener = StdUnixListener::bind(&path).unwrap(); + assert!(is_listening_stream_socket(listener.as_raw_fd())); + + let dgram = std::os::unix::net::UnixDatagram::unbound().unwrap(); + assert!(!is_listening_stream_socket(dgram.as_raw_fd())); + + let raw = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None).unwrap(); + assert!(!is_listening_stream_socket(raw.as_raw_fd())); + + drop(listener); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn scan_adopts_matching_unix_fd() { + use super::unix_fd_matches; + use crate::transport::server::socket_activation::scan_preallocated_fds; + use std::os::unix::io::AsRawFd; + + let path = temp_socket_path("scan"); + let listener = StdUnixListener::bind(&path).unwrap(); + let fd = listener.as_raw_fd(); + + let n_fds = fd - 2; + let found = scan_preallocated_fds(std::process::id(), n_fds, |candidate| { + unix_fd_matches(candidate, &path) + }); + + assert_eq!(found, Some(fd)); + + drop(listener); + let _ = std::fs::remove_file(&path); + } +}