use std::cell::RefCell;
use std::os::fd::{AsFd, OwnedFd};
use std::path::Path;
use std::sync::atomic::{
AtomicBool, AtomicI32, AtomicU32, AtomicU64, AtomicU8, AtomicUsize, Ordering,
};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::{Duration, Instant};
use super::deadline::PhaseDeadline;
use super::end::{EndReason, EndedBy, ServeStep, SessionEnd};
use super::fd_mode::FileDescriptorTransportMode;
use super::lifecycle::SessionLifecycle;
use crate::binder::{SIBinder, Stability, FLAG_ONEWAY, INTERFACE_TRANSACTION, PING_TRANSACTION};
use crate::error::{Result, StatusCode};
use crate::parcel::{CopiedBinders, Parcel, RpcParcelOps};
use super::address::{
AddressSpace, RpcAddress, SpecialTransaction, RPC_ADDR_LEN, RPC_SESSION_ID_NEW,
};
use super::proxy::RpcProxy;
use super::state::RpcState;
use super::transport::{PeerIdentity, RpcTransport};
use super::wire::{
R34Codec, WireCodec, WireMessage, WireReply, WireReplyRef, WireTransaction, WireTransactionRef,
};
use super::wire_android13::{
client_connect_with_id, client_read_connection_init, client_write_connection_header,
control_only_refusal, read_aosp_message, read_aosp_message_gated, read_aosp_message_with_fds,
read_r34_session_preamble, server_accept_deferred_init, write_aosp_message,
write_aosp_message_with_fds, Android13PlusCodec, RawTransportIo, A13_ADDR_LEN,
A13_DEC_STRONG_LEN, FD_MODE_NONE, FD_MODE_UNIX, PROTOCOL_V1, PROTOCOL_V2,
};
use super::{RpcError, RpcResult};
type Android13PlusAccept = (Box<dyn RpcTransport>, Android13PlusCodec, u8, Vec<u8>, bool);
#[derive(Clone, Copy)]
enum RpcUnixAddr<'a> {
Path(&'a Path),
#[cfg(any(target_os = "linux", target_os = "android"))]
Abstract(&'a [u8]),
}
#[deprecated(
since = "0.12.0",
note = "use `RpcClientConfig::unix`/`unix_abstract`, which carry the same knobs on every transport"
)]
pub struct RpcUnixClientConfig<'a> {
addr: RpcUnixAddr<'a>,
max_version: u32,
session_id: &'a [u8],
outgoing_connections: u32,
incoming_connections: u32,
fd_mode: Option<FileDescriptorTransportMode>,
timeout: Option<Duration>,
handshake_timeout: Option<Duration>,
}
#[expect(
deprecated,
reason = "implements the deprecated config while it is honored"
)]
impl<'a> RpcUnixClientConfig<'a> {
fn new(addr: RpcUnixAddr<'a>, max_version: u32) -> Self {
Self {
addr,
max_version,
session_id: &[],
outgoing_connections: 1,
incoming_connections: 0,
fd_mode: None,
timeout: None,
handshake_timeout: None,
}
}
pub fn path(path: &'a Path, max_version: u32) -> Self {
Self::new(RpcUnixAddr::Path(path), max_version)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn abstract_name(name: &'a [u8], max_version: u32) -> Self {
Self::new(RpcUnixAddr::Abstract(name), max_version)
}
pub fn session_id(mut self, session_id: &'a [u8]) -> Self {
self.session_id = session_id;
self
}
pub fn outgoing_connections(mut self, n: u32) -> Self {
self.outgoing_connections = n;
self
}
pub fn incoming_connections(mut self, n: u32) -> Self {
self.incoming_connections = n;
self
}
pub fn fd_mode(mut self, mode: FileDescriptorTransportMode) -> Self {
self.fd_mode = Some(mode);
self
}
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
pub fn handshake_timeout(mut self, timeout: Duration) -> Self {
self.handshake_timeout = Some(timeout);
self
}
fn into_generic(self) -> RpcClientConfig<'a> {
RpcClientConfig {
source: ClientSource::Unix(self.addr),
max_version: self.max_version,
session_id: self.session_id,
outgoing_connections: self.outgoing_connections,
incoming_connections: self.incoming_connections,
fd_mode: self.fd_mode,
timeout: self.timeout,
handshake_timeout: self.handshake_timeout,
}
}
fn into_attach(self) -> RpcClientConfig<'a> {
RpcClientConfig {
timeout: None,
..self.into_generic()
}
}
}
fn unix_connect(addr: RpcUnixAddr<'_>) -> Result<Box<dyn RpcTransport>> {
let t = match addr {
RpcUnixAddr::Path(path) => super::transport::UnixTransport::connect(path),
#[cfg(any(target_os = "linux", target_os = "android"))]
RpcUnixAddr::Abstract(name) => super::transport::UnixTransport::connect_abstract(name),
}
.map_err(StatusCode::from)?;
Ok(Box::new(t))
}
type Connector<'a> = Box<dyn FnMut() -> Result<Box<dyn RpcTransport>> + Send + 'a>;
enum ClientSource<'a> {
Unix(RpcUnixAddr<'a>),
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
Vsock {
cid: u32,
port: u32,
},
#[cfg(feature = "rpc-tcp-debug")]
TcpDebug(std::net::SocketAddr),
#[cfg(feature = "rpc-tls")]
Tls {
host: &'a str,
port: u16,
server_name: &'a str,
config: std::sync::Arc<rustls::ClientConfig>,
},
Custom(Connector<'a>),
}
impl<'a> ClientSource<'a> {
fn into_connector(self, handshake_timeout: Option<Duration>) -> Connector<'a> {
let _ = handshake_timeout;
match self {
ClientSource::Unix(addr) => Box::new(move || unix_connect(addr)),
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
ClientSource::Vsock { cid, port } => Box::new(move || {
Ok(
Box::new(super::transport::VsockTransport::connect(cid, port)?)
as Box<dyn RpcTransport>,
)
}),
#[cfg(feature = "rpc-tcp-debug")]
ClientSource::TcpDebug(addr) => Box::new(move || {
let tcp = tcp_connect(&addr, handshake_timeout).map_err(connect_status)?;
Ok(
Box::new(super::transport::TcpDebugTransport::from_stream(tcp)?)
as Box<dyn RpcTransport>,
)
}),
#[cfg(feature = "rpc-tls")]
ClientSource::Tls {
host,
port,
server_name,
config,
} => {
let mut pinned = None;
Box::new(move || {
connect_tls(
host,
port,
&mut pinned,
server_name,
&config,
handshake_timeout,
)
})
}
ClientSource::Custom(connect) => connect,
}
}
}
#[cfg(any(feature = "rpc-tcp-debug", feature = "rpc-tls"))]
fn tcp_connect(
addr: &std::net::SocketAddr,
deadline: Option<Duration>,
) -> std::io::Result<std::net::TcpStream> {
match deadline {
Some(d) => std::net::TcpStream::connect_timeout(addr, d),
None => std::net::TcpStream::connect(addr),
}
}
#[cfg(feature = "rpc-tls")]
const CONNECT_SHARE_FLOOR: Duration = Duration::from_secs(2);
#[cfg(feature = "rpc-tls")]
fn connect_share(left: Duration, tries: u32) -> Duration {
(left / tries).max(left.min(CONNECT_SHARE_FLOOR))
}
#[cfg(feature = "rpc-tls")]
fn tcp_connect_any(
addrs: &[std::net::SocketAddr],
deadline: Option<Duration>,
) -> std::io::Result<(std::net::TcpStream, std::net::SocketAddr)> {
let end = deadline.map(|d| Instant::now().checked_add(d));
let mut last = None;
for (i, addr) in addrs.iter().enumerate() {
let share = match end {
Some(Some(end)) => {
let left = end.saturating_duration_since(Instant::now());
let tries = u32::try_from(addrs.len() - i).unwrap_or(u32::MAX);
let share = connect_share(left, tries);
if share.is_zero() {
return Err(std::io::ErrorKind::TimedOut.into());
}
Some(share)
}
Some(None) | None => None,
};
match tcp_connect(addr, share) {
Ok(t) => return Ok((t, *addr)),
Err(e) => last = Some(e),
}
}
Err(last.unwrap_or_else(|| {
std::io::Error::new(std::io::ErrorKind::NotFound, "no address resolved")
}))
}
#[cfg(any(feature = "rpc-tcp-debug", feature = "rpc-tls"))]
fn connect_status(e: std::io::Error) -> StatusCode {
if e.kind() == std::io::ErrorKind::TimedOut {
StatusCode::TimedOut
} else {
StatusCode::from(e)
}
}
#[cfg(feature = "rpc-tls")]
fn connect_tls(
host: &str,
port: u16,
pinned: &mut Option<std::net::SocketAddr>,
server_name: &str,
config: &std::sync::Arc<rustls::ClientConfig>,
handshake_timeout: Option<Duration>,
) -> Result<Box<dyn RpcTransport>> {
let tcp = match *pinned {
Some(addr) => tcp_connect(&addr, handshake_timeout).map_err(connect_status)?,
None => {
use std::net::ToSocketAddrs;
let addrs: Vec<_> = (host, port).to_socket_addrs()?.collect();
let (t, addr) = tcp_connect_any(&addrs, handshake_timeout).map_err(connect_status)?;
*pinned = Some(addr);
t
}
};
if let Some(d) = handshake_timeout {
tcp.set_read_timeout(Some(d))?;
tcp.set_write_timeout(Some(d))?;
}
let mut whole = PhaseDeadline::arm_with(handshake_timeout, || {
use std::os::fd::AsFd;
super::transport::socket_shutdown_handle(tcp.as_fd())
});
let t = match super::transport::TlsTransport::connect(tcp, server_name, config.clone()) {
Ok(t) => t,
Err(_) if whole.fired() => return Err(StatusCode::TimedOut),
Err(e) => return Err(StatusCode::from(e)),
};
if !whole.disarm() {
return Err(StatusCode::TimedOut);
}
if handshake_timeout.is_some() {
t.set_read_timeout(None).map_err(StatusCode::from)?;
t.set_write_timeout(None).map_err(StatusCode::from)?;
}
Ok(Box::new(t))
}
struct AttachParts<'a> {
connect: Connector<'a>,
max_version: u32,
session_id: &'a [u8],
fd_mode: FileDescriptorTransportMode,
handshake_timeout: Option<Duration>,
}
pub struct RpcClientConfig<'a> {
source: ClientSource<'a>,
max_version: u32,
session_id: &'a [u8],
outgoing_connections: u32,
incoming_connections: u32,
fd_mode: Option<FileDescriptorTransportMode>,
timeout: Option<Duration>,
handshake_timeout: Option<Duration>,
}
impl<'a> RpcClientConfig<'a> {
fn with_source(source: ClientSource<'a>, max_version: u32) -> Self {
Self {
source,
max_version,
session_id: &[],
outgoing_connections: 1,
incoming_connections: 0,
fd_mode: None,
timeout: None,
handshake_timeout: None,
}
}
pub fn unix(path: &'a Path, max_version: u32) -> Self {
Self::with_source(ClientSource::Unix(RpcUnixAddr::Path(path)), max_version)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn unix_abstract(name: &'a [u8], max_version: u32) -> Self {
Self::with_source(ClientSource::Unix(RpcUnixAddr::Abstract(name)), max_version)
}
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
pub fn vsock(cid: u32, port: u32, max_version: u32) -> Self {
Self::with_source(ClientSource::Vsock { cid, port }, max_version)
}
#[cfg(feature = "rpc-tcp-debug")]
pub fn tcp_debug(addr: std::net::SocketAddr, max_version: u32) -> Self {
Self::with_source(ClientSource::TcpDebug(addr), max_version)
}
#[cfg(feature = "rpc-tls")]
pub fn tls(
host: &'a str,
port: u16,
server_name: &'a str,
config: std::sync::Arc<rustls::ClientConfig>,
max_version: u32,
) -> Self {
Self::with_source(
ClientSource::Tls {
host,
port,
server_name,
config,
},
max_version,
)
}
pub fn new(
max_version: u32,
connect: impl FnMut() -> Result<Box<dyn RpcTransport>> + Send + 'a,
) -> Self {
Self::with_source(ClientSource::Custom(Box::new(connect)), max_version)
}
pub fn session_id(mut self, session_id: &'a [u8]) -> Self {
self.session_id = session_id;
self
}
pub fn outgoing_connections(mut self, n: u32) -> Self {
self.outgoing_connections = n;
self
}
pub fn incoming_connections(mut self, n: u32) -> Self {
self.incoming_connections = n;
self
}
pub fn fd_mode(mut self, mode: FileDescriptorTransportMode) -> Self {
self.fd_mode = Some(mode);
self
}
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = Some(timeout);
self
}
#[deprecated(
since = "0.12.0",
note = "`timeout` bounds each handshake step too; set it instead (plan 2-24 D6)"
)]
pub fn handshake_timeout(mut self, timeout: Duration) -> Self {
self.handshake_timeout = Some(timeout);
self
}
fn handshake_deadline(&self) -> Option<Duration> {
self.handshake_timeout
.or(self.timeout.filter(|d| !d.is_zero()))
}
pub(crate) fn connect_once(self) -> Result<Box<dyn RpcTransport>> {
let deadline = self.handshake_deadline();
(self.source.into_connector(deadline))()
}
}
enum WireProfile {
R34(R34Codec),
Android13Plus(Android13PlusCodec),
}
impl WireProfile {
fn codec(&self) -> &dyn WireCodec {
match self {
WireProfile::R34(c) => c,
WireProfile::Android13Plus(c) => c,
}
}
fn counts_transaction_targets(&self) -> bool {
matches!(self, WireProfile::Android13Plus(_))
}
fn pays_target_before_reply(&self) -> bool {
matches!(self, WireProfile::Android13Plus(_))
}
fn dec_strong_body_len(&self) -> usize {
match self {
WireProfile::R34(_) => RPC_ADDR_LEN,
WireProfile::Android13Plus(_) => A13_DEC_STRONG_LEN,
}
}
fn binder_stability_repr(&self, stability: Option<Stability>) -> i32 {
match self {
WireProfile::R34(_) => {
crate::binder::android12_category_repr(stability.map_or(0, Stability::level))
}
WireProfile::Android13Plus(_) => stability.map_or(0, i32::from),
}
}
fn check_binder_stability(&self, repr: i32, null: bool) -> Result<()> {
let level = match self {
WireProfile::R34(_) => {
let version = repr & 0xff;
if version < 1 {
log::error!("RPC: binder stability {repr:#x} has wire format version 0");
return Err(StatusCode::BadType);
}
(repr >> 24) & 0xff
}
WireProfile::Android13Plus(_) => repr,
};
if null && level != 0 {
log::error!("RPC: null binder with stability {repr:#x}");
return Err(StatusCode::BadType);
}
Ok(())
}
fn wire_version(&self) -> Option<u32> {
match self {
WireProfile::R34(_) => None,
WireProfile::Android13Plus(c) => Some(c.version()),
}
}
fn records_binder_positions(&self) -> bool {
matches!(self.wire_version(), Some(v) if v >= PROTOCOL_V2)
}
fn records_fd_positions(&self) -> bool {
matches!(self.wire_version(), Some(v) if v >= PROTOCOL_V1)
}
}
pub(crate) fn write_rpc_interface_token(p: &mut Parcel, descriptor: &str) -> Result<()> {
p.write(&descriptor)?;
Ok(())
}
fn consume_rpc_interface_token(reader: &mut Parcel, expected: &str) -> Result<()> {
crate::parcelable::read_string16_matches(reader, expected)?.map_err(|got| {
log::error!("RPC interface token mismatch: expected '{expected}', got '{got}'");
StatusCode::BadType
})
}
fn write_addr(p: &mut Parcel, addr: &RpcAddress) -> Result<()> {
p.write_aligned_data(addr.as_wire_bytes().as_slice())
}
fn read_addr(p: &mut Parcel) -> Result<RpcAddress> {
let slice = p.read_aligned_data(RPC_ADDR_LEN)?;
let mut bytes = [0u8; RPC_ADDR_LEN];
bytes.copy_from_slice(slice);
Ok(RpcAddress::from_wire_bytes(bytes))
}
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct RpcSessionId([u8; 32]);
impl RpcSessionId {
pub(crate) fn new(bytes: [u8; 32]) -> Self {
Self(bytes)
}
pub(crate) fn as_bytes(&self) -> &[u8; 32] {
&self.0
}
pub(crate) fn try_from_slice(s: &[u8]) -> Option<Self> {
<[u8; 32]>::try_from(s).ok().map(Self)
}
}
impl std::fmt::Debug for RpcSessionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("RpcSessionId(...)")
}
}
fn gen_rpc_session_id() -> RpcResult<RpcSessionId> {
let mut id = [0u8; 32];
getrandom::fill(&mut id).map_err(|e| {
RpcError::Io(std::io::Error::other(format!(
"CSPRNG getrandom failed for RPC session id: {e}"
)))
})?;
Ok(RpcSessionId::new(id))
}
pub(super) fn reject_zero_deadline(timeout: Option<Duration>, msg: &str) -> Option<Duration> {
match timeout {
Some(d) if d.is_zero() => {
log::error!("{msg}");
None
}
other => other,
}
}
thread_local! {
static ARMED_READ: RefCell<Vec<(usize, Option<Duration>)>> = const { RefCell::new(Vec::new()) };
}
fn transport_key(t: &dyn RpcTransport) -> usize {
t as *const dyn RpcTransport as *const () as usize
}
fn read_deadline_in_effect(t: &dyn RpcTransport, baseline: Option<Duration>) -> Option<Duration> {
let key = transport_key(t);
ARMED_READ
.try_with(|a| a.borrow().iter().rev().find(|e| e.0 == key).map(|e| e.1))
.ok()
.flatten()
.unwrap_or(baseline)
}
fn push_read_deadline(t: &dyn RpcTransport, d: Option<Duration>) -> RpcResult<bool> {
if let Err(e) = t.set_read_timeout(d) {
return if t.peer_closed() == Some(true) {
Ok(false)
} else {
Err(e)
};
}
let _ = ARMED_READ.try_with(|a| a.borrow_mut().push((transport_key(t), d)));
Ok(true)
}
fn pop_read_deadline(t: &dyn RpcTransport, restore: Option<Duration>) {
let key = transport_key(t);
let _ = ARMED_READ.try_with(|a| {
let mut a = a.borrow_mut();
if let Some(i) = a.iter().rposition(|e| e.0 == key) {
a.remove(i);
}
});
let _ = t.set_read_timeout(restore);
}
struct ReplyDeadlineGuard<'a> {
transport: &'a dyn RpcTransport,
armed: bool,
restore: Option<Duration>,
}
impl<'a> ReplyDeadlineGuard<'a> {
fn arm(
transport: &'a dyn RpcTransport,
deadline: Option<Duration>,
baseline: Option<Duration>,
) -> RpcResult<Self> {
let restore = read_deadline_in_effect(transport, baseline);
let armed = match deadline {
Some(d) => push_read_deadline(transport, Some(d))?,
None => false,
};
Ok(Self {
transport,
armed,
restore,
})
}
}
impl Drop for ReplyDeadlineGuard<'_> {
fn drop(&mut self) {
if self.armed {
pop_read_deadline(self.transport, self.restore);
}
}
}
struct HandshakeDeadline<'a> {
transport: &'a dyn RpcTransport,
armed: bool,
whole: PhaseDeadline,
}
impl<'a> HandshakeDeadline<'a> {
fn arm(transport: &'a dyn RpcTransport, deadline: Option<Duration>) -> RpcResult<Self> {
if deadline.is_some_and(|d| d.is_zero()) {
log::error!(
"rsbinder RPC: a zero handshake deadline is not a deadline — pass a positive \
duration, or `None` to wait indefinitely on purpose"
);
return Err(RpcError::Io(std::io::Error::from(
std::io::ErrorKind::InvalidInput,
)));
}
let armed = match deadline {
Some(_) => match transport
.set_read_timeout(deadline)
.and_then(|()| transport.set_write_timeout(deadline))
{
Ok(()) => true,
Err(_) if transport.peer_closed() == Some(true) => false,
Err(e) => return Err(e),
},
None => false,
};
let whole = PhaseDeadline::arm_with(deadline, || transport.shutdown_handle());
Ok(Self {
transport,
armed,
whole,
})
}
fn finish(&mut self) -> RpcResult<()> {
if self.whole.disarm() {
Ok(())
} else {
Err(RpcError::Timeout)
}
}
fn classify(&self, e: RpcError) -> RpcError {
if self.whole.fired() {
RpcError::Timeout
} else {
e
}
}
}
pub(crate) fn reject_zero_handshake_timeout(timeout: Option<Duration>, setter: &str) -> Result<()> {
if timeout.is_some_and(|d| d.is_zero()) {
log::error!(
"rsbinder RPC: {setter} was given a zero duration, which is not a deadline; pass a \
positive duration, or leave it unset to wait indefinitely"
);
return Err(StatusCode::BadValue);
}
Ok(())
}
impl Drop for HandshakeDeadline<'_> {
fn drop(&mut self) {
if self.armed {
let _ = self.transport.set_read_timeout(None);
let _ = self.transport.set_write_timeout(None);
}
}
}
fn confirm_attach(
transport: &dyn RpcTransport,
codec: &Android13PlusCodec,
session_id: &[u8],
) -> RpcResult<()> {
let txn = WireTransaction {
address: RpcAddress::zero(),
code: SpecialTransaction::GetSessionId.code(),
flags: 0,
async_number: 0,
data: Vec::new(),
object_positions: Vec::new(),
};
let frame = codec.encode_transact(&txn)?;
let mut io = RawTransportIo(transport);
write_aosp_message(&mut io, &frame)?;
let reply = read_aosp_message(&mut io)?;
let peer_id = match codec.decode_message(&reply)? {
WireMessage::Reply(WireReply {
status: 0, data, ..
}) => {
let mut p = Parcel::from_vec(data);
p.set_data_position(0);
p.read::<Vec<u8>>()
.map_err(|_| RpcError::Protocol("malformed GET_SESSION_ID reply on an attach"))?
}
_ => {
return Err(RpcError::Protocol(
"peer did not answer GET_SESSION_ID on an attach",
))
}
};
if peer_id == session_id {
Ok(())
} else {
Err(RpcError::Protocol(
"peer put this attach in a different session than the id it echoed",
))
}
}
fn log_attach_refused(e: &RpcError) {
if matches!(
e,
RpcError::Timeout | RpcError::Truncated | RpcError::DeadlineMidFrame
) {
log::error!(
"android-13+ RPC: the attach admission probe (GET_SESSION_ID) did not complete \
({e}) — a read deadline armed by this caller ends it this way too, so this is \
not necessarily a refusal"
);
return;
}
log::error!(
"android-13+ RPC: the peer refused this attach ({e}) — it no longer knows this \
session, its outgoing-slot cap (`set_max_threads`) is spent, or it is shutting down. \
Keep the connection count within `RpcSession::negotiate()`"
);
}
fn refuse_id_on_new_session(session_id: &[u8], what: &str) -> Result<()> {
if session_id.is_empty() {
return Ok(());
}
log::error!(
"{what}: a session id would build a second client RpcSession on a server session \
another RpcSession founded, with its own oneway numbering, binder addresses and \
lifetime; add the connection with `add_outgoing_connection_with_config` on that \
RpcSession instead"
);
Err(StatusCode::BadValue)
}
fn client_handshake_err(e: RpcError) -> StatusCode {
match &e {
RpcError::EndOfStream => log::error!(
"rsbinder RPC: the android-13+ handshake failed at the transport ({e}) after \
the peer accepted the connection — it may be speaking the r34 (default) \
profile. Connect with `WireProfile::R34` (no `?profile=android13plus`), or \
enable the android-13+ wire on the server (`RpcServer::set_android13plus`)"
),
RpcError::Truncated => log::error!(
"rsbinder RPC: the android-13+ handshake failed part-way through a response \
({e}) — the peer closed mid-frame; it may be speaking the r34 (default) \
profile. Connect with `WireProfile::R34` (no \
`?profile=android13plus`), or enable the android-13+ wire on the server \
(`RpcServer::set_android13plus`)"
),
RpcError::Timeout | RpcError::DeadlineMidFrame => log::error!(
"rsbinder RPC: the android-13+ handshake did not complete within a deadline \
armed on this connection — that deadline is the caller's own (the \
whole-step bound of `RpcClientConfig::timeout` / `ClientOptions::timeout` \
or their deprecated `handshake_timeout`, the 10s whole-handshake bound \
`RpcSession::from_preconnected_fd` arms, or a read deadline set on the \
transport directly), so it may simply be shorter than this peer's \
legitimate response time, a peer answering a byte at a time included. A \
peer that should have answered well within it may be speaking the r34 \
(default) profile instead"
),
RpcError::Protocol(_) => log::error!(
"rsbinder RPC: the android-13+ handshake failed ({e}) — either the peer's \
answer violated the wire or the caller offered a `max_version` this build \
does not implement"
),
_ => {}
}
StatusCode::from(e)
}
struct NestedDeadlineGuard<'a> {
transport: &'a dyn RpcTransport,
restore: Option<Duration>,
}
impl<'a> NestedDeadlineGuard<'a> {
fn lift(transport: &'a dyn RpcTransport, deadline: Option<Duration>) -> RpcResult<Self> {
let lifted = match deadline {
Some(_) => push_read_deadline(transport, None)?,
None => false,
};
Ok(Self {
transport,
restore: deadline.filter(|_| lifted),
})
}
}
impl Drop for NestedDeadlineGuard<'_> {
fn drop(&mut self) {
if let Some(d) = self.restore {
pop_read_deadline(self.transport, Some(d));
}
}
}
thread_local! {
static DRIVING: RefCell<Vec<(usize, u64)>> = const { RefCell::new(Vec::new()) };
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum SlotRole {
Incoming,
Outgoing,
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum ConnUse {
Client,
ClientAsync,
Reply,
}
impl ConnUse {
fn is_new_transaction(self) -> bool {
matches!(self, ConnUse::Client | ConnUse::ClientAsync)
}
}
enum DecRoute<'a> {
Conn(ConnGuard<'a>),
Pending(u64),
Reaper,
}
const HELD_DEC_STRONG_LIMIT: usize = 10_000;
#[derive(Default)]
struct DrainedReleases {
released: Vec<SIBinder>,
releases: Vec<(RpcAddress, u32)>,
}
enum ReplyNotSent {
Refused(StatusCode),
Failed(StatusCode),
}
impl ReplyNotSent {
fn refused(e: RpcError) -> Self {
match e {
RpcError::FrameTooLarge { .. } => ReplyNotSent::Refused(StatusCode::FailedTransaction),
e => ReplyNotSent::Refused(e.into()),
}
}
}
type Tid = std::thread::ThreadId;
#[inline]
fn current_tid() -> Tid {
std::thread::current().id()
}
mod slot_pool {
use std::ops::{Index, IndexMut};
use std::sync::Arc;
use super::{ConnTraits, RpcAddress, RpcTransport, SlotRole, Tid};
pub(super) struct ConnSlot {
pub(super) transport: Arc<dyn RpcTransport>,
pub(super) exclusive_tid: Option<Tid>,
pub(super) id: u64,
pub(super) role: SlotRole,
pub(super) allow_nested: bool,
pub(super) pending_dec: std::collections::HashMap<RpcAddress, u32>,
_pooled: (),
}
pub(super) struct ConnState {
pub(super) slots: SlotPool,
pub(super) next_slot_id: u64,
pub(super) traits: ConnTraits,
}
impl ConnState {
pub(super) fn new(
founding: Arc<dyn RpcTransport>,
role: SlotRole,
traits: ConnTraits,
) -> Self {
let mut slots = SlotPool(Vec::new());
slots.push(founding, 1, role, None);
ConnState {
slots,
next_slot_id: 2,
traits,
}
}
}
pub(super) struct SlotPool(Vec<ConnSlot>);
impl SlotPool {
pub(super) fn push(
&mut self,
transport: Arc<dyn RpcTransport>,
id: u64,
role: SlotRole,
exclusive_tid: Option<Tid>,
) {
self.0.push(ConnSlot {
transport,
exclusive_tid,
id,
role,
allow_nested: false,
pending_dec: std::collections::HashMap::new(),
_pooled: (),
});
}
pub(super) fn iter(&self) -> std::slice::Iter<'_, ConnSlot> {
self.0.iter()
}
pub(super) fn iter_mut(&mut self) -> std::slice::IterMut<'_, ConnSlot> {
self.0.iter_mut()
}
pub(super) fn len(&self) -> usize {
self.0.len()
}
pub(super) fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub(super) fn unpush_retired(&mut self, slot_id: u64) {
self.0.retain(|s| s.id != slot_id);
}
pub(super) fn clear_at_session_end(&mut self) {
self.0.clear();
}
}
impl Index<usize> for SlotPool {
type Output = ConnSlot;
fn index(&self, i: usize) -> &ConnSlot {
&self.0[i]
}
}
impl IndexMut<usize> for SlotPool {
fn index_mut(&mut self, i: usize) -> &mut ConnSlot {
&mut self.0[i]
}
}
}
use slot_pool::{ConnSlot, ConnState};
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
struct ConnTraits {
passes_fds: bool,
local_peer: bool,
}
impl ConnTraits {
fn of(transport: &dyn RpcTransport) -> Self {
ConnTraits {
passes_fds: transport.supports_fd_passing(),
local_peer: transport.peer_identity().is_local(),
}
}
}
impl ConnState {
fn admits(&self, transport: &dyn RpcTransport) -> bool {
let traits = ConnTraits::of(transport);
if traits != self.traits {
log::error!(
"RPC: connection refused — its transport ({traits:?}) differs from the \
session's founding connection ({:?})",
self.traits
);
}
traits == self.traits
}
}
struct ConnGuard<'a> {
inner: &'a RpcSessionInner,
slot_id: u64,
transport: Arc<dyn RpcTransport>,
reentrant: bool,
}
impl ConnGuard<'_> {
#[inline]
fn transport(&self) -> &dyn RpcTransport {
&*self.transport
}
}
impl Drop for ConnGuard<'_> {
fn drop(&mut self) {
if self.reentrant {
return;
}
let key = (self.inner as *const _ as usize, self.slot_id);
DRIVING.with(|d| {
let mut v = d.borrow_mut();
if let Some(pos) = v.iter().rposition(|&k| k == key) {
v.remove(pos);
}
});
{
let mut st = self.inner.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == self.slot_id) {
s.exclusive_tid = None;
}
drop(st);
self.inner.slot_cv.notify_all();
}
}
}
struct SlotClaim<'a> {
inner: &'a RpcSessionInner,
slot_id: u64,
transport: Arc<dyn RpcTransport>,
}
impl SlotClaim<'_> {
fn retire(self) {
{
let mut st = self.inner.conn_state.lock().expect("conn_state poisoned");
st.slots.unpush_retired(self.slot_id);
}
let _serial = self
.inner
.shared
.liveness
.lock()
.expect("liveness poisoned");
if let Err(e) = self.transport.shutdown() {
log::warn!("RPC: shutting a retired connection down failed: {e}");
}
}
}
impl Drop for SlotClaim<'_> {
fn drop(&mut self) {
{
let mut st = self.inner.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == self.slot_id) {
s.exclusive_tid = None;
}
}
self.inner.slot_cv.notify_all();
}
}
struct AllowNestedGuard<'a> {
inner: &'a RpcSessionInner,
slot_id: u64,
prev: bool,
}
impl<'a> AllowNestedGuard<'a> {
fn arm(inner: &'a RpcSessionInner, allow: bool) -> Option<Self> {
let sess_ptr = inner as *const RpcSessionInner as usize;
let slot_id = DRIVING.with(|d| {
d.borrow()
.iter()
.rev()
.find_map(|&(sp, sid)| if sp == sess_ptr { Some(sid) } else { None })
})?;
let mut st = inner.conn_state.lock().expect("conn_state poisoned");
let slot = st.slots.iter_mut().find(|s| s.id == slot_id)?;
let prev = std::mem::replace(&mut slot.allow_nested, allow);
Some(Self {
inner,
slot_id,
prev,
})
}
}
impl Drop for AllowNestedGuard<'_> {
fn drop(&mut self) {
let mut st = self.inner.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == self.slot_id) {
s.allow_nested = self.prev;
}
}
}
struct OpenCall<'a>(&'a SharedSession);
impl<'a> OpenCall<'a> {
fn enter(shared: &'a SharedSession) -> Self {
shared.open.fetch_add(1, Ordering::SeqCst);
OpenCall(shared)
}
}
impl Drop for OpenCall<'_> {
fn drop(&mut self) {
self.0.io_gen.fetch_add(1, Ordering::Relaxed);
self.0.open.fetch_sub(1, Ordering::SeqCst);
}
}
#[derive(Clone)]
pub(crate) struct SessionActivity(Arc<SharedSession>);
impl SessionActivity {
pub(crate) fn bump(&self) {
if self.0.serve_read_deadline.load(Ordering::Relaxed) != 0 {
self.0.io_gen.fetch_add(1, Ordering::Relaxed);
}
}
}
struct CountedIo<'a>(RawTransportIo<'a>, &'a AtomicU64);
impl std::io::Read for CountedIo<'_> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = std::io::Read::read(&mut self.0, buf)?;
if n > 0 {
self.1.fetch_add(1, Ordering::Relaxed);
}
Ok(n)
}
}
pub(crate) struct SharedSession {
state: Mutex<RpcState>,
root: Mutex<Option<SIBinder>>,
max_threads: AtomicU32,
negotiated: AtomicU32,
timeout: Mutex<Option<Duration>>,
serve_read_deadline: AtomicU64,
fd_mode: AtomicU8,
fd_unix_supported: AtomicBool,
rpc_session_id: RpcSessionId,
r34_session_id: AtomicI32,
server_session_id: Mutex<Option<Vec<u8>>>,
lifecycle: SessionLifecycle,
ended_locally: AtomicBool,
serve_declared: AtomicUsize,
space: AddressSpace,
open: AtomicUsize,
io_gen: AtomicU64,
liveness: Mutex<()>,
#[cfg(test)]
park_hook: Mutex<Option<mpsc::Sender<()>>>,
#[cfg(test)]
cci_failed_hook: Mutex<Option<Box<dyn FnOnce() + Send>>>,
#[cfg(test)]
serve_wait_hook: Mutex<Option<mpsc::Sender<()>>>,
#[cfg(test)]
fail_incoming_spawn: AtomicBool,
}
impl SharedSession {
fn fd_mode(&self) -> FileDescriptorTransportMode {
match self.fd_mode.load(Ordering::Acquire) {
FD_MODE_UNIX => FileDescriptorTransportMode::Unix,
_ => FileDescriptorTransportMode::None,
}
}
fn set_fd_mode(&self, mode: FileDescriptorTransportMode) {
let bits = match mode {
FileDescriptorTransportMode::None => FD_MODE_NONE,
FileDescriptorTransportMode::Unix => FD_MODE_UNIX,
};
self.fd_mode.store(bits, Ordering::Release);
}
pub(crate) fn local_node_count(&self) -> usize {
self.state
.lock()
.expect("rpc state poisoned")
.local_node_count()
}
pub(crate) fn live_conn_count(&self) -> usize {
self.lifecycle.live_count()
}
pub(crate) fn try_bump_live_conns(&self) -> bool {
self.lifecycle.try_bump_live()
}
pub(crate) fn space(&self) -> AddressSpace {
self.space
}
fn serve_read_deadline(&self) -> Option<Duration> {
match self.serve_read_deadline.load(Ordering::Relaxed) {
0 => None,
ns => Some(Duration::from_nanos(ns)),
}
}
fn spawn_incoming(
&self,
builder: std::thread::Builder,
f: impl FnOnce() + Send + 'static,
) -> std::io::Result<std::thread::JoinHandle<()>> {
#[cfg(test)]
if self.fail_incoming_spawn.load(Ordering::SeqCst) {
return Err(std::io::ErrorKind::OutOfMemory.into());
}
builder.spawn(f)
}
#[cfg(test)]
fn run_serve_wait_hook(&self) {
if let Some(tx) = &*self.serve_wait_hook.lock().expect("serve hook") {
let _ = tx.send(());
}
}
#[cfg(test)]
fn run_cci_failed_hook(&self) {
let hook = self.cci_failed_hook.lock().expect("cci hook").take();
if let Some(hook) = hook {
hook();
}
}
}
pub(crate) struct RpcSessionInner {
conn_state: Mutex<ConnState>,
slot_cv: Condvar,
profile: WireProfile,
self_weak: Weak<RpcSessionInner>,
parcel_ops: Arc<SessionParcelOps>,
dec_strong_tx: mpsc::Sender<(RpcAddress, u32)>,
shared: Arc<SharedSession>,
incoming_threads: Mutex<Vec<(u64, std::thread::JoinHandle<()>)>>,
incoming_live: AtomicUsize,
incoming_joined: AtomicUsize,
awaits_preamble: AtomicBool,
}
struct SessionParcelOps(Weak<RpcSessionInner>);
impl RpcParcelOps for SessionParcelOps {
fn write_binder(&self, binder: Option<&SIBinder>, parcel: &mut Parcel) -> Result<()> {
let inner = self.0.upgrade().ok_or(StatusCode::DeadObject)?;
inner.write_binder(binder, parcel)
}
fn read_binder(&self, parcel: &mut Parcel) -> Result<Option<SIBinder>> {
let inner = self.0.upgrade().ok_or(StatusCode::DeadObject)?;
inner.read_binder(parcel)
}
fn cancel_leaving(&self, addrs: &[RpcAddress]) {
if let Some(inner) = self.0.upgrade() {
inner.cancel_leaving(addrs);
}
}
fn session_id(&self) -> *const () {
self.0.as_ptr().cast()
}
fn records_binder_positions(&self) -> Result<bool> {
let inner = self.0.upgrade().ok_or(StatusCode::DeadObject)?;
Ok(inner.profile.records_binder_positions())
}
fn acquire_copied(&self, objects: &[&[u8]]) -> Result<CopiedBinders> {
let inner = self.0.upgrade().ok_or(StatusCode::DeadObject)?;
inner.acquire_copied(objects)
}
}
impl RpcSessionInner {
pub(crate) fn timeout(&self) -> Option<Duration> {
*self.shared.timeout.lock().expect("timeout poisoned")
}
fn find_conn(&self) -> Result<ConnGuard<'_>> {
self.find_conn_impl(ConnUse::Client)
}
fn find_conn_async(&self) -> Result<ConnGuard<'_>> {
self.find_conn_impl(ConnUse::ClientAsync)
}
fn find_conn_impl(&self, use_: ConnUse) -> Result<ConnGuard<'_>> {
let mut wait_until: Option<Option<Instant>> = None;
let tid = current_tid();
let pinned = self.driving_slot();
if let Some(slot_id) = pinned {
let reusable = {
let st = self.conn_state.lock().expect("conn_state poisoned");
match st.slots.iter().find(|s| s.id == slot_id) {
None => return Err(StatusCode::DeadObject),
Some(s)
if s.role == SlotRole::Outgoing
|| (s.allow_nested && use_ != ConnUse::ClientAsync)
|| !use_.is_new_transaction() =>
{
Some(Arc::clone(&s.transport))
}
Some(_) => None,
}
};
if let Some(transport) = reusable {
return Ok(ConnGuard {
inner: self,
slot_id,
transport,
reentrant: true,
});
}
}
let mut st = self.conn_state.lock().expect("conn_state poisoned");
loop {
if self.shared.lifecycle.is_torn_down() || st.slots.is_empty() {
return Err(StatusCode::DeadObject);
}
let free = |s: &ConnSlot| {
s.exclusive_tid.is_none() || (pinned.is_none() && s.exclusive_tid == Some(tid))
};
let pick = st
.slots
.iter()
.position(|s| free(s) && s.role == SlotRole::Outgoing);
if let Some(idx) = pick {
return Ok(self.claim_slot(&mut st.slots[idx], tid));
}
if !st.slots.iter().any(|s| s.role == SlotRole::Outgoing) {
if self.profile.wire_version().is_none() {
log::error!(
"RPC: r34 session has no outgoing connection — this endpoint \
accepted the connection, and the r34 profile cannot open or \
attach one. Only a nested twoway call (from inside a twoway \
handler) can transact here — a oneway never nests; use \
`WireProfile::Android13Plus` (`?profile=android13plus`) for oneway \
calls and for callbacks outside a handler"
);
} else {
log::error!(
"RPC: session has no outgoing connection — a non-nested call (from \
another thread, or any oneway) needs the peer to open \
incoming connections (RpcClientConfig::incoming_connections / \
ClientOptions::incoming_connections / \
ARpcSession_setMaxIncomingThreads); refusing instead of waiting forever"
);
}
return Err(StatusCode::WouldBlock);
}
let deadline = *self.shared.timeout.lock().expect("timeout poisoned");
let at = deadline.and_then(|d| {
*wait_until.get_or_insert_with(|| Instant::now().checked_add(d))
});
st = match (deadline, at) {
(Some(d), Some(at)) => {
let remaining = at.saturating_duration_since(Instant::now());
if remaining.is_zero() {
log::warn!(
"RPC: no connection slot became available within {d:?} \
(every slot is driven by another thread — a session \
served on one thread and transacted on another needs \
more than one connection); nothing was sent"
);
return Err(StatusCode::WouldBlock);
}
self.slot_cv
.wait_timeout(st, remaining)
.expect("slot_cv poisoned")
.0
}
_ => self.slot_cv.wait(st).expect("slot_cv poisoned"),
};
}
}
fn driving_slot(&self) -> Option<u64> {
let sess_ptr = self as *const RpcSessionInner as usize;
DRIVING.with(|d| {
d.borrow()
.iter()
.rev()
.find_map(|&(sp, sid)| if sp == sess_ptr { Some(sid) } else { None })
})
}
fn claim_slot(&self, slot: &mut ConnSlot, tid: Tid) -> ConnGuard<'_> {
slot.exclusive_tid = Some(tid);
let sess_ptr = self as *const RpcSessionInner as usize;
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot.id)));
ConnGuard {
inner: self,
slot_id: slot.id,
transport: Arc::clone(&slot.transport),
reentrant: false,
}
}
fn dec_route(&self) -> DecRoute<'_> {
let tid = current_tid();
let pinned = self.driving_slot();
let mut st = self.conn_state.lock().expect("conn_state poisoned");
let mut pending_on = None;
if let Some(slot_id) = pinned {
match st.slots.iter().find(|s| s.id == slot_id) {
Some(s) if s.role == SlotRole::Outgoing || s.allow_nested => {
return DecRoute::Conn(ConnGuard {
inner: self,
slot_id,
transport: Arc::clone(&s.transport),
reentrant: true,
});
}
Some(_) => pending_on = Some(slot_id),
None => {}
}
}
let free = st
.slots
.iter()
.position(|s| s.exclusive_tid.is_none() && s.role == SlotRole::Outgoing);
if let Some(idx) = free {
return DecRoute::Conn(self.claim_slot(&mut st.slots[idx], tid));
}
match pending_on {
Some(slot_id) => DecRoute::Pending(slot_id),
None => DecRoute::Reaper,
}
}
fn find_conn_for_reaper(&self) -> Option<ConnGuard<'_>> {
let tid = current_tid();
let mut st = self.conn_state.lock().expect("conn_state poisoned");
loop {
if self.shared.lifecycle.is_torn_down() || st.slots.is_empty() {
return None;
}
let free = |s: &ConnSlot| s.exclusive_tid == Some(tid) || s.exclusive_tid.is_none();
let pick = st
.slots
.iter()
.position(|s| free(s) && s.role == SlotRole::Outgoing);
if let Some(idx) = pick {
return Some(self.claim_slot(&mut st.slots[idx], tid));
}
if !st.slots.iter().any(|s| s.role == SlotRole::Outgoing) {
return None;
}
st = self.slot_cv.wait(st).expect("slot_cv poisoned");
}
}
fn find_conn_pinned(&self, want_slot_id: u64) -> Result<ConnGuard<'_>> {
let tid = current_tid();
let sess_ptr = self as *const RpcSessionInner as usize;
if DRIVING.with(|d| {
d.borrow()
.iter()
.any(|&(sp, sid)| sp == sess_ptr && sid == want_slot_id)
}) {
let transport = {
let st = self.conn_state.lock().expect("conn_state poisoned");
match st.slots.iter().find(|s| s.id == want_slot_id) {
Some(s) => Arc::clone(&s.transport),
None => return Err(StatusCode::DeadObject),
}
};
return Ok(ConnGuard {
inner: self,
slot_id: want_slot_id,
transport,
reentrant: true,
});
}
let mut st = self.conn_state.lock().expect("conn_state poisoned");
loop {
let target = st.slots.iter_mut().find(|s| s.id == want_slot_id);
let Some(slot) = target else {
return Err(StatusCode::DeadObject);
};
if slot.exclusive_tid.is_none() || slot.exclusive_tid == Some(tid) {
return Ok(self.claim_slot(slot, tid));
}
#[cfg(test)]
if let Some(tx) = &*self.shared.park_hook.lock().expect("park hook") {
let _ = tx.send(());
}
st = self.slot_cv.wait(st).expect("slot_cv poisoned");
}
}
fn clear_slot_read_timeout(&self, slot_id: u64) {
let transport = {
let st = self.conn_state.lock().expect("conn_state poisoned");
st.slots
.iter()
.find(|s| s.id == slot_id)
.map(|s| Arc::clone(&s.transport))
};
if let Some(t) = transport {
if let Err(e) = t.set_read_timeout(None) {
log::debug!("RPC: failed to clear first-frame read deadline: {e:?}");
}
}
}
fn activity(&self) -> u64 {
self.shared.io_gen.load(Ordering::Relaxed)
}
fn active_since(&self, seen: &mut u64) -> bool {
let open = self.shared.open.load(Ordering::SeqCst) > 0;
let now = self.activity();
let moved = now != *seen;
*seen = now;
open || moved
}
fn clear_handshake_timeouts(&self) {
for t in self.slot_transports() {
let _ = t.set_read_timeout(None);
self.arm_liveness(&*t);
}
}
fn slot_transports(&self) -> Vec<Arc<dyn RpcTransport>> {
let st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.iter().map(|s| Arc::clone(&s.transport)).collect()
}
fn arm_liveness(&self, transport: &dyn RpcTransport) {
let _serial = self.shared.liveness.lock().expect("liveness poisoned");
self.arm_liveness_locked(transport);
}
fn arm_liveness_locked(&self, transport: &dyn RpcTransport) {
let timeout = *self.shared.timeout.lock().expect("timeout poisoned");
if let Err(e) = transport.set_write_timeout(self.send_deadline()) {
log::warn!("RPC: failed to arm a connection's send deadline: {e:?}");
}
if let Err(e) = transport.set_liveness(timeout) {
log::warn!("RPC: failed to arm a connection's liveness check: {e:?}");
}
}
fn send_deadline(&self) -> Option<Duration> {
let timeout = *self.shared.timeout.lock().expect("timeout poisoned");
match (timeout, self.shared.serve_read_deadline()) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
}
}
fn arm_liveness_all(&self, store: impl FnOnce()) {
let _serial = self.shared.liveness.lock().expect("liveness poisoned");
store();
for t in self.slot_transports() {
self.arm_liveness_locked(&*t);
}
}
pub(crate) fn parcel_ops(&self) -> Arc<dyn RpcParcelOps> {
self.parcel_ops.clone()
}
fn add_slot_inner(&self, transport: Box<dyn RpcTransport>, role: SlotRole) -> Result<u64> {
let transport: Arc<dyn RpcTransport> = Arc::from(transport);
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if self.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
if !st.admits(&*transport) {
return Err(StatusCode::BadType);
}
let id = st.next_slot_id;
st.next_slot_id += 1;
let armed = Arc::clone(&transport);
st.slots.push(transport, id, role, None);
drop(st);
self.arm_liveness(&*armed);
self.slot_cv.notify_all();
Ok(id)
}
fn add_incoming_slot_capped(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
) -> Result<u64> {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if st
.slots
.iter()
.filter(|s| s.role == SlotRole::Incoming)
.count()
>= cap
{
return Err(StatusCode::FailedTransaction);
}
if !st.admits(&*transport) {
return Err(StatusCode::BadType);
}
if !self.shared.try_bump_live_conns() {
return Err(StatusCode::DeadObject);
}
let transport: Arc<dyn RpcTransport> = Arc::from(transport);
let id = st.next_slot_id;
st.next_slot_id += 1;
let armed = Arc::clone(&transport);
st.slots.push(transport, id, SlotRole::Incoming, None);
drop(st);
self.shared.io_gen.fetch_add(1, Ordering::Relaxed);
self.arm_liveness(&*armed);
self.slot_cv.notify_all();
Ok(id)
}
fn add_outgoing_slot(&self, transport: Box<dyn RpcTransport>) -> Result<u64> {
self.add_slot_inner(transport, SlotRole::Outgoing)
}
#[cfg(test)]
fn add_slot_inner_capped(&self, transport: Box<dyn RpcTransport>, cap: usize) -> Result<u64> {
self.push_outgoing_capped(transport, cap, false)
.map(|(id, _)| id)
}
fn add_claimed_slot_capped(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
) -> Result<SlotClaim<'_>> {
let (slot_id, transport) = self.push_outgoing_capped(transport, cap, true)?;
Ok(SlotClaim {
inner: self,
slot_id,
transport,
})
}
fn push_outgoing_capped(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
claimed: bool,
) -> Result<(u64, Arc<dyn RpcTransport>)> {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if self.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
if st
.slots
.iter()
.filter(|s| s.role == SlotRole::Outgoing)
.count()
>= cap
{
return Err(StatusCode::FailedTransaction);
}
if !st.admits(&*transport) {
return Err(StatusCode::BadType);
}
let transport: Arc<dyn RpcTransport> = Arc::from(transport);
let id = st.next_slot_id;
st.next_slot_id += 1;
let armed = Arc::clone(&transport);
st.slots
.push(transport, id, SlotRole::Outgoing, claimed.then(current_tid));
drop(st);
self.arm_liveness(&*armed);
self.slot_cv.notify_all();
Ok((id, armed))
}
fn end_after_failed_send(&self, e: &RpcError) {
if matches!(e, RpcError::FrameTooLarge { .. } | RpcError::Protocol(_)) {
return;
}
if !matches!(e, RpcError::EndOfStream) {
log::warn!("RPC: a send failed ({e}); ending the session");
}
self.fail_session();
}
fn send_error_status(e: RpcError) -> StatusCode {
match e {
RpcError::Io(io) if super::transport::is_timeout(&io) => StatusCode::TimedOut,
e => e.into(),
}
}
pub(crate) fn fail_session(&self) {
if self.shared.lifecycle.force_dying() {
self.on_session_dead();
}
}
pub(crate) fn close(&self) {
if self.shared.lifecycle.force_dying() {
self.shared.ended_locally.store(true, Ordering::SeqCst);
self.on_session_dead();
}
}
pub(crate) fn on_session_dead(&self) {
{
let _serial = self.shared.liveness.lock().expect("liveness poisoned");
self.shutdown_all_transports();
}
{
let mut st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.clear_at_session_end();
}
self.slot_cv.notify_all();
self.send_session_obituaries();
self.shared.lifecycle.mark_dead();
let root = self.shared.root.lock().expect("root poisoned").take();
let locals = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
st.clear_remote_sends();
st.clear_local()
};
drop(locals);
drop(root);
}
pub(crate) fn fd_mode(&self) -> FileDescriptorTransportMode {
self.shared.fd_mode()
}
pub(crate) fn records_fd_positions(&self) -> bool {
self.profile.records_fd_positions()
}
fn send_msg(
&self,
transport: &dyn RpcTransport,
frame: &[u8],
fds: &[OwnedFd],
drain: Option<&mut dyn FnMut() -> RpcResult<()>>,
) -> RpcResult<()> {
let _sending = OpenCall::enter(&self.shared);
if self.fd_mode() == FileDescriptorTransportMode::Unix {
let borrowed: Vec<_> = fds.iter().map(|f| f.as_fd()).collect();
return write_aosp_message_with_fds(transport, frame, &borrowed, drain);
}
debug_assert!(fds.is_empty(), "a non-Unix session must not carry fds");
let _ = fds; write_aosp_message_with_fds(transport, frame, &[], drain)
}
fn drain_one<'t>(
&self,
transport: &'t dyn RpcTransport,
slot_id: u64,
deadline: &mut Option<ReplyDeadlineGuard<'t>>,
after: &mut DrainedReleases,
fault: &std::cell::Cell<Option<StatusCode>>,
) -> RpcResult<()> {
if deadline.is_none() {
let baseline = self.slot_baseline_read_deadline(slot_id);
*deadline = Some(ReplyDeadlineGuard::arm(
transport,
self.send_deadline(),
baseline,
)?);
}
let refuse = |status: StatusCode, what: &'static str| {
log::error!(
"RPC: {what} arrived while a transaction was being sent; ending the session"
);
fault.set(Some(status));
Err(RpcError::Protocol(what))
};
let (frame, _fds) =
self.recv_msg_gated(transport, |header| {
match control_only_refusal(header, self.profile.dec_strong_body_len()) {
None => Ok(()),
Some((status, what)) => refuse(status, what),
}
})?;
match self.profile.codec().decode_message(&frame)? {
WireMessage::DecStrong(addr, amount) => {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
after.released.extend(st.dec_strong_local(&addr, amount));
let held = st.pay_proxy_sends(&addr, amount);
if held > 0 {
after.releases.push((addr, held));
}
Ok(())
}
WireMessage::Transact(_) => refuse(StatusCode::BadType, "a TRANSACT"),
WireMessage::Reply(_) => refuse(StatusCode::DeadObject, "a REPLY"),
}
}
fn recv_msg(&self, transport: &dyn RpcTransport) -> RpcResult<(Vec<u8>, Vec<OwnedFd>)> {
self.recv_msg_gated(transport, |_| Ok(()))
}
fn recv_msg_gated(
&self,
transport: &dyn RpcTransport,
header_gate: impl FnOnce(&[u8]) -> RpcResult<()>,
) -> RpcResult<(Vec<u8>, Vec<OwnedFd>)> {
let io_gen = &self.shared.io_gen;
if self.fd_mode() == FileDescriptorTransportMode::Unix {
return read_aosp_message_with_fds(
|buf| {
let got = transport.recv_raw_with_fds(buf)?;
if got.0 > 0 {
io_gen.fetch_add(1, Ordering::Relaxed);
}
Ok(got)
},
header_gate,
);
}
let mut io = CountedIo(RawTransportIo(transport), io_gen);
Ok((read_aosp_message_gated(&mut io, header_gate)?, Vec::new()))
}
fn read_session_preamble(&self, transport: &dyn RpcTransport) -> RpcResult<i32> {
let mut io = CountedIo(RawTransportIo(transport), &self.shared.io_gen);
read_r34_session_preamble(&mut io)
}
fn self_weak(&self) -> Weak<RpcSessionInner> {
self.self_weak.clone()
}
pub(crate) fn local_node_count(&self) -> usize {
self.shared.local_node_count()
}
pub(crate) fn live_conn_count(&self) -> usize {
self.shared.live_conn_count()
}
pub(crate) fn slot_count(&self) -> usize {
self.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.len()
}
pub(crate) fn callback_conn_count(&self) -> usize {
let callback_role = match self.shared.space() {
AddressSpace::Initiator => SlotRole::Incoming,
AddressSpace::Acceptor => SlotRole::Outgoing,
};
self.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.filter(|s| s.role == callback_role)
.count()
}
pub(crate) fn ring_activity(&self) -> SessionActivity {
SessionActivity(self.shared.clone())
}
pub(crate) fn caps(&self) -> crate::TransportCaps {
use crate::TransportCaps as C;
let traits = self.conn_state.lock().expect("conn_state poisoned").traits;
let mut caps = C::NONE;
if self.fd_mode() == FileDescriptorTransportMode::Unix && traits.passes_fds {
caps |= C::FD_PASSING;
}
if traits.local_peer {
caps |= C::TRUSTED_UID | C::SAME_HOST;
}
if self.callback_conn_count() > 0 {
caps |= C::CALLBACKS;
}
caps
}
fn slot_role(&self, slot_id: u64) -> Option<SlotRole> {
self.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.find(|s| s.id == slot_id)
.map(|s| s.role)
}
fn slot_baseline_read_deadline(&self, slot_id: u64) -> Option<Duration> {
let idle = self.shared.serve_read_deadline()?;
let serve_driven = self
.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.any(|s| s.id == slot_id && s.role == SlotRole::Incoming);
serve_driven.then_some(idle)
}
fn handler_read_deadline(&self) -> Option<Duration> {
let idle = self.shared.serve_read_deadline()?;
let sess_ptr = self as *const RpcSessionInner as usize;
let st = self.conn_state.lock().expect("conn_state poisoned");
let serving = DRIVING.with(|d| {
d.borrow().iter().any(|&(sp, sid)| {
sp == sess_ptr
&& st
.slots
.iter()
.any(|s| s.id == sid && s.role == SlotRole::Incoming)
})
});
serving.then_some(idle)
}
pub(crate) fn notices_connection_loss(&self) -> bool {
self.shared.space() == AddressSpace::Acceptor
|| self.shared.serve_declared.load(Ordering::SeqCst) > 0
|| self.incoming_slot_count() > 0
}
fn incoming_slot_count(&self) -> usize {
self.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.filter(|s| s.role == SlotRole::Incoming)
.count()
}
pub(crate) fn peer_closed(&self) -> bool {
let transports: Vec<Arc<dyn RpcTransport>> = self
.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.map(|s| Arc::clone(&s.transport))
.collect();
transports.iter().any(|t| t.peer_closed() == Some(true))
}
fn shutdown_all_transports(&self) {
let transports: Vec<Arc<dyn RpcTransport>> = self
.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.map(|s| Arc::clone(&s.transport))
.collect();
for t in transports {
if let Err(e) = t.shutdown() {
log::warn!("RPC: shutting a connection's transport down failed: {e}");
}
}
}
fn take_incoming_threads(&self) -> Vec<(u64, std::thread::JoinHandle<()>)> {
std::mem::take(
&mut *self
.incoming_threads
.lock()
.expect("incoming_threads poisoned"),
)
}
pub(crate) fn set_r34_session_id(&self, id: i32) {
self.shared.r34_session_id.store(id, Ordering::SeqCst);
}
pub(crate) fn max_threads_value(&self) -> u32 {
self.shared.max_threads.load(Ordering::Relaxed)
}
pub(crate) fn wire_protocol_version(&self) -> Option<u32> {
self.profile.wire_version()
}
fn wire_write_binder_addr(&self, p: &mut Parcel, addr: &RpcAddress) -> Result<()> {
match &self.profile {
WireProfile::R34(_) => write_addr(p, addr),
WireProfile::Android13Plus(_) => {
p.write_aligned_data(&Android13PlusCodec::encode_addr(addr))
}
}
}
fn wire_read_binder_addr(&self, p: &mut Parcel) -> Result<RpcAddress> {
match &self.profile {
WireProfile::R34(_) => read_addr(p),
WireProfile::Android13Plus(_) => {
let slice = p.read_aligned_data(A13_ADDR_LEN)?;
Android13PlusCodec::decode_addr(slice, 0).map_err(StatusCode::from)
}
}
}
fn write_binder(&self, binder: Option<&SIBinder>, parcel: &mut Parcel) -> Result<()> {
match binder {
None => {
parcel.write(&0i32)?;
parcel.write(&self.profile.binder_stability_repr(None))?;
Ok(())
}
Some(b) => {
let addr = if let Some(rp) = (**b).as_any().downcast_ref::<RpcProxy>() {
if !std::ptr::eq(rp.session_ptr(), self) {
log::error!("RPC: cannot send a binder from an unrelated RPC session");
return Err(StatusCode::InvalidOperation);
}
if !parcel.rpc_is_unsent() {
log::error!("RPC: cannot write a binder into a sent or received parcel");
return Err(StatusCode::InvalidOperation);
}
let addr = rp.address();
if self.profile.records_binder_positions() {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.on_proxy_leaving(addr);
parcel.rpc_record_leaving_addr(addr);
}
parcel.rpc_pin_binder(b.clone());
addr
} else if (**b).is_remote() {
log::error!("RPC: cannot send a kernel binder proxy over sockets");
return Err(StatusCode::InvalidOperation);
} else {
if !parcel.rpc_is_unsent() {
log::error!("RPC: cannot write a local binder into a sent parcel");
return Err(StatusCode::BadType);
}
let addr = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
if self.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
st.on_binder_leaving(b)?
};
parcel.rpc_record_leaving_addr(addr);
addr
};
let obj_pos = parcel.data_position();
parcel.write(&1i32)?;
self.wire_write_binder_addr(parcel, &addr)?;
if self.profile.records_binder_positions() {
parcel.rpc_record_object_position(obj_pos);
}
parcel.write(&self.profile.binder_stability_repr(Some(b.stability())))?;
b.set_parceled();
Ok(())
}
}
}
fn read_binder(&self, parcel: &mut Parcel) -> Result<Option<SIBinder>> {
let obj_pos = parcel.data_position();
let present: i32 = parcel.read()?;
if present == 0 {
let stability: i32 = parcel.read()?;
self.profile.check_binder_stability(stability, true)?;
return Ok(None);
}
if self.profile.records_binder_positions() && !parcel.rpc_object_position_present(obj_pos) {
return Err(StatusCode::BadValue);
}
let addr = self.wire_read_binder_addr(parcel)?;
let binder = self.binder_at(parcel, obj_pos, addr)?;
let stability: i32 = parcel.read()?;
self.profile.check_binder_stability(stability, false)?;
Ok(Some(binder))
}
fn binder_at(&self, parcel: &mut Parcel, obj_pos: usize, addr: RpcAddress) -> Result<SIBinder> {
if !parcel.rpc_received_at(obj_pos) {
return self.lookup_binder(addr);
}
if let Some((entered, binder)) = parcel.rpc_entered_at(obj_pos) {
if entered != addr {
log::error!("RPC: position {obj_pos} entered {entered:?}, now reads {addr:?}");
return Err(StatusCode::BadValue);
}
return Ok(binder);
}
let binder = self.enter_binder(addr)?;
parcel.rpc_record_entered(obj_pos, addr, binder.clone());
Ok(binder)
}
fn lookup_binder(&self, addr: RpcAddress) -> Result<SIBinder> {
let found = {
let st = self.shared.state.lock().expect("rpc state poisoned");
st.lookup_local(&addr).or_else(|| st.lookup_remote(&addr))
};
found.ok_or_else(|| {
log::error!("RPC: {addr:?} names no live binder in a parcel that was not received");
StatusCode::BadValue
})
}
fn enter_binder(&self, addr: RpcAddress) -> Result<SIBinder> {
let strong = self.self_weak().upgrade().ok_or(StatusCode::DeadObject)?;
let (binder, owed) = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
if let Some(local) = st.lookup_local(&addr) {
(local, true)
} else {
if self.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
st.remote_proxy(addr, || {
SIBinder::new(Arc::new(RpcProxy::new(addr, strong)))
.expect("SIBinder::new(RpcProxy)")
})?
}
};
if owed {
self.send_dec_strong(addr, 1);
}
Ok(binder)
}
fn receive_parcel(
&self,
data: Vec<u8>,
object_positions: Vec<u32>,
in_fds: Vec<OwnedFd>,
) -> (Parcel, Result<()>) {
let mut p = Parcel::from_vec(data);
p.configure_rpc(
self.parcel_ops(),
self.fd_mode(),
self.records_fd_positions(),
);
p.rpc_set_in_fds(in_fds);
p.rpc_set_object_positions(object_positions);
p.rpc_mark_received();
let entered = if self.profile.records_binder_positions() {
self.enter_every_binder(&mut p)
} else {
Ok(())
};
p.set_data_position(0);
(p, entered)
}
fn enter_every_binder(&self, p: &mut Parcel) -> Result<()> {
use super::wire_android13::TYPE_NATIVE_FILE_DESCRIPTOR;
const TYPE_BINDER: i32 = 1;
let mut result = Ok(());
let positions = p.rpc_object_positions().to_vec();
for pos in positions {
p.set_data_position(pos as usize);
let entered = match p.read::<i32>() {
Ok(TYPE_BINDER) => {
p.set_data_position(pos as usize);
self.read_binder(p).map(drop)
}
Ok(TYPE_NATIVE_FILE_DESCRIPTOR) => Ok(()),
Ok(other) => {
log::error!("RPC: object position {pos} holds unknown type {other}");
Err(StatusCode::BadValue)
}
Err(e) => Err(e),
};
if let (Ok(()), Err(e)) = (&result, entered) {
result = Err(e);
}
}
result
}
fn acquire_copied(&self, objects: &[&[u8]]) -> Result<CopiedBinders> {
let addrs = objects
.iter()
.map(|bytes| self.decode_copied_addr(bytes))
.collect::<Result<Vec<_>>>()?;
let mut took = CopiedBinders::default();
let mut release: Vec<SIBinder> = Vec::new();
let mut releases: Vec<(RpcAddress, u32)> = Vec::new();
let result = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
let result = self.acquire_copied_locked(&mut st, &addrs, &mut took, &mut release);
if result.is_err() {
for addr in took.leaving.drain(..) {
let (node, held) = st.cancel_leaving(&addr);
release.extend(node);
releases.push((addr, held));
}
}
result
};
drop(release);
self.send_held_releases(releases);
result.map(|()| took)
}
fn acquire_copied_locked(
&self,
st: &mut RpcState,
addrs: &[RpcAddress],
took: &mut CopiedBinders,
release: &mut Vec<SIBinder>,
) -> Result<()> {
if self.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
for addr in addrs {
if let Some(local) = st.lookup_local(addr) {
let leaving = st.on_binder_leaving(&local);
release.push(local);
let leaving = leaving?;
if leaving != *addr {
log::error!("RPC: copied address {addr:?} names a node at {leaving:?}");
release.extend(st.cancel_binder_leaving(&leaving));
return Err(StatusCode::BadValue);
}
took.leaving.push(leaving);
} else if addr.is_zero() || addr.minted_by(self.shared.space) {
log::error!("RPC: a copied binder names no node of this session: {addr:?}");
return Err(StatusCode::BadValue);
} else if let Some(proxy) = st.lookup_remote(addr) {
if self.profile.records_binder_positions() {
st.on_proxy_leaving(*addr);
took.leaving.push(*addr);
}
took.pinned.push(proxy);
}
}
Ok(())
}
fn decode_copied_addr(&self, bytes: &[u8]) -> Result<RpcAddress> {
match &self.profile {
WireProfile::R34(_) => {
let raw = bytes
.get(..RPC_ADDR_LEN)
.and_then(|b| <[u8; RPC_ADDR_LEN]>::try_from(b).ok())
.ok_or(StatusCode::BadValue)?;
Ok(RpcAddress::from_wire_bytes(raw))
}
WireProfile::Android13Plus(_) => {
Android13PlusCodec::decode_addr(bytes, 0).map_err(StatusCode::from)
}
}
}
fn cancel_leaving(&self, addrs: &[RpcAddress]) {
if addrs.is_empty() {
return;
}
let (released, releases): (Vec<SIBinder>, Vec<(RpcAddress, u32)>) = {
let mut st = self
.shared
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut released = Vec::new();
let mut releases = Vec::new();
for a in addrs {
let (node, held) = st.cancel_leaving(a);
released.extend(node);
releases.push((*a, held));
}
(released, releases)
};
drop(released);
self.send_held_releases(releases);
}
fn send_held_releases(&self, releases: Vec<(RpcAddress, u32)>) {
for (addr, amount) in releases {
self.send_dec_strong(addr, amount);
}
}
fn apply_dec_strong(&self, addr: RpcAddress, amount: u32) {
let (released, held) = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
(
st.dec_strong_local(&addr, amount),
st.pay_proxy_sends(&addr, amount),
)
};
drop(released);
self.send_dec_strong(addr, held);
}
pub(crate) fn release_proxy(&self, addr: RpcAddress, who: *const ()) {
let now = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.release_proxy(&addr, who);
self.send_dec_strong(addr, now);
}
fn cancel_oneway_number(&self, addr: RpcAddress, consumed: u64) {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.cancel_send_async_number(addr, consumed);
}
pub(crate) fn client_transact(
&self,
addr: RpcAddress,
code: u32,
data: &Parcel,
flags: u32,
) -> Result<Option<Parcel>> {
let oneway = (flags & FLAG_ONEWAY) != 0;
self.check_parcel_session(data)?;
data.rpc_begin_send()?;
let found = if oneway {
self.find_conn_async()
} else {
self.find_conn()
};
let conn = match found {
Ok(conn) => conn,
Err(e) => {
data.rpc_end_send(false);
return Err(e);
}
};
let transport = conn.transport();
let counts_target = !addr.is_zero() && self.profile.counts_transaction_targets();
let async_number = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
if counts_target {
st.on_proxy_leaving(addr);
}
if oneway {
st.next_send_async_number(addr)
} else {
0
}
};
let txn = WireTransactionRef {
address: &addr,
code,
flags,
async_number,
data: data.rpc_data_bytes(),
object_positions: data.rpc_object_positions(),
};
let rollback = || {
if oneway {
self.cancel_oneway_number(addr, async_number);
}
data.rpc_end_send(false);
if counts_target {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
st.pay_proxy_sends(&addr, 1)
} else {
0
}
};
let frame = match self.profile.codec().encode_transact_ref(txn) {
Ok(frame) => frame,
Err(e) => {
let held = rollback();
self.send_dec_strong(addr, held);
return Err(e.into());
}
};
let _open = (!oneway).then(|| OpenCall::enter(&self.shared));
let mut drained = DrainedReleases::default();
let mut drain_deadline = None;
let fault = std::cell::Cell::new(None);
let sent = self.send_msg(
transport,
&frame,
data.rpc_out_fds(),
Some(&mut || {
let slot = conn.slot_id;
self.drain_one(transport, slot, &mut drain_deadline, &mut drained, &fault)
}),
);
drop(drain_deadline);
let DrainedReleases { released, releases } = drained;
if let Err(e) = sent {
let held = rollback();
self.end_after_failed_send(&e);
drop(released);
self.send_held_releases(releases);
self.send_dec_strong(addr, held);
return Err(fault.get().unwrap_or_else(|| Self::send_error_status(e)));
}
drop(released);
self.send_held_releases(releases);
data.rpc_end_send(true);
if oneway {
return Ok(None);
}
let fail = |status: StatusCode| {
self.fail_session();
Err(status)
};
let baseline = self.slot_baseline_read_deadline(conn.slot_id);
let deadline = self.timeout().or_else(|| self.handler_read_deadline());
let _deadline_guard = match ReplyDeadlineGuard::arm(transport, deadline, baseline) {
Ok(g) => g,
Err(e) => return fail(e.into()),
};
loop {
let (frame, in_fds) = match self.recv_msg(transport) {
Ok(v) => v,
Err(e) => return fail(e.into()),
};
let message = match self.profile.codec().decode_message(&frame) {
Ok(m) => m,
Err(e) => return fail(e.into()),
};
match message {
WireMessage::Reply(WireReply {
status,
data,
object_positions,
}) => {
if status != 0 {
return Err(StatusCode::from(status));
}
let (reply, entered) = self.receive_parcel(data, object_positions, in_fds);
entered?;
return Ok(Some(reply));
}
WireMessage::DecStrong(a, amount) => self.apply_dec_strong(a, amount),
WireMessage::Transact(t) => {
let _restore = match NestedDeadlineGuard::lift(transport, deadline) {
Ok(g) => g,
Err(e) => return fail(e.into()),
};
let peer = transport.peer_identity();
if let Err(e) = self.dispatch_transact(t, in_fds, peer) {
return fail(e);
}
}
}
}
}
pub(crate) fn send_dec_strong(&self, addr: RpcAddress, amount: u32) {
if amount == 0 || self.shared.lifecycle.is_torn_down() {
return;
}
match self.dec_route() {
DecRoute::Conn(conn) => {
let _ = self.write_dec_strong(&conn, addr, amount);
}
DecRoute::Pending(slot_id) => match self.held_past_limit(slot_id, &addr) {
Some(conn) => {
let _ = self.write_dec_strong(&conn, addr, amount);
}
None => self.hold_dec_strong(slot_id, addr, amount),
},
DecRoute::Reaper => {
let _ = self.dec_strong_tx.send((addr, amount));
}
}
}
fn write_dec_strong(
&self,
conn: &ConnGuard<'_>,
addr: RpcAddress,
amount: u32,
) -> RpcResult<()> {
let Some((frame, times)) = self.profile.codec().encode_dec_strong(&addr, amount) else {
return Ok(());
};
for _ in 0..times {
if let Err(e) = self.send_msg(conn.transport(), &frame, &[], None) {
self.end_after_failed_send(&e);
return Err(e);
}
}
Ok(())
}
fn held_past_limit(&self, slot_id: u64, addr: &RpcAddress) -> Option<ConnGuard<'_>> {
let st = self.conn_state.lock().expect("conn_state poisoned");
let s = st.slots.iter().find(|s| s.id == slot_id)?;
if s.pending_dec.len() < HELD_DEC_STRONG_LIMIT || s.pending_dec.contains_key(addr) {
return None;
}
Some(ConnGuard {
inner: self,
slot_id,
transport: Arc::clone(&s.transport),
reentrant: true,
})
}
fn hold_dec_strong(&self, slot_id: u64, addr: RpcAddress, amount: u32) {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
let Some(slot) = st.slots.iter_mut().find(|s| s.id == slot_id) else {
return;
};
let owed = slot.pending_dec.entry(addr).or_insert(0);
*owed = owed.saturating_add(amount);
}
fn take_held_dec_strongs(&self, slot_id: u64) -> std::collections::HashMap<RpcAddress, u32> {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
st.slots
.iter_mut()
.find(|s| s.id == slot_id)
.map(|s| std::mem::take(&mut s.pending_dec))
.unwrap_or_default()
}
pub(crate) fn send_session_obituaries(&self) {
let snapshot = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.remote_proxy_snapshot();
for arc in snapshot {
let Some(proxy) = arc.as_any().downcast_ref::<RpcProxy>() else {
continue;
};
let sib = SIBinder::from_arc(arc.clone());
let who = SIBinder::downgrade(&sib);
proxy.send_obituary(&who);
}
}
fn send_reply(
&self,
status: i32,
data: &[u8],
object_positions: &[u32],
fds: &[OwnedFd],
) -> Result<()> {
self.write_reply(status, data, object_positions, fds)
.map_err(|e| match e {
ReplyNotSent::Refused(s) | ReplyNotSent::Failed(s) => s,
})
}
fn write_reply(
&self,
status: i32,
data: &[u8],
object_positions: &[u32],
fds: &[OwnedFd],
) -> std::result::Result<(), ReplyNotSent> {
let frame = self
.profile
.codec()
.encode_reply_ref(WireReplyRef {
status,
data,
object_positions,
})
.map_err(ReplyNotSent::refused)?;
let conn = self
.find_conn_impl(ConnUse::Reply)
.map_err(ReplyNotSent::Failed)?;
for (addr, amount) in self.take_held_dec_strongs(conn.slot_id) {
if let Err(e) = self.write_dec_strong(&conn, addr, amount) {
if !matches!(e, RpcError::FrameTooLarge { .. } | RpcError::Protocol(_)) {
return Err(ReplyNotSent::Failed(Self::send_error_status(e)));
}
}
}
if let Err(e) = self.send_msg(conn.transport(), &frame, fds, None) {
self.end_after_failed_send(&e);
return Err(match e {
RpcError::FrameTooLarge { .. } | RpcError::Protocol(_) => ReplyNotSent::refused(e),
e => ReplyNotSent::Failed(Self::send_error_status(e)),
});
}
Ok(())
}
fn check_parcel_session(&self, parcel: &Parcel) -> Result<()> {
match parcel.rpc_session_id() {
Some(id) if id != (self as *const Self).cast() => {
log::error!("RPC: the parcel was built for another session");
Err(StatusCode::BadType)
}
_ => Ok(()),
}
}
fn check_reply_parcel(&self, reply: &Parcel) -> Result<()> {
if reply.rpc_session_id().is_none() {
log::error!("RPC: the reply was built for no session; write into the one handed in");
return Err(StatusCode::BadType);
}
self.check_parcel_session(reply)?;
let fds = reply.rpc_out_fds().len();
if fds > super::transport::unix::MAX_FDS_PER_FRAME {
log::error!("RPC: a reply carries {fds} fds, over the per-frame limit");
return Err(StatusCode::BadValue);
}
Ok(())
}
fn send_reply_parcel(&self, reply: &Parcel) -> Result<()> {
if let Err(e) = self
.check_reply_parcel(reply)
.and_then(|()| reply.rpc_begin_send())
{
return self.send_reply(e.into(), &[], &[], &[]);
}
let r = self.write_reply(
0,
reply.rpc_data_bytes(),
reply.rpc_object_positions(),
reply.rpc_out_fds(),
);
reply.rpc_end_send(r.is_ok());
match r {
Ok(()) => Ok(()),
Err(ReplyNotSent::Refused(s)) => self.send_reply(s.into(), &[], &[], &[]),
Err(ReplyNotSent::Failed(s)) => Err(s),
}
}
fn dispatch_transact(
&self,
t: WireTransaction,
in_fds: Vec<OwnedFd>,
peer: PeerIdentity,
) -> Result<()> {
let _open = OpenCall::enter(&self.shared);
let oneway = (t.flags & FLAG_ONEWAY) != 0;
if t.address.is_zero() {
return self.serve_special(&t, oneway);
}
let peer = Arc::new(peer);
if oneway {
self.dispatch_oneway_ordered(t, in_fds, peer)
} else {
self.execute_dispatched(t, in_fds, false, peer)
}
}
fn dispatch_oneway_ordered(
&self,
t: WireTransaction,
in_fds: Vec<OwnedFd>,
peer: Arc<PeerIdentity>,
) -> Result<()> {
let addr = t.address;
let wire_async = t.async_number;
let counts_targets = self.profile.counts_transaction_targets();
let mut owed: u32 = 0;
let decision = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.dispatch_async_or_enqueue(addr, wire_async, t, in_fds);
let mut next = match decision {
super::state::AsyncDecision::Dispatch(t, fds) => {
owed = 1;
Some((t, fds))
}
super::state::AsyncDecision::Enqueued => {
log::trace!(
"RPC oneway parked: addr={:?} async#={} (out of order)",
addr,
wire_async
);
None
}
super::state::AsyncDecision::Drop(reason) => {
log::debug!(
"RPC oneway dropped: addr={:?} async#={} reason={:?}",
addr,
wire_async,
reason
);
if matches!(reason, super::state::DropReason::StaleAsyncNumber) {
owed = 1;
}
None
}
super::state::AsyncDecision::Terminate(num_pending) => {
log::error!(
"RPC: {num_pending} pending oneway transactions on {addr:?}; \
flushing backlog and ending the session"
);
if counts_targets {
let flushed = u32::try_from(num_pending).unwrap_or(u32::MAX);
self.send_dec_strong(addr, flushed);
}
return Err(StatusCode::FailedTransaction);
}
};
let mut result = Ok(());
while let Some((t, fds)) = next {
if let Err(e) = self.execute_dispatched(t, fds, true, Arc::clone(&peer)) {
result = Err(e);
break;
}
let advance = {
let mut state = self.shared.state.lock().expect("rpc state poisoned");
state.advance_and_pop_async(addr)
};
owed = owed.saturating_add(advance.purged);
if advance.next.is_some() {
owed = owed.saturating_add(1);
}
next = advance.next;
}
if counts_targets {
self.send_dec_strong(addr, owed);
}
result
}
fn execute_dispatched(
&self,
t: WireTransaction,
in_fds: Vec<OwnedFd>,
oneway: bool,
peer: Arc<PeerIdentity>,
) -> Result<()> {
let target = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.lookup_local(&t.address);
let Some(target) = target else {
if oneway {
log::debug!(
"RPC oneway to unknown/released address {:?} dropped",
t.address
);
} else {
self.send_reply(StatusCode::DeadObject.into(), &[], &[], &[])?;
}
return Ok(());
};
if !oneway {
if self.profile.pays_target_before_reply() {
if let Some(slot_id) = self.driving_slot() {
self.hold_dec_strong(slot_id, t.address, 1);
}
}
match t.code {
INTERFACE_TRANSACTION => {
let mut reply = Parcel::new();
reply.attach_rpc_ops(self.parcel_ops());
reply.write(&target.descriptor())?;
return self.send_reply_parcel(&reply);
}
PING_TRANSACTION => {
return self.send_reply(0, &[], &[], &[]);
}
_ => {}
}
}
let _nested = AllowNestedGuard::arm(self, !oneway);
let (mut reader, entered) = self.receive_parcel(t.data, t.object_positions, in_fds);
if let Err(e) = entered {
log::error!("RPC: a transaction's binder could not be entered ({e:?}); refused");
return if oneway {
Ok(())
} else {
self.send_reply(StatusCode::BadValue.into(), &[], &[], &[])
};
}
let mut reply = if oneway {
Parcel::with_capacity(0)
} else {
Parcel::new()
};
reply.configure_rpc(
self.parcel_ops(),
self.fd_mode(),
self.records_fd_positions(),
);
let caps = self.caps();
let code = t.code;
let observed_ctx = || {
let (calling_uid, calling_pid) = crate::thread_state::peer_uid_pid(&peer);
crate::observe::TxnContext {
descriptor: target.descriptor(),
code,
method: target
.as_transactable()
.and_then(|transactable| transactable.transaction_name(code)),
is_oneway: oneway,
calling_uid,
calling_pid,
transport: caps,
}
};
let result = {
let _calling = crate::thread_state::RpcCallingGuard::install(Arc::clone(&peer), caps);
let _work_source = crate::thread_state::WorkSourceDispatchGuard::enter();
crate::observe::observed(observed_ctx, || {
consume_rpc_interface_token(&mut reader, target.descriptor()).and_then(|()| {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
target.rpc_transact(code, &mut reader, &mut reply)
}))
.unwrap_or_else(|payload| {
let msg = payload
.downcast_ref::<&'static str>()
.copied()
.or_else(|| payload.downcast_ref::<String>().map(String::as_str))
.unwrap_or("<non-string panic payload>");
log::error!("RPC on_transact panicked for code {code}: {msg}");
Err(crate::StatusCode::Unknown)
})
})
})
};
drop(reader);
if oneway {
if let Err(e) = result {
log::error!("oneway RPC transaction failed (dropped): {e:?}");
}
return Ok(());
}
match result {
Ok(()) => self.send_reply_parcel(&reply),
Err(e) => self.send_reply(crate::binder::handler_reply_status(e).into(), &[], &[], &[]),
}
}
fn serve_once_on_slot(&self, slot_id: u64) -> ServeStep {
let Ok(conn) = self.find_conn_pinned(slot_id) else {
return ServeStep::Ended(EndReason::SessionEnded);
};
let transport = conn.transport();
if slot_id == RpcSession::FOUNDING_SLOT_ID
&& self.awaits_preamble.swap(false, Ordering::SeqCst)
{
match self.read_session_preamble(transport) {
Ok(RPC_SESSION_ID_NEW) => {}
Ok(id) => {
log::error!(
"RPC r34: the client asked to join session id {id}; joining an existing \
session needs an `RpcServer`, so this connection is closed"
);
return ServeStep::Ended(EndReason::Frame(
RpcError::Protocol("r34 preamble names an existing session").into(),
));
}
Err(e) => return ServeStep::Ended(recv_end_reason(e)),
}
}
let (frame, in_fds) = match self.recv_msg(transport) {
Ok(f) => f,
Err(e) => return ServeStep::Ended(recv_end_reason(e)),
};
if self.shared.ended_locally.load(Ordering::SeqCst) {
return ServeStep::Ended(EndReason::Interrupted);
}
let msg = match self.profile.codec().decode_message(&frame) {
Ok(m) => m,
Err(e) => return ServeStep::Ended(EndReason::Frame(e.into())),
};
match msg {
WireMessage::Transact(t) => {
let peer = transport.peer_identity();
match self.dispatch_transact(t, in_fds, peer) {
Ok(()) => ServeStep::Continue,
Err(e) => ServeStep::Ended(EndReason::Dispatch(e)),
}
}
WireMessage::DecStrong(a, amount) => {
self.apply_dec_strong(a, amount);
ServeStep::Continue
}
WireMessage::Reply(_) => {
log::warn!("RPC server received an unexpected REPLY; ending the session");
ServeStep::Ended(EndReason::Frame(StatusCode::BadType))
}
}
}
fn serve_special(&self, t: &WireTransaction, oneway: bool) -> Result<()> {
if oneway {
return Ok(());
}
match SpecialTransaction::from_code(t.code) {
Some(SpecialTransaction::GetRoot) => {
let root = self.shared.root.lock().expect("root poisoned").clone();
let mut reply = Parcel::new();
reply.attach_rpc_ops(self.parcel_ops());
reply.write(&root)?;
self.send_reply_parcel(&reply)
}
Some(SpecialTransaction::GetMaxThreads) => {
let n = self.shared.max_threads.load(Ordering::SeqCst) as i32;
let mut reply = Parcel::new();
reply.write(&n)?;
self.send_reply(0, reply.rpc_data_bytes(), &[], &[])
}
Some(SpecialTransaction::GetSessionId) => {
let mut reply = Parcel::new();
if self.profile.wire_version().is_some() {
reply.write(&self.shared.rpc_session_id.as_bytes()[..])?;
} else {
let id = self.shared.r34_session_id.load(Ordering::SeqCst);
if id == RPC_SESSION_ID_NEW {
let status = StatusCode::UnknownTransaction.into();
return self.send_reply(status, &[], &[], &[]);
}
reply.write(&id)?;
}
self.send_reply(0, reply.rpc_data_bytes(), &[], &[])
}
Some(SpecialTransaction::GetFdMode) if self.profile.wire_version().is_some() => {
self.send_reply(StatusCode::UnknownTransaction.into(), &[], &[], &[])
}
Some(SpecialTransaction::GetFdMode) => {
if self.fd_mode() == FileDescriptorTransportMode::Unix {
let mut reply = Parcel::new();
reply.write(&1i32)?;
return self.send_reply(0, reply.rpc_data_bytes(), &[], &[]);
}
let mut req = Parcel::from_vec(t.data.clone());
req.set_data_position(0);
let want_unix = match req.read::<i32>() {
Ok(v) => v == 1,
Err(e) => {
log::debug!("RPC GET_FD_MODE: malformed body ({e:?}); defaulting to None");
false
}
};
let agreed = if want_unix && self.shared.fd_unix_supported.load(Ordering::SeqCst) {
FileDescriptorTransportMode::Unix
} else {
FileDescriptorTransportMode::None
};
let mut reply = Parcel::new();
reply.write(
&(if agreed == FileDescriptorTransportMode::Unix {
1i32
} else {
0i32
}),
)?;
self.send_reply(0, reply.rpc_data_bytes(), &[], &[])?;
self.shared.set_fd_mode(agreed);
Ok(())
}
None => self.send_reply(StatusCode::UnknownTransaction.into(), &[], &[], &[]),
}
}
}
fn recv_end_reason(e: RpcError) -> EndReason {
match e {
RpcError::EndOfStream => EndReason::EndOfStream,
RpcError::UncleanEndOfStream => EndReason::UncleanEndOfStream,
RpcError::DeadlineMidFrame => EndReason::DeadlineMidFrame,
RpcError::Io(e) if e.kind() == std::io::ErrorKind::TimedOut => {
EndReason::Frame(StatusCode::DeadObject)
}
e => EndReason::Frame(e.into()),
}
}
#[derive(Clone)]
pub struct RpcSession {
inner: Arc<RpcSessionInner>,
}
impl RpcSession {
pub fn new(transport: Box<dyn RpcTransport>, space: AddressSpace) -> RpcResult<RpcSession> {
match space {
AddressSpace::Initiator => {
transport.send_raw(&R34Codec.encode_session_preamble(RPC_SESSION_ID_NEW))?;
Self::new_accepted(transport, space)
}
AddressSpace::Acceptor => {
let session = Self::new_accepted(transport, space)?;
session.inner.awaits_preamble.store(true, Ordering::SeqCst);
Ok(session)
}
}
}
pub(crate) fn new_accepted(
transport: Box<dyn RpcTransport>,
space: AddressSpace,
) -> RpcResult<RpcSession> {
RpcSession::with_profile(transport, space, WireProfile::R34(R34Codec))
}
fn with_profile(
transport: Box<dyn RpcTransport>,
space: AddressSpace,
profile: WireProfile,
) -> RpcResult<RpcSession> {
Ok(Self::with_shared(
transport,
profile,
Self::fresh_shared(space)?,
))
}
fn fresh_shared(space: AddressSpace) -> RpcResult<Arc<SharedSession>> {
Ok(Arc::new(SharedSession {
state: Mutex::new(RpcState::new(space)),
root: Mutex::new(None),
max_threads: AtomicU32::new(1),
negotiated: AtomicU32::new(0),
timeout: Mutex::new(None),
serve_read_deadline: AtomicU64::new(0),
fd_mode: AtomicU8::new(FD_MODE_NONE),
fd_unix_supported: AtomicBool::new(false),
rpc_session_id: gen_rpc_session_id()?,
r34_session_id: AtomicI32::new(RPC_SESSION_ID_NEW),
server_session_id: Mutex::new(None),
lifecycle: SessionLifecycle::new(),
ended_locally: AtomicBool::new(false),
serve_declared: AtomicUsize::new(0),
space,
open: AtomicUsize::new(0),
io_gen: AtomicU64::new(0),
liveness: Mutex::new(()),
#[cfg(test)]
park_hook: Mutex::new(None),
#[cfg(test)]
cci_failed_hook: Mutex::new(None),
#[cfg(test)]
serve_wait_hook: Mutex::new(None),
#[cfg(test)]
fail_incoming_spawn: AtomicBool::new(false),
}))
}
fn with_shared(
transport: Box<dyn RpcTransport>,
profile: WireProfile,
shared: Arc<SharedSession>,
) -> RpcSession {
let founding_role = match shared.space() {
AddressSpace::Initiator => SlotRole::Outgoing,
AddressSpace::Acceptor => SlotRole::Incoming,
};
let traits = ConnTraits::of(&*transport);
let founding: Arc<dyn RpcTransport> = Arc::from(transport);
let armed = Arc::clone(&founding);
let (dec_strong_tx, dec_strong_rx) = mpsc::channel();
let inner = Arc::new_cyclic(|weak: &Weak<RpcSessionInner>| RpcSessionInner {
conn_state: Mutex::new(ConnState::new(founding, founding_role, traits)),
slot_cv: Condvar::new(),
profile,
self_weak: weak.clone(),
parcel_ops: Arc::new(SessionParcelOps(weak.clone())),
shared,
dec_strong_tx,
incoming_threads: Mutex::new(Vec::new()),
incoming_live: AtomicUsize::new(0),
incoming_joined: AtomicUsize::new(0),
awaits_preamble: AtomicBool::new(false),
});
inner.arm_liveness(&*armed);
let weak_for_reaper = Arc::downgrade(&inner);
if let Err(e) = std::thread::Builder::new()
.name("rsbinder-rpc-reaper".into())
.spawn(move || reaper_loop(weak_for_reaper, dec_strong_rx))
{
log::error!("RPC: reaper thread spawn failed ({e}); deferred DEC_STRONG will be lost");
}
RpcSession { inner }
}
#[cfg(test)]
pub(crate) fn inner_weak(&self) -> Weak<RpcSessionInner> {
Arc::downgrade(&self.inner)
}
pub(crate) const FOUNDING_SLOT_ID: u64 = 1;
pub(crate) fn add_incoming_slot_capped(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
) -> Result<u64> {
self.inner.add_incoming_slot_capped(transport, cap)
}
pub(crate) fn add_callback_slot_and_init(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
codec: &Android13PlusCodec,
) -> Result<u64> {
if self.inner.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
let claim = self.inner.add_claimed_slot_capped(transport, cap)?;
let sent = {
let mut io = RawTransportIo(&*claim.transport);
super::wire_android13::server_write_connection_init(&mut io, codec)
};
match sent {
Ok(()) => {
self.inner.shared.io_gen.fetch_add(1, Ordering::Relaxed);
Ok(claim.slot_id)
}
Err(e) => {
#[cfg(test)]
self.inner.shared.run_cci_failed_hook();
claim.retire();
Err(StatusCode::from(e))
}
}
}
#[cfg(test)]
pub(crate) fn add_callback_slot(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
) -> Result<u64> {
if self.inner.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
self.inner.add_slot_inner_capped(transport, cap)
}
pub(crate) fn inner_arc(&self) -> Arc<RpcSessionInner> {
Arc::clone(&self.inner)
}
pub(crate) fn wrap_inner(inner: Arc<RpcSessionInner>) -> Self {
RpcSession { inner }
}
pub(crate) fn android13plus_accept_handshake(
transport: Box<dyn RpcTransport>,
server_max_version: u32,
) -> RpcResult<Android13PlusAccept> {
let (codec, client_fd_mode, client_id, incoming) = {
let mut io = RawTransportIo(transport.as_ref());
server_accept_deferred_init(&mut io, server_max_version)?
};
Ok((transport, codec, client_fd_mode, client_id, incoming))
}
pub(crate) fn from_android13plus(
transport: Box<dyn RpcTransport>,
codec: Android13PlusCodec,
client_fd_mode: u8,
server_fd_unix: bool,
) -> RpcResult<RpcSession> {
if !(client_fd_mode == FD_MODE_NONE || (client_fd_mode == FD_MODE_UNIX && server_fd_unix)) {
log::error!(
"android-13+ RPC: rejecting connection: FileDescriptorTransportMode \
{client_fd_mode} is not supported (Unix needs \
`RpcServer::set_supported_fd_modes`)"
);
return Err(RpcError::Protocol(
"client requested an unsupported FileDescriptorTransportMode",
));
}
let negotiated = codec.version();
let shared = Self::fresh_shared(AddressSpace::Acceptor)?;
let session = Self::with_shared(transport, WireProfile::Android13Plus(codec), shared);
if server_fd_unix && client_fd_mode == FD_MODE_UNIX && negotiated >= PROTOCOL_V1 {
session
.inner
.shared
.set_fd_mode(FileDescriptorTransportMode::Unix);
}
Ok(session)
}
pub fn connect_android13plus(
transport: Box<dyn RpcTransport>,
max_version: u32,
) -> Result<RpcSession> {
Self::connect_android13plus_fd(transport, max_version, FileDescriptorTransportMode::None)
}
pub fn connect_android13plus_fd(
transport: Box<dyn RpcTransport>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
) -> Result<RpcSession> {
Self::connect_android13plus_fd_hs(transport, max_version, fd_mode, None)
}
#[deprecated(
since = "0.12.0",
note = "a non-empty id is refused with BadValue: use `connect_android13plus_fd` for a new session, then `add_outgoing_connection_with_config` with `RpcClientConfig::session_id` on that session to add a connection"
)]
pub fn connect_android13plus_fd_with_id(
transport: Box<dyn RpcTransport>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
session_id: &[u8],
) -> Result<RpcSession> {
refuse_id_on_new_session(session_id, "RpcSession::connect_android13plus_fd_with_id")?;
Self::connect_android13plus_fd_hs(transport, max_version, fd_mode, None)
}
pub(crate) fn connect_android13plus_fd_hs(
transport: Box<dyn RpcTransport>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
handshake_timeout: Option<Duration>,
) -> Result<RpcSession> {
let want_unix = fd_mode == FileDescriptorTransportMode::Unix;
let hdr_fd_mode = if want_unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let codec = {
let mut hs = HandshakeDeadline::arm(transport.as_ref(), handshake_timeout)
.map_err(StatusCode::from)?;
let mut io = RawTransportIo(transport.as_ref());
let codec = client_connect_with_id(&mut io, max_version, false, hdr_fd_mode, &[])
.map_err(|e| client_handshake_err(hs.classify(e)))?;
hs.finish().map_err(client_handshake_err)?;
codec
};
let negotiated = codec.version();
let session = RpcSession::with_profile(
transport,
AddressSpace::Initiator,
WireProfile::Android13Plus(codec),
)
.map_err(StatusCode::from)?;
if want_unix && negotiated >= PROTOCOL_V1 {
session
.inner
.shared
.set_fd_mode(FileDescriptorTransportMode::Unix);
}
Ok(session)
}
pub fn accept_android13plus(
transport: Box<dyn RpcTransport>,
server_max_version: u32,
) -> Result<RpcSession> {
Self::accept_android13plus_fd(transport, server_max_version, false)
}
pub fn accept_android13plus_fd(
transport: Box<dyn RpcTransport>,
server_max_version: u32,
server_fd_unix: bool,
) -> Result<RpcSession> {
let (transport, codec, client_fd_mode, _client_id, incoming) =
Self::android13plus_accept_handshake(transport, server_max_version)
.map_err(StatusCode::from)?;
if incoming {
return Err(StatusCode::BadType);
}
Self::from_android13plus(transport, codec, client_fd_mode, server_fd_unix)
.map_err(StatusCode::from)
}
pub fn wire_protocol_version(&self) -> Option<u32> {
match &self.inner.profile {
WireProfile::Android13Plus(c) => Some(c.version()),
WireProfile::R34(_) => None,
}
}
pub fn session_id(&self) -> [u8; 32] {
*self.inner.shared.rpc_session_id.as_bytes()
}
pub fn get_session_id(&self) -> Result<Vec<u8>> {
let data = Parcel::new();
let mut reply = self
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&data,
0,
)?
.ok_or(StatusCode::UnexpectedNull)?;
let id = if self.inner.profile.wire_version().is_some() {
reply.read::<Vec<u8>>()?
} else {
reply.read::<i32>()?.to_le_bytes().to_vec()
};
if self.inner.shared.space() == AddressSpace::Initiator {
let mut kept = self.server_id_lock();
if kept.is_none() {
*kept = Some(id.clone());
}
}
Ok(id)
}
fn server_id_lock(&self) -> std::sync::MutexGuard<'_, Option<Vec<u8>>> {
self.inner
.shared
.server_session_id
.lock()
.expect("server_session_id poisoned")
}
fn own_attach_id(&self) -> Result<Vec<u8>> {
if self.inner.shared.space() != AddressSpace::Initiator {
return Ok(self.session_id().to_vec());
}
let kept = self.server_id_lock().clone();
match kept {
Some(id) => Ok(id),
None => self.get_session_id(),
}
}
fn refuse_foreign_attach_id(&self, session_id: &[u8]) -> Result<()> {
if session_id == self.own_attach_id()?.as_slice() {
return Ok(());
}
if session_id == self.session_id().as_slice() {
log::error!(
"android-13+ RPC: this attach echoes the client-local `RpcSession::session_id()`; \
echo the server-minted id from `RpcSession::get_session_id()`"
);
} else {
log::error!(
"android-13+ RPC: this attach names a server session other than this \
RpcSession's own; two client states on one server session would keep separate \
oneway numbering, binder addresses and lifetimes. Echo this session's \
`RpcSession::get_session_id()`"
);
}
Err(StatusCode::BadValue)
}
pub fn set_supported_fd_modes(&self, modes: &[FileDescriptorTransportMode]) {
let unix = modes.contains(&FileDescriptorTransportMode::Unix);
self.inner
.shared
.fd_unix_supported
.store(unix, Ordering::SeqCst);
}
pub fn negotiate_fd_transport(
&self,
want: FileDescriptorTransportMode,
) -> Result<FileDescriptorTransportMode> {
if let Some(version) = self.inner.profile.wire_version() {
let current = self.inner.fd_mode();
if want == FileDescriptorTransportMode::Unix && current != want {
if version < PROTOCOL_V1 {
log::error!(
"RpcSession::negotiate_fd_transport: wire v0 (android-13) carries no fd \
mode; offer `max_version >= 1` to a v1+ server and request Unix with \
`RpcClientConfig::fd_mode`"
);
} else {
log::error!(
"RpcSession::negotiate_fd_transport: an android-13+ session fixes the \
fd mode in its connection header ({current:?}); request Unix with \
`RpcClientConfig::fd_mode` before connecting"
);
}
return Err(StatusCode::InvalidOperation);
}
return Ok(current);
}
let want_unix = want == FileDescriptorTransportMode::Unix;
let mut req = Parcel::new();
req.write(&(if want_unix { 1i32 } else { 0i32 }))?;
let mut reply = match self.inner.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetFdMode.code(),
&req,
0,
) {
Err(StatusCode::UnknownTransaction) => return Ok(FileDescriptorTransportMode::None),
r => r?.ok_or(StatusCode::UnexpectedNull)?,
};
let agreed = if reply.read::<i32>()? == 1 {
FileDescriptorTransportMode::Unix
} else {
FileDescriptorTransportMode::None
};
self.inner.shared.set_fd_mode(agreed);
Ok(agreed)
}
pub fn fd_transport_mode(&self) -> FileDescriptorTransportMode {
self.inner.fd_mode()
}
pub fn caps(&self) -> crate::TransportCaps {
self.inner.caps()
}
pub fn set_root(&self, binder: SIBinder) -> Result<()> {
super::refuse_remote(&binder, "RpcSession::set_root")?;
*self.inner.shared.root.lock().expect("root poisoned") = Some(binder);
Ok(())
}
pub fn get_root(&self) -> Result<SIBinder> {
let data = Parcel::new();
let reply = self
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetRoot.code(),
&data,
0,
)?
.ok_or(StatusCode::UnexpectedNull)?;
let mut reply = reply;
reply.read::<SIBinder>()
}
pub fn serve_blocking(&self) -> SessionEnd {
self.serve_blocking_on(Self::FOUNDING_SLOT_ID)
}
pub fn spawn_serve(&self) -> Result<std::thread::JoinHandle<SessionEnd>> {
let declared = &self.inner.shared.serve_declared;
let serves = self.inner.slot_role(Self::FOUNDING_SLOT_ID).is_some();
if serves {
declared.fetch_add(1, Ordering::SeqCst);
}
let session = self.clone();
std::thread::Builder::new()
.name("rsbinder-rpc-serve".into())
.spawn(move || {
let none = PhaseDeadline::none();
session.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, false, false, none, serves)
})
.map_err(|e| {
log::error!("RpcSession::spawn_serve: cannot start the serve thread: {e}");
if serves {
declared.fetch_sub(1, Ordering::SeqCst);
}
StatusCode::from(e)
})
}
pub fn serve_blocking_on(&self, slot_id: u64) -> SessionEnd {
self.serve_blocking_on_inner(slot_id, false, false, PhaseDeadline::none(), false)
}
pub fn serve_blocking_clearing_deadline_after_first(&self) -> SessionEnd {
let none = PhaseDeadline::none();
self.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, true, true, none, false)
}
pub(crate) fn serve_blocking_clearing_admission_deadline(
&self,
armed: bool,
admission: PhaseDeadline,
) -> SessionEnd {
self.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, true, armed, admission, false)
}
fn serve_blocking_on_inner(
&self,
slot_id: u64,
clear_deadline_after_first: bool,
admission_deadline_armed: bool,
mut admission: PhaseDeadline,
spawn_declared: bool,
) -> SessionEnd {
let declared = &self.inner.shared.serve_declared;
if self.inner.slot_role(slot_id).is_none() {
if spawn_declared && !self.inner.shared.lifecycle.is_torn_down() {
declared.fetch_sub(1, Ordering::SeqCst);
}
let ended_locally = self.inner.shared.ended_locally.load(Ordering::SeqCst);
return SessionEnd::new(EndReason::SessionEnded, ended_locally, false);
}
declared.fetch_add(1, Ordering::SeqCst);
let baseline_deadline = self.inner.slot_baseline_read_deadline(slot_id).is_some();
let (reason, deadline_armed) = {
let mut first = clear_deadline_after_first;
let mut deadline_armed = baseline_deadline || admission_deadline_armed;
let mut seen = self.inner.activity();
#[cfg(test)]
self.inner.shared.run_serve_wait_hook();
loop {
match self.inner.serve_once_on_slot(slot_id) {
ServeStep::Continue => {
if first {
self.inner.clear_slot_read_timeout(slot_id);
first = false;
if !admission.disarm() {
break (EndReason::Frame(StatusCode::TimedOut), deadline_armed);
}
deadline_armed = false;
}
seen = self.inner.activity();
}
ServeStep::Ended(_) if first && admission.fired() => {
break (EndReason::Frame(StatusCode::TimedOut), deadline_armed);
}
ServeStep::Ended(EndReason::Frame(StatusCode::TimedOut))
if baseline_deadline
&& !first
&& self.inner.slot_role(slot_id).is_none() =>
{
break (EndReason::SessionEnded, deadline_armed);
}
ServeStep::Ended(EndReason::Frame(StatusCode::TimedOut))
if baseline_deadline && !first && self.inner.active_since(&mut seen) => {}
ServeStep::Ended(reason) => break (reason, deadline_armed),
}
}
};
let end = SessionEnd::new(
reason,
self.inner.shared.ended_locally.load(Ordering::SeqCst),
deadline_armed,
);
match (reason, end.by) {
(EndReason::SessionEnded, _) => {}
(_, EndedBy::Local) => self.inner.close(),
_ => self.inner.fail_session(),
}
end
}
pub(crate) fn set_max_threads(&self, n: u32) {
self.inner
.shared
.max_threads
.store(n.max(1), Ordering::SeqCst);
}
pub fn close_session(&self) {
self.inner.close();
let me = std::thread::current().id();
for (slot_id, handle) in self.inner.take_incoming_threads() {
if handle.thread().id() == me {
log::debug!(
"RPC: close_session from incoming connection {slot_id}'s own thread; not joined"
);
continue;
}
if handle.join().is_err() {
log::warn!("RPC: incoming connection {slot_id} thread panicked");
}
self.inner.incoming_joined.fetch_add(1, Ordering::SeqCst);
}
}
pub fn is_ended(&self) -> bool {
self.inner.shared.lifecycle.is_torn_down()
}
pub(crate) fn notices_connection_loss(&self) -> bool {
self.inner.notices_connection_loss()
}
pub(crate) fn peer_closed(&self) -> bool {
self.inner.peer_closed()
}
pub(crate) fn fail(&self) {
self.inner.fail_session();
}
pub fn set_timeout(&self, timeout: Option<Duration>) {
let timeout = reject_zero_deadline(
timeout,
"RpcSession::set_timeout: a zero duration is not a valid deadline; ignoring",
);
self.inner.arm_liveness_all(|| {
*self.inner.shared.timeout.lock().expect("timeout poisoned") = timeout;
});
}
pub(crate) fn set_serve_read_deadline(&self, deadline: Option<Duration>) {
let ns = deadline.map_or(0, |d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX));
self.inner.arm_liveness_all(|| {
self.inner
.shared
.serve_read_deadline
.store(ns, Ordering::Relaxed);
});
}
pub fn negotiated_max_threads(&self) -> u32 {
self.inner.shared.negotiated.load(Ordering::SeqCst)
}
pub fn negotiate(&self, local_max: u32) -> Result<u32> {
let data = Parcel::new();
let mut reply = self
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetMaxThreads.code(),
&data,
0,
)?
.ok_or(StatusCode::UnexpectedNull)?;
let remote: i32 = reply.read()?;
if remote < 1 {
return Err(StatusCode::BadValue);
}
let negotiated = local_max.min(remote as u32).max(1);
self.inner
.shared
.negotiated
.store(negotiated, Ordering::SeqCst);
Ok(negotiated)
}
pub fn setup_unix_client(path: impl AsRef<std::path::Path>) -> Result<RpcSession> {
let t = super::transport::UnixTransport::connect(path)?;
RpcSession::new(Box::new(t), AddressSpace::Initiator).map_err(StatusCode::from)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn setup_unix_client_abstract(name: &[u8]) -> Result<RpcSession> {
let t = super::transport::UnixTransport::connect_abstract(name)?;
RpcSession::new(Box::new(t), AddressSpace::Initiator).map_err(StatusCode::from)
}
pub fn setup_unix_client_android13plus(
path: impl AsRef<std::path::Path>,
max_version: u32,
) -> Result<RpcSession> {
let t = super::transport::UnixTransport::connect(path)?;
RpcSession::connect_android13plus(Box::new(t), max_version)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
pub fn setup_unix_client_android13plus_abstract(
name: &[u8],
max_version: u32,
) -> Result<RpcSession> {
let t = super::transport::UnixTransport::connect_abstract(name)?;
RpcSession::connect_android13plus(Box::new(t), max_version)
}
#[cfg(feature = "rpc-tls")]
pub fn setup_tcp_client_tls(
addr: impl std::net::ToSocketAddrs,
server_name: &str,
config: std::sync::Arc<rustls::ClientConfig>,
) -> Result<RpcSession> {
let tcp = std::net::TcpStream::connect(addr)?;
let t = super::transport::TlsTransport::connect(tcp, server_name, config)
.map_err(StatusCode::from)?;
RpcSession::new(Box::new(t), AddressSpace::Initiator).map_err(StatusCode::from)
}
#[cfg(feature = "rpc-tls")]
pub fn setup_tcp_client_tls_android13plus(
addr: impl std::net::ToSocketAddrs,
server_name: &str,
config: std::sync::Arc<rustls::ClientConfig>,
max_version: u32,
) -> Result<RpcSession> {
let tcp = std::net::TcpStream::connect(addr)?;
let t = super::transport::TlsTransport::connect(tcp, server_name, config)
.map_err(StatusCode::from)?;
RpcSession::connect_android13plus(Box::new(t), max_version)
}
#[deprecated(
since = "0.12.0",
note = "a non-empty id is refused with BadValue: use `setup_client_android13plus_with_config(RpcClientConfig::unix(path, v))` for a new session, then `add_outgoing_connection_with_config(RpcClientConfig::unix(path, v).session_id(id))` on that session to add a connection"
)]
pub fn setup_unix_client_android13plus_with_id(
path: impl AsRef<std::path::Path>,
max_version: u32,
session_id: &[u8],
) -> Result<RpcSession> {
Self::setup_client_android13plus_with_config(
RpcClientConfig::unix(path.as_ref(), max_version).session_id(session_id),
)
}
#[deprecated(
since = "0.12.0",
note = "use `setup_client_android13plus_with_config` with `RpcClientConfig::unix`/`unix_abstract`"
)]
#[expect(deprecated, reason = "takes the deprecated `RpcUnixClientConfig`")]
pub fn setup_unix_client_android13plus_with_config(
config: RpcUnixClientConfig,
) -> Result<RpcSession> {
Self::setup_client_android13plus(
config.into_generic(),
"RpcSession::setup_unix_client_android13plus_with_config",
"RpcUnixClientConfig::handshake_timeout",
)
}
pub fn setup_client_android13plus_with_config(config: RpcClientConfig) -> Result<RpcSession> {
Self::setup_client_android13plus(
config,
"RpcSession::setup_client_android13plus_with_config",
"RpcClientConfig::handshake_timeout",
)
}
fn setup_client_android13plus(
config: RpcClientConfig,
entry: &str,
what: &str,
) -> Result<RpcSession> {
reject_zero_handshake_timeout(config.handshake_timeout, what)?;
let handshake_deadline = config.handshake_deadline();
let RpcClientConfig {
source,
max_version,
session_id: requested_id,
outgoing_connections,
incoming_connections: incoming,
fd_mode: requested_fd_mode,
timeout,
handshake_timeout: _,
} = config;
let handshake_timeout = handshake_deadline;
let local = outgoing_connections.max(1);
refuse_id_on_new_session(requested_id, entry)?;
let mut connect = source.into_connector(handshake_timeout);
let founding = connect()?;
if requested_fd_mode == Some(FileDescriptorTransportMode::Unix)
&& !founding.supports_fd_passing()
{
log::error!(
"RPC client: FileDescriptorTransportMode::Unix needs a Unix-domain transport"
);
return Err(StatusCode::BadValue);
}
let session = RpcSession::connect_android13plus_fd_hs(
founding,
max_version,
requested_fd_mode.unwrap_or(FileDescriptorTransportMode::None),
handshake_timeout,
)?;
if timeout.is_some() {
session.set_timeout(timeout);
}
if local == 1 && incoming == 0 {
return Ok(session);
}
let mut build = || -> Result<()> {
let negotiated = if local > 1 {
session.negotiate(local)?
} else {
1
};
let session_id = session.get_session_id()?;
let fd_mode = session.fd_transport_mode();
for _ in 1..negotiated {
session.add_outgoing_connection_android13plus_transport(
&mut connect,
max_version,
&session_id,
fd_mode,
handshake_timeout,
)?;
}
for _ in 0..incoming {
session.add_incoming_connection_android13plus_transport(
&mut connect,
max_version,
&session_id,
fd_mode,
handshake_timeout,
)?;
}
Ok(())
};
if let Err(e) = build() {
session.close_session();
return Err(e);
}
Ok(session)
}
#[deprecated(
since = "0.12.0",
note = "use `add_outgoing_connection_with_config(RpcClientConfig::unix(path, v).session_id(id))`"
)]
pub fn add_outgoing_connection_android13plus(
&self,
path: impl AsRef<std::path::Path>,
max_version: u32,
session_id: &[u8],
) -> Result<u64> {
self.add_outgoing_connection_with_config(
RpcClientConfig::unix(path.as_ref(), max_version).session_id(session_id),
)
}
#[deprecated(
since = "0.12.0",
note = "use `add_outgoing_connection_with_config` with a `RpcClientConfig`"
)]
#[expect(deprecated, reason = "takes the deprecated `RpcUnixClientConfig`")]
pub fn add_outgoing_connection_android13plus_with_config(
&self,
config: RpcUnixClientConfig,
) -> Result<u64> {
self.add_outgoing_connection_named(
config.into_attach(),
"RpcUnixClientConfig::handshake_timeout",
)
}
pub fn add_outgoing_connection_with_config(&self, config: RpcClientConfig) -> Result<u64> {
self.add_outgoing_connection_named(config, "RpcClientConfig::handshake_timeout")
}
fn add_outgoing_connection_named(&self, config: RpcClientConfig, what: &str) -> Result<u64> {
let AttachParts {
mut connect,
max_version,
session_id,
fd_mode,
handshake_timeout,
} = self.attach_parts(config, what)?;
self.add_outgoing_connection_android13plus_transport(
&mut connect,
max_version,
session_id,
fd_mode,
handshake_timeout,
)
}
fn attach_parts<'a>(&self, config: RpcClientConfig<'a>, what: &str) -> Result<AttachParts<'a>> {
reject_zero_handshake_timeout(config.handshake_timeout, what)?;
if config.outgoing_connections.max(1) != 1
|| config.incoming_connections != 0
|| config.session_id.len() != 32
|| config.timeout.is_some()
{
return Err(StatusCode::BadValue);
}
let fd_mode = self.fd_transport_mode();
if config.fd_mode.is_some_and(|mode| mode != fd_mode) {
return Err(StatusCode::BadValue);
}
let session_timeout = *self.inner.shared.timeout.lock().expect("timeout poisoned");
let handshake_timeout = config.handshake_timeout.or(session_timeout);
Ok(AttachParts {
connect: config.source.into_connector(handshake_timeout),
max_version: config.max_version,
session_id: config.session_id,
fd_mode,
handshake_timeout,
})
}
fn add_outgoing_connection_android13plus_transport(
&self,
connect: impl FnOnce() -> Result<Box<dyn RpcTransport>>,
max_version: u32,
session_id: &[u8],
fd_mode: FileDescriptorTransportMode,
handshake_timeout: Option<Duration>,
) -> Result<u64> {
if session_id.len() != 32 {
return Err(StatusCode::BadValue);
}
let session_version = match &self.inner.profile {
WireProfile::Android13Plus(c) => c.version(),
WireProfile::R34(_) => return Err(StatusCode::BadType),
};
if self.inner.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
let effective_max = max_version.min(session_version);
if effective_max != session_version {
log::error!(
"android-13+ RPC: this attach asks for wire v{effective_max} but the session \
runs v{session_version} — a caller-supplied `max_version` below the session's \
negotiated version can never attach; pass `RpcSession::wire_protocol_version()`"
);
return Err(StatusCode::BadType);
}
self.refuse_foreign_attach_id(session_id)?;
let hdr_fd_mode = if fd_mode == FileDescriptorTransportMode::Unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let t = connect()?;
if !self
.inner
.conn_state
.lock()
.expect("conn_state poisoned")
.admits(&*t)
{
return Err(StatusCode::BadType);
}
let codec = {
let mut hs =
HandshakeDeadline::arm(&*t, handshake_timeout).map_err(StatusCode::from)?;
let mut io = RawTransportIo(&*t);
let codec = client_write_connection_header(
&mut io,
effective_max,
false,
hdr_fd_mode,
session_id,
)
.map_err(|e| StatusCode::from(hs.classify(e)))?;
let init = std::io::Write::write_all(&mut io, &codec.encode_connection_init());
if let Err(e) = init {
let e = hs.classify(RpcError::from(e));
return Err(self.end_past_outgoing_header(StatusCode::from(e)));
}
if let Err(e) = hs.finish() {
return Err(self.end_past_outgoing_header(StatusCode::from(e)));
}
codec
};
{
let probe_deadline =
handshake_timeout.or(*self.inner.shared.timeout.lock().expect("timeout poisoned"));
let mut hs = match HandshakeDeadline::arm(&*t, probe_deadline) {
Ok(hs) => hs,
Err(e) => return Err(self.end_past_outgoing_header(StatusCode::from(e))),
};
if let Err(e) = confirm_attach(&*t, &codec, session_id).and_then(|()| hs.finish()) {
let e = hs.classify(e);
log_attach_refused(&e);
return Err(self.end_past_outgoing_header(StatusCode::from(e)));
}
}
match self.inner.add_outgoing_slot(t) {
Ok(id) => Ok(id),
Err(e) => Err(self.end_past_outgoing_header(e)),
}
}
fn end_past_outgoing_header(&self, status: StatusCode) -> StatusCode {
log::error!("RPC: outgoing attach failed past its header ({status:?}); ending the session");
self.inner.fail_session();
status
}
#[deprecated(
since = "0.12.0",
note = "use `add_incoming_connection_with_config` with a `RpcClientConfig`"
)]
#[expect(deprecated, reason = "takes the deprecated `RpcUnixClientConfig`")]
pub fn add_incoming_connection_android13plus_with_config(
&self,
config: RpcUnixClientConfig,
) -> Result<u64> {
self.add_incoming_connection_named(
config.into_attach(),
"RpcUnixClientConfig::handshake_timeout",
)
}
pub fn add_incoming_connection_with_config(&self, config: RpcClientConfig) -> Result<u64> {
self.add_incoming_connection_named(config, "RpcClientConfig::handshake_timeout")
}
fn add_incoming_connection_named(&self, config: RpcClientConfig, what: &str) -> Result<u64> {
let AttachParts {
mut connect,
max_version,
session_id,
fd_mode,
handshake_timeout,
} = self.attach_parts(config, what)?;
self.add_incoming_connection_android13plus_transport(
&mut connect,
max_version,
session_id,
fd_mode,
handshake_timeout,
)
}
fn add_incoming_connection_android13plus_transport(
&self,
connect: impl FnOnce() -> Result<Box<dyn RpcTransport>>,
max_version: u32,
session_id: &[u8],
fd_mode: FileDescriptorTransportMode,
handshake_timeout: Option<Duration>,
) -> Result<u64> {
if self.inner.shared.space() != AddressSpace::Initiator {
return Err(StatusCode::InvalidOperation);
}
if session_id.len() != 32 {
return Err(StatusCode::BadValue);
}
let session_version = match &self.inner.profile {
WireProfile::Android13Plus(c) => c.version(),
WireProfile::R34(_) => return Err(StatusCode::BadType),
};
if self.inner.shared.lifecycle.is_torn_down() {
return Err(StatusCode::DeadObject);
}
let effective_max = max_version.min(session_version);
if effective_max != session_version {
log::error!(
"android-13+ RPC: this incoming attach asks for wire v{effective_max} but the \
session runs v{session_version}; pass `RpcSession::wire_protocol_version()`"
);
return Err(StatusCode::BadType);
}
self.refuse_foreign_attach_id(session_id)?;
let hdr_fd_mode = if fd_mode == FileDescriptorTransportMode::Unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let t = connect()?;
if !self
.inner
.conn_state
.lock()
.expect("conn_state poisoned")
.admits(&*t)
{
return Err(StatusCode::BadType);
}
{
let mut hs =
HandshakeDeadline::arm(&*t, handshake_timeout).map_err(StatusCode::from)?;
let mut io = RawTransportIo(&*t);
let codec = client_write_connection_header(
&mut io,
effective_max,
true,
hdr_fd_mode,
session_id,
)
.map_err(|e| StatusCode::from(hs.classify(e)))?;
if let Err((e, received)) = client_read_connection_init(&mut io, &codec) {
return Err(self.fail_awaiting_cci(hs.classify(e), received));
}
if let Err(e) = hs.finish() {
let received = super::wire_android13::A13_CONN_INIT_LEN;
return Err(self.fail_awaiting_cci(e, received));
}
}
let slot_id = match self.inner.add_slot_inner(t, SlotRole::Incoming) {
Ok(id) => id,
Err(e) => return Err(self.fail_after_cci(e)),
};
let inner = Arc::clone(&self.inner);
self.inner.incoming_live.fetch_add(1, Ordering::SeqCst);
let builder = std::thread::Builder::new().name(format!("rsbinder-rpc-in-{slot_id}"));
let spawned = self.inner.shared.spawn_incoming(builder, move || {
let session = RpcSession::wrap_inner(inner);
session
.serve_blocking_on(slot_id)
.log(&format!("RPC: incoming connection {slot_id} ended"));
session.inner.incoming_live.fetch_sub(1, Ordering::SeqCst);
});
match spawned {
Ok(handle) => {
self.inner
.incoming_threads
.lock()
.expect("incoming_threads poisoned")
.push((slot_id, handle));
if !self.inner.shared.lifecycle.is_torn_down() {
return Ok(slot_id);
}
let mine = {
let mut threads = self
.inner
.incoming_threads
.lock()
.expect("incoming_threads poisoned");
let at = threads.iter().position(|(id, _)| *id == slot_id);
at.map(|i| threads.remove(i).1)
};
if let Some(handle) = mine {
if handle.join().is_err() {
log::warn!("RPC: incoming connection {slot_id} thread panicked");
}
self.inner.incoming_joined.fetch_add(1, Ordering::SeqCst);
}
Err(StatusCode::DeadObject)
}
Err(e) => {
log::error!("RPC: incoming connection thread spawn failed ({e})");
self.inner.incoming_live.fetch_sub(1, Ordering::SeqCst);
Err(self.fail_after_cci(StatusCode::from(e)))
}
}
}
fn fail_after_cci(&self, status: StatusCode) -> StatusCode {
log::error!(
"RPC: incoming attach failed past the server's \"cci\" ({status:?}); ending the session"
);
self.inner.fail_session();
status
}
fn fail_awaiting_cci(&self, e: RpcError, received: usize) -> StatusCode {
if received == 0 && matches!(e, RpcError::EndOfStream | RpcError::UncleanEndOfStream) {
return StatusCode::from(e);
}
log::error!(
"RPC: incoming attach failed awaiting the server's \"cci\" ({e}); ending the session"
);
self.inner.fail_session();
StatusCode::from(e)
}
#[deprecated(
since = "0.12.0",
note = "use `setup_client_android13plus_with_config(RpcClientConfig::unix(path, v).outgoing_connections(n))`"
)]
pub fn setup_unix_client_android13plus_fan_out(
path: impl AsRef<std::path::Path>,
max_version: u32,
local_max_outgoing: u32,
) -> Result<RpcSession> {
Self::setup_client_android13plus_with_config(
RpcClientConfig::unix(path.as_ref(), max_version)
.outgoing_connections(local_max_outgoing),
)
}
pub fn setup_unix_client_android13plus_fd(
path: impl AsRef<std::path::Path>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
) -> Result<RpcSession> {
let t = super::transport::UnixTransport::connect(path)?;
RpcSession::connect_android13plus_fd(Box::new(t), max_version, fd_mode)
}
pub fn from_preconnected_fd(fd: OwnedFd, max_version: u32) -> Result<RpcSession> {
let local = rustix::net::getsockname(fd.as_fd())
.map_err(|e| RpcError::from(std::io::Error::from(e)))?;
let family = local.address_family();
let flags = rustix::fs::fcntl_getfl(fd.as_fd())
.map_err(|e| RpcError::from(std::io::Error::from(e)))?;
if flags.contains(rustix::fs::OFlags::NONBLOCK) {
rustix::fs::fcntl_setfl(fd.as_fd(), flags - rustix::fs::OFlags::NONBLOCK)
.map_err(|e| RpcError::from(std::io::Error::from(e)))?;
}
let transport: Box<dyn RpcTransport> = match family {
rustix::net::AddressFamily::UNIX => {
Box::new(super::transport::UnixTransport::from_owned_fd(fd)?)
}
#[cfg(all(feature = "rpc-vsock", any(target_os = "linux", target_os = "android")))]
rustix::net::AddressFamily::VSOCK => {
Box::new(super::transport::VsockTransport::from_owned_fd(fd)?)
}
#[cfg(feature = "rpc-tcp-debug")]
rustix::net::AddressFamily::INET | rustix::net::AddressFamily::INET6 => {
Box::new(super::transport::TcpDebugTransport::from_owned_fd(fd)?)
}
_ => {
log::warn!(
"RPC preconnected fd has unsupported socket family ({:?}); \
rejecting (AOSP IAccessor::ERROR_UNSUPPORTED_SOCKET_FAMILY)",
family.as_raw()
);
return Err(StatusCode::BadType);
}
};
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
let session = RpcSession::connect_android13plus_fd_hs(
transport,
max_version,
FileDescriptorTransportMode::None,
Some(HANDSHAKE_TIMEOUT),
)?;
session.inner.clear_handshake_timeouts();
Ok(session)
}
#[cfg(feature = "test-util")]
#[doc(hidden)]
pub fn __slot_count(&self) -> usize {
self.inner.slot_count()
}
#[cfg(feature = "test-util")]
#[doc(hidden)]
pub fn __held_dec_strong_limit() -> usize {
HELD_DEC_STRONG_LIMIT
}
#[cfg(feature = "test-util")]
#[doc(hidden)]
pub fn __incoming_thread_count(&self) -> usize {
self.inner
.incoming_threads
.lock()
.expect("incoming_threads poisoned")
.len()
}
#[cfg(feature = "test-util")]
#[doc(hidden)]
pub fn __incoming_thread_live_count(&self) -> usize {
self.inner.incoming_live.load(Ordering::SeqCst)
}
#[cfg(feature = "test-util")]
#[doc(hidden)]
pub fn __incoming_thread_joined_count(&self) -> usize {
self.inner.incoming_joined.load(Ordering::SeqCst)
}
pub fn local_node_count(&self) -> usize {
self.inner
.shared
.state
.lock()
.expect("rpc state poisoned")
.local_node_count()
}
}
fn reaper_loop(weak: Weak<RpcSessionInner>, rx: mpsc::Receiver<(RpcAddress, u32)>) {
while let Ok((addr, amount)) = rx.recv() {
let Some(inner) = weak.upgrade() else {
return;
};
if inner.shared.lifecycle.is_torn_down() {
continue;
}
let Some(conn) = inner.find_conn_for_reaper() else {
drop(inner);
continue;
};
let _ = inner.write_dec_strong(&conn, addr, amount);
drop(conn);
drop(inner);
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::os::fd::{AsFd, OwnedFd};
#[test]
fn rpc_session_keeps_its_auto_traits() {
fn assert_traits<
T: Send + Sync + Unpin + std::panic::RefUnwindSafe + std::panic::UnwindSafe,
>() {
}
assert_traits::<RpcSession>();
}
#[cfg(feature = "rpc-tls")]
#[test]
fn tls_connections_stay_on_the_first_resolved_address() {
use rustls::{ClientConfig, RootCertStore};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
for s in listener.incoming().take(2) {
drop(s);
let _ = tx.send(());
}
});
let cfg = std::sync::Arc::new(
ClientConfig::builder()
.with_root_certificates(RootCertStore::empty())
.with_no_client_auth(),
);
let mut pinned = None;
let timeout = Some(Duration::from_secs(2));
let wait = Duration::from_secs(2);
let first = connect_tls(
"127.0.0.1",
addr.port(),
&mut pinned,
"localhost",
&cfg,
timeout,
);
assert!(first.is_err());
rx.recv_timeout(wait).expect("first TCP connect");
assert_eq!(pinned, Some(addr));
let second = connect_tls(
"rsbinder-pin-test.invalid",
addr.port(),
&mut pinned,
"localhost",
&cfg,
timeout,
);
assert!(second.is_err());
rx.recv_timeout(wait)
.expect("second TCP connect went to the pinned address");
}
#[cfg(all(feature = "rpc-tls", any(target_os = "linux", target_os = "android")))]
fn tcp_silent_listener() -> (std::net::TcpListener, Vec<std::net::TcpStream>) {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
rustix::net::listen(&listener, 0).expect("listen");
let addr = listener.local_addr().expect("addr");
let mut queued = Vec::new();
loop {
match std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(200)) {
Ok(s) => queued.push(s),
Err(e) if e.kind() == std::io::ErrorKind::TimedOut => break,
Err(e) => panic!("filling the accept queue: {e}"),
}
}
(listener, queued)
}
#[cfg(all(feature = "rpc-tls", any(target_os = "linux", target_os = "android")))]
#[test]
fn a_tls_connect_bounds_all_addresses_by_one_deadline() {
let (silent, _queued) = tcp_silent_listener();
let silent = silent.local_addr().expect("addr");
let d = Duration::from_secs(1);
let start = Instant::now();
let e = tcp_connect_any(&[silent, silent], Some(d)).expect_err("nothing answers");
let took = start.elapsed();
assert_eq!(e.kind(), std::io::ErrorKind::TimedOut);
assert!(
took < d + d / 2,
"two silent addresses took {took:?}, deadline {d:?}"
);
let live = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let live = live.local_addr().expect("addr");
let d = CONNECT_SHARE_FLOOR + CONNECT_SHARE_FLOOR / 2;
let start = Instant::now();
let (_, reached) = tcp_connect_any(&[silent, live], Some(d)).expect("the live address");
let took = start.elapsed();
assert_eq!(reached, live);
assert!(
took >= CONNECT_SHARE_FLOOR * 9 / 10,
"the silent address got {took:?}, under the floor"
);
}
#[cfg(feature = "rpc-tls")]
#[test]
fn a_connect_share_is_even_above_the_floor() {
let s = Duration::from_secs;
assert_eq!(connect_share(s(10), 2), s(5));
assert_eq!(connect_share(s(10), 10), CONNECT_SHARE_FLOOR);
assert_eq!(connect_share(s(1), 2), s(1));
assert_eq!(connect_share(s(3), 1), s(3));
}
#[cfg(all(feature = "rpc-tls", any(target_os = "linux", target_os = "android")))]
#[test]
fn a_refused_address_leaves_its_share_to_the_next() {
let (silent, _queued) = tcp_silent_listener();
let silent = silent.local_addr().expect("addr");
let refused = std::net::TcpListener::bind("127.0.0.1:0")
.expect("bind")
.local_addr()
.expect("addr");
let d = CONNECT_SHARE_FLOOR * 2;
let start = Instant::now();
let e = tcp_connect_any(&[refused, silent], Some(d)).expect_err("nothing answers");
let took = start.elapsed();
assert_eq!(e.kind(), std::io::ErrorKind::TimedOut);
assert!(
took >= d * 9 / 10,
"the silent address got {took:?} of {d:?}"
);
}
fn unix_socketpair_fd() -> (OwnedFd, OwnedFd) {
use rustix::net::{AddressFamily, SocketFlags, SocketType};
rustix::net::socketpair(
AddressFamily::UNIX,
SocketType::STREAM,
SocketFlags::empty(),
None,
)
.expect("socketpair")
}
#[test]
fn zero_handshake_deadline_is_refused_before_it_reaches_a_transport() {
let (a, _b) = super::super::transport::MemTransport::pair();
assert!(
HandshakeDeadline::arm(&a, Some(Duration::ZERO)).is_err(),
"a zero duration is not a deadline"
);
assert!(HandshakeDeadline::arm(&a, None).is_ok());
assert!(HandshakeDeadline::arm(&a, Some(Duration::from_millis(50))).is_ok());
}
struct RefusesDeadlines {
closed: bool,
}
impl RpcTransport for RefusesDeadlines {
fn send_frame(&self, _buf: &[u8]) -> RpcResult<()> {
Err(RpcError::EndOfStream)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
Err(RpcError::EndOfStream)
}
fn peer_identity(&self) -> super::super::transport::PeerIdentity {
super::super::transport::PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"refuses-deadlines"
}
fn shutdown(&self) -> RpcResult<()> {
Ok(())
}
fn set_read_timeout(&self, _timeout: Option<Duration>) -> RpcResult<()> {
Err(RpcError::Io(std::io::ErrorKind::InvalidInput.into()))
}
fn peer_closed(&self) -> Option<bool> {
Some(self.closed)
}
}
#[test]
fn a_handshake_deadline_refused_on_a_closed_peer_is_skipped() {
let d = Some(Duration::from_secs(1));
assert!(HandshakeDeadline::arm(&RefusesDeadlines { closed: true }, d).is_ok());
assert!(HandshakeDeadline::arm(&RefusesDeadlines { closed: false }, d).is_err());
}
#[cfg(any(feature = "rpc-tcp-debug", feature = "rpc-tls"))]
#[test]
fn a_connect_deadline_is_timed_out_not_unknown() {
let own_deadline = std::io::Error::from(std::io::ErrorKind::TimedOut);
assert_eq!(own_deadline.raw_os_error(), None, "the case under test");
assert_eq!(connect_status(own_deadline), StatusCode::TimedOut);
let etimedout = rustix::io::Errno::TIMEDOUT.raw_os_error();
assert_eq!(
connect_status(std::io::Error::from_raw_os_error(etimedout)),
StatusCode::TimedOut
);
let refused = rustix::io::Errno::CONNREFUSED.raw_os_error();
assert_eq!(
connect_status(std::io::Error::from_raw_os_error(refused)),
StatusCode::from(std::io::Error::from_raw_os_error(refused)),
"every other failure keeps its errno"
);
}
#[test]
#[allow(deprecated)] fn zero_handshake_timeout_is_bad_value_at_the_config_entries() {
let path = std::path::Path::new("/nonexistent/rsb-zero-handshake.sock");
assert_eq!(
RpcSession::setup_client_android13plus_with_config(
RpcClientConfig::unix(path, 2).handshake_timeout(Duration::ZERO),
)
.err(),
Some(StatusCode::BadValue),
"rejected before the connect — the path is never touched"
);
}
#[test]
fn a_nested_call_that_loses_the_stream_ends_the_session() {
use crate::rpc::wire_android13::write_aosp_message;
use std::os::unix::net::UnixStream;
let (client_fd, peer_fd) = unix_socketpair_fd();
let mut peer = UnixStream::from(peer_fd);
let session = RpcSession::with_profile(
Box::new(
super::super::transport::UnixTransport::from_stream(UnixStream::from(client_fd))
.expect("transport"),
),
AddressSpace::Initiator,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("session");
let mut undecodable = [0u8; 16];
undecodable[0..4].copy_from_slice(&7u32.to_le_bytes());
write_aosp_message(&mut peer, &undecodable).expect("queue the undecodable frame");
let (slot_id, slot_transport) = {
let st = session.inner.conn_state.lock().expect("conn_state");
(st.slots[0].id, Arc::clone(&st.slots[0].transport))
};
let sess_ptr = &*session.inner as *const RpcSessionInner as usize;
struct DrivingMark;
impl Drop for DrivingMark {
fn drop(&mut self) {
DRIVING.with(|d| {
d.borrow_mut().pop();
});
}
}
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
let _mark = DrivingMark;
let err = session
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
)
.expect_err("an undecodable frame fails the nested call");
assert_eq!(
err,
StatusCode::RpcError,
"the nested call gets its own failure"
);
let lifecycle = &session.inner.shared.lifecycle;
assert!(lifecycle.is_torn_down(), "a lost stream ends the session");
assert!(
!session.inner.shared.ended_locally.load(Ordering::SeqCst),
"a fault, not this end's decision"
);
assert!(
slot_transport.send_raw(b"x").is_err(),
"the borrowed connection must be shut down, not left usable"
);
assert!(
matches!(
session.inner.serve_once_on_slot(slot_id),
ServeStep::Ended(EndReason::SessionEnded)
),
"the serve loop must stop, not read another frame on the slot"
);
}
#[test]
fn a_frame_read_after_a_local_end_is_not_dispatched() {
use crate::rpc::wire_android13::write_aosp_message;
use std::os::unix::net::UnixStream;
let build = || {
let (client_fd, peer_fd) = unix_socketpair_fd();
let mut peer = UnixStream::from(peer_fd);
let session = RpcSession::with_profile(
Box::new(
super::super::transport::UnixTransport::from_stream(UnixStream::from(
client_fd,
))
.expect("transport"),
),
AddressSpace::Acceptor,
WireProfile::Android13Plus(
Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2"),
),
)
.expect("session");
let mut undecodable = [0u8; 16];
undecodable[0..4].copy_from_slice(&7u32.to_le_bytes());
write_aosp_message(&mut peer, &undecodable).expect("queue a frame");
(session, peer)
};
let (ended, _peer) = build();
ended
.inner
.shared
.ended_locally
.store(true, Ordering::SeqCst);
assert!(
matches!(
ended.inner.serve_once_on_slot(RpcSession::FOUNDING_SLOT_ID),
ServeStep::Ended(EndReason::Interrupted)
),
"a frame read after a local end must end the loop, not be dispatched"
);
let (live, _peer) = build();
assert!(
matches!(
live.inner.serve_once_on_slot(RpcSession::FOUNDING_SLOT_ID),
ServeStep::Ended(EndReason::Frame(_))
),
"the same frame on a live session reaches the decoder"
);
}
#[test]
fn a_real_nested_call_that_loses_the_stream_fails_the_outer_call() {
use crate::rpc::wire_android13::read_aosp_message;
use crate::{Binder, Interface, Remotable, TransactionCode};
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
const DESC: &str = "rsbinder.test.INestedLoser";
const TX_OUTER: TransactionCode = crate::FIRST_CALL_TRANSACTION;
const TX_CALLBACK: TransactionCode = crate::FIRST_CALL_TRANSACTION + 1;
const TX_NESTED: TransactionCode = crate::FIRST_CALL_TRANSACTION + 2;
struct Loser {
session: Arc<Mutex<Option<Arc<RpcSessionInner>>>>,
nested: Arc<Mutex<Option<Result<()>>>>,
}
impl Interface for Loser {}
impl Remotable for Loser {
fn descriptor() -> &'static str {
DESC
}
fn on_transact(
&self,
code: TransactionCode,
_reader: &mut Parcel,
_reply: &mut Parcel,
) -> Result<()> {
assert_eq!(code, TX_CALLBACK, "the peer's nested TRANSACT");
let session = self
.session
.lock()
.expect("session cell")
.clone()
.expect("session installed before the outer call");
let r = session
.client_transact(RpcAddress::zero(), TX_NESTED, &Parcel::new(), 0)
.map(|_| ());
*self.nested.lock().expect("nested cell") = Some(r);
Ok(())
}
fn on_dump(&self, _w: &mut dyn std::io::Write, _a: &[String]) -> Result<()> {
Ok(())
}
}
let (client_fd, peer_fd) = unix_socketpair_fd();
let mut peer = UnixStream::from(peer_fd);
peer.set_read_timeout(Some(Duration::from_secs(10)))
.expect("peer read timeout");
let peer_dup = peer.try_clone().expect("dup the peer socket");
let session = RpcSession::with_profile(
Box::new(
super::super::transport::UnixTransport::from_stream(UnixStream::from(client_fd))
.expect("transport"),
),
AddressSpace::Initiator,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("session");
session.set_timeout(Some(Duration::from_secs(10)));
let session_cell = Arc::new(Mutex::new(Some(Arc::clone(&session.inner))));
let nested_cell: Arc<Mutex<Option<Result<()>>>> = Arc::new(Mutex::new(None));
let cb: SIBinder = Interface::as_binder(&Binder::new(Loser {
session: Arc::clone(&session_cell),
nested: Arc::clone(&nested_cell),
}));
let cb_addr = session
.inner
.shared
.state
.lock()
.expect("rpc state")
.on_binder_leaving(&cb)
.expect("address for the callback");
let (slot_id, slot_transport) = {
let st = session.inner.conn_state.lock().expect("conn_state");
(st.slots[0].id, Arc::clone(&st.slots[0].transport))
};
let peer_thread = std::thread::spawn(move || {
let codec = Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2");
let frame = read_aosp_message(&mut peer).expect("the outer TRANSACT");
match codec
.decode_message(&frame)
.expect("decode the outer TRANSACT")
{
WireMessage::Transact(t) => assert_eq!(t.code, TX_OUTER),
other => panic!("expected the outer TRANSACT, got {other:?}"),
}
let mut token = Parcel::new();
write_rpc_interface_token(&mut token, DESC).expect("interface token");
let nested = codec
.encode_transact(&WireTransaction {
address: cb_addr,
code: TX_CALLBACK,
flags: 0,
async_number: 0,
data: token.rpc_data_bytes().to_vec(),
object_positions: Vec::new(),
})
.expect("encode the nested TRANSACT");
let mut undecodable = [0u8; 16];
undecodable[0..4].copy_from_slice(&7u32.to_le_bytes());
let reply = codec
.encode_reply(&WireReply::default())
.expect("encode a REPLY");
let mut burst = nested;
burst.extend_from_slice(&undecodable);
burst.extend_from_slice(&reply);
burst.extend_from_slice(&reply);
peer.write_all(&burst).expect("queue the burst");
peer.flush().expect("flush the burst");
let mut sink = Vec::new();
let _ = peer.read_to_end(&mut sink);
});
let outer = session
.inner
.client_transact(RpcAddress::zero(), TX_OUTER, &Parcel::new(), 0);
assert!(
matches!(outer, Err(StatusCode::DeadObject)),
"the outer call must fail as DeadObject, got {outer:?}"
);
assert!(
matches!(
*nested_cell.lock().expect("nested cell"),
Some(Err(StatusCode::RpcError))
),
"the nested call must have run and failed on the undecodable frame"
);
assert!(
session
.inner
.conn_state
.lock()
.expect("conn_state")
.slots
.is_empty(),
"the session's end empties the pool, slot {slot_id} with it"
);
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"the nested call's lost stream ended the session"
);
assert!(
slot_transport.send_raw(b"x").is_err(),
"the lost connection must be shut down, not left usable"
);
session.close_session();
drop(slot_transport);
let _ = peer_dup.shutdown(std::net::Shutdown::Both);
peer_thread.join().expect("peer thread");
}
#[test]
fn a_oneway_from_a_twoway_dispatch_never_nests_on_the_serving_slot() {
let (a, _b) = super::super::transport::UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Acceptor).expect("session");
let slot_id = {
let mut st = session.inner.conn_state.lock().expect("conn_state");
let slot = &mut st.slots[0];
slot.role = SlotRole::Incoming;
slot.allow_nested = true;
slot.exclusive_tid = Some(current_tid());
slot.id
};
let sess_ptr = &*session.inner as *const RpcSessionInner as usize;
struct DrivingMark;
impl Drop for DrivingMark {
fn drop(&mut self) {
DRIVING.with(|d| {
d.borrow_mut().pop();
});
}
}
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
let _mark = DrivingMark;
let nested = session
.inner
.find_conn()
.expect("a twoway nests on the serving slot");
assert_eq!(nested.slot_id, slot_id);
drop(nested);
assert!(
matches!(session.inner.find_conn_async(), Err(StatusCode::WouldBlock)),
"a oneway needs an outgoing slot, and this session has none"
);
}
#[test]
fn from_preconnected_fd_unix_dispatches_then_fails_cleanly_on_eof() {
let (a, b) = unix_socketpair_fd();
drop(b);
let err = match RpcSession::from_preconnected_fd(a, 2) {
Ok(_) => panic!("expected Err on closed peer"),
Err(e) => e,
};
assert!(
matches!(
err,
StatusCode::DeadObject | StatusCode::NotEnoughData | StatusCode::Unknown
),
"unexpected status for closed peer: {err}"
);
}
#[test]
fn from_preconnected_fd_clears_o_nonblock_before_dispatch() {
use std::os::fd::IntoRawFd;
let (a, _b) = unix_socketpair_fd();
let flags = rustix::fs::fcntl_getfl(a.as_fd()).expect("getfl");
rustix::fs::fcntl_setfl(a.as_fd(), flags | rustix::fs::OFlags::NONBLOCK)
.expect("setfl NONBLOCK");
assert!(
rustix::fs::fcntl_getfl(a.as_fd())
.unwrap()
.contains(rustix::fs::OFlags::NONBLOCK),
"test setup: O_NONBLOCK must be set"
);
let observer = rustix::io::fcntl_dupfd_cloexec(a.as_fd(), 0).expect("dup");
let bridge_t = std::thread::spawn(move || {
let _ = RpcSession::from_preconnected_fd(a, 2);
});
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
let observed = loop {
let fl = rustix::fs::fcntl_getfl(observer.as_fd()).expect("getfl observer");
if !fl.contains(rustix::fs::OFlags::NONBLOCK) || std::time::Instant::now() > deadline {
break fl;
}
std::thread::sleep(std::time::Duration::from_millis(5));
};
assert!(
!observed.contains(rustix::fs::OFlags::NONBLOCK),
"from_preconnected_fd did NOT clear O_NONBLOCK — handshake will trip EAGAIN \
against a non-blocking peer fd from libbinder"
);
drop(_b);
bridge_t.join().expect("bridge thread");
let _ = observer.into_raw_fd();
}
#[test]
fn from_preconnected_fd_rejects_non_socket_fd() {
let fd = rustix::fs::open(
"/dev/null",
rustix::fs::OFlags::RDWR | rustix::fs::OFlags::CLOEXEC,
rustix::fs::Mode::empty(),
)
.expect("open /dev/null");
assert!(
RpcSession::from_preconnected_fd(fd, 2).is_err(),
"non-socket fd must reject before any handshake I/O"
);
}
#[test]
fn dec_strong_never_scans_a_serve_driven_slot() {
use crate::rpc::transport::MemTransport;
let (t0, _p0) = MemTransport::pair();
let session = RpcSession::from_android13plus(
Box::new(t0),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("build session");
assert_eq!(session.inner.slot_count(), 1);
assert!(
matches!(session.inner.dec_route(), DecRoute::Reaper),
"a DEC_STRONG must not claim the free serve-driven slot"
);
assert!(
session.inner.find_conn_for_reaper().is_none(),
"the reaper must skip rather than take it (or park holding the session)"
);
let (t, _p) = MemTransport::pair();
session
.add_callback_slot(Box::new(t), 2)
.expect("callback slot");
assert!(
matches!(session.inner.dec_route(), DecRoute::Conn(_)),
"with an outgoing slot the DEC_STRONG goes out normally"
);
}
#[test]
fn a_oneway_pin_holds_a_bounded_number_of_addresses() {
use super::super::transport::UnixTransport;
let (t0, peer) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::from_android13plus(
Box::new(t0),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("build session");
let inner = &*session.inner;
let slot_id = RpcSession::FOUNDING_SLOT_ID;
let sess_ptr = inner as *const RpcSessionInner as usize;
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
let held = || {
let st = inner.conn_state.lock().expect("conn_state");
st.slots
.iter()
.find(|s| s.id == slot_id)
.map(|s| s.pending_dec.len())
};
let mut counter = 0;
for _ in 0..HELD_DEC_STRONG_LIMIT {
inner.send_dec_strong(RpcAddress::unique(&mut counter, AddressSpace::Initiator), 1);
}
assert_eq!(held(), Some(HELD_DEC_STRONG_LIMIT));
inner.send_dec_strong(RpcAddress::unique(&mut counter, AddressSpace::Initiator), 1);
DRIVING.with(|d| d.borrow_mut().pop());
assert_eq!(
held(),
Some(HELD_DEC_STRONG_LIMIT),
"the extra address is not held"
);
peer.set_read_timeout(Some(Duration::from_secs(5)))
.expect("peer timeout");
let mut buf = [0u8; 64];
assert!(
peer.recv_raw(&mut buf).expect("the DEC_STRONG frame") > 0,
"it went out on the pin"
);
assert!(
!inner.shared.lifecycle.is_torn_down(),
"the session stays up"
);
}
#[test]
fn callback_slot_cap_is_enforced() {
use crate::rpc::transport::MemTransport;
let (t0, _p0) = MemTransport::pair();
let session = RpcSession::from_android13plus(
Box::new(t0),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("build session");
let base = session.inner.slot_count();
let cap = 2;
let mut peers = Vec::new();
let mut admitted = 0usize;
let mut refused = 0usize;
for _ in 0..5 {
let (t, p) = MemTransport::pair();
peers.push(p); match session.add_callback_slot(Box::new(t), cap) {
Ok(_) => admitted += 1,
Err(_) => refused += 1,
}
}
assert_eq!(
session.inner.slot_count(),
base + cap,
"pool never exceeds founding + cap"
);
assert_eq!(admitted, cap, "exactly cap callback slots admitted");
assert!(refused >= 1, "attaches past the cap are refused");
}
#[test]
fn callback_slot_init_failure_retires_the_slot() {
use crate::rpc::transport::MemTransport;
let (t0, _p0) = MemTransport::pair();
let codec = Android13PlusCodec::android14_15();
let session = RpcSession::from_android13plus(Box::new(t0), codec, FD_MODE_NONE, false)
.expect("build session");
let base = session.inner.slot_count();
let (t, p) = MemTransport::pair();
drop(p);
assert!(
session
.add_callback_slot_and_init(Box::new(t), 2, &codec)
.is_err(),
"a failed connection-init must be reported"
);
assert_eq!(
session.inner.slot_count(),
base,
"a failed connection-init must leave no slot behind"
);
let mut peers = Vec::new();
for _ in 0..2 {
let (t, p) = MemTransport::pair();
peers.push(p);
session
.add_callback_slot(Box::new(t), 2)
.expect("callback budget still has room");
}
assert_eq!(session.inner.slot_count(), base + 2);
}
#[test]
fn a_sender_racing_a_failed_callback_init_never_picks_its_slot() {
use crate::rpc::transport::UnixTransport;
let (t0, _p0) = UnixTransport::pair().expect("socketpair");
let codec = Android13PlusCodec::android14_15();
let session = RpcSession::from_android13plus(Box::new(t0), codec, FD_MODE_NONE, false)
.expect("build session");
let base = session.inner.slot_count();
session.set_timeout(Some(Duration::from_millis(300)));
let (t, p) = UnixTransport::pair().expect("socketpair");
drop(p);
let racer_inner = Arc::clone(&session.inner);
let (raced_tx, raced_rx) = mpsc::channel();
let hook = move || {
let raced = std::thread::spawn(move || {
racer_inner
.client_transact(RpcAddress::zero(), 1, &Parcel::new(), FLAG_ONEWAY)
.map(|_| ())
})
.join()
.expect("racer");
let _ = raced_tx.send(raced);
};
*session
.inner
.shared
.cci_failed_hook
.lock()
.expect("cci hook") = Some(Box::new(hook));
assert!(
session
.add_callback_slot_and_init(Box::new(t), 2, &codec)
.is_err(),
"a failed connection-init must be reported"
);
let raced = raced_rx
.recv_timeout(Duration::from_secs(10))
.expect("the hook ran");
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"a retired callback slot must not end the session"
);
assert_eq!(
raced,
Err(StatusCode::WouldBlock),
"the racer must find no free slot, not the one being retired"
);
assert_eq!(session.inner.slot_count(), base, "the slot left the pool");
}
#[test]
fn an_incoming_connection_whose_thread_fails_to_spawn_ends_the_session() {
use super::super::transport::UnixTransport;
use crate::rpc::wire_android13::server_accept;
use std::os::unix::net::UnixStream;
let (client_fd, founding_peer_fd) = unix_socketpair_fd();
let session = RpcSession::with_profile(
Box::new(UnixTransport::from_stream(UnixStream::from(client_fd)).expect("transport")),
AddressSpace::Initiator,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("session");
seed_server_id(&session);
let mut founding_peer = UnixStream::from(founding_peer_fd);
founding_peer
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("read timeout");
let (attach_fd, server_fd) = unix_socketpair_fd();
let server = std::thread::spawn(move || {
let mut s = UnixStream::from(server_fd);
server_accept(&mut s, PROTOCOL_V2).map(|_| s)
});
let attach = UnixTransport::from_stream(UnixStream::from(attach_fd)).expect("transport");
session
.inner
.shared
.fail_incoming_spawn
.store(true, Ordering::SeqCst);
let added = session.add_incoming_connection_android13plus_transport(
move || Ok(Box::new(attach) as Box<dyn RpcTransport>),
PROTOCOL_V2,
&[7u8; 32],
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
let _server_side = server.join().expect("server half").expect("handshake");
assert!(added.is_err(), "the spawn failure is reported");
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"the server pooled the connection as a callback slot, so the session must end"
);
assert_eq!(session.inner.slot_count(), 0, "the dead pool is empty");
assert_eq!(session.inner.incoming_live.load(Ordering::SeqCst), 0);
let mut byte = [0u8; 1];
assert_eq!(
std::io::Read::read(&mut founding_peer, &mut byte).expect("read"),
0,
"the founding connection's peer sees the session end"
);
}
fn v2_initiator() -> (RpcSession, std::os::unix::net::UnixStream) {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
let (client_fd, founding_peer_fd) = unix_socketpair_fd();
let session = RpcSession::with_profile(
Box::new(UnixTransport::from_stream(UnixStream::from(client_fd)).expect("transport")),
AddressSpace::Initiator,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("session");
seed_server_id(&session);
(session, UnixStream::from(founding_peer_fd))
}
fn seed_server_id(session: &RpcSession) {
*session.server_id_lock() = Some(vec![7u8; 32]);
}
fn incoming_attach_against(
session: &RpcSession,
max_version: u32,
deadline: Duration,
wrap: impl FnOnce(super::super::transport::UnixTransport) -> Box<dyn RpcTransport>,
server: impl FnOnce(std::os::unix::net::UnixStream) + Send + 'static,
) -> Result<u64> {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
let (attach_fd, server_fd) = unix_socketpair_fd();
let server_side = armed_server_side(server_fd);
let server = std::thread::spawn(move || server(server_side));
let attach = wrap(UnixTransport::from_stream(UnixStream::from(attach_fd)).expect("unix"));
let added = session.add_incoming_connection_android13plus_transport(
move || Ok(attach),
max_version,
&[7u8; 32],
FileDescriptorTransportMode::None,
Some(deadline),
);
server.join().expect("server half");
added
}
fn armed_server_side(server_fd: OwnedFd) -> std::os::unix::net::UnixStream {
let s = std::os::unix::net::UnixStream::from(server_fd);
s.set_read_timeout(Some(Duration::from_secs(10)))
.expect("read timeout");
s
}
fn hold_until_client_closes(mut s: std::os::unix::net::UnixStream) {
let mut rest = Vec::new();
if let Err(e) = std::io::Read::read_to_end(&mut s, &mut rest) {
assert!(
!matches!(
e.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
),
"the client kept the connection for 10 s: its attach must have pooled it"
);
}
}
fn cut_inside_cci(mut s: std::os::unix::net::UnixStream) {
use crate::rpc::wire_android13::server_accept_deferred_init;
let (codec, ..) = server_accept_deferred_init(&mut s, PROTOCOL_V2).expect("server header");
std::io::Write::write_all(&mut s, &codec.encode_connection_init()[..1]).expect("cci byte");
}
struct NoCloseNotify(super::super::transport::UnixTransport);
impl RpcTransport for NoCloseNotify {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.0.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.0.recv_frame()
}
fn peer_identity(&self) -> PeerIdentity {
self.0.peer_identity()
}
fn describe(&self) -> &str {
"no-close-notify"
}
fn supports_fd_passing(&self) -> bool {
self.0.supports_fd_passing()
}
fn set_read_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
self.0.set_read_timeout(timeout)
}
fn set_write_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
self.0.set_write_timeout(timeout)
}
fn shutdown(&self) -> RpcResult<()> {
self.0.shutdown()
}
fn send_raw(&self, buf: &[u8]) -> RpcResult<()> {
self.0.send_raw(buf)
}
fn recv_raw(&self, buf: &mut [u8]) -> RpcResult<usize> {
match self.0.recv_raw(buf) {
Ok(0) | Err(RpcError::EndOfStream) => Err(RpcError::UncleanEndOfStream),
other => other,
}
}
}
struct NoFds(super::super::transport::UnixTransport);
impl RpcTransport for NoFds {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.0.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.0.recv_frame()
}
fn peer_identity(&self) -> PeerIdentity {
self.0.peer_identity()
}
fn describe(&self) -> &str {
"no-fds"
}
fn set_read_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
self.0.set_read_timeout(timeout)
}
fn set_write_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
self.0.set_write_timeout(timeout)
}
fn shutdown(&self) -> RpcResult<()> {
self.0.shutdown()
}
fn send_raw(&self, buf: &[u8]) -> RpcResult<()> {
self.0.send_raw(buf)
}
fn recv_raw(&self, buf: &mut [u8]) -> RpcResult<usize> {
self.0.recv_raw(buf)
}
}
fn expect_no_header(mut s: std::os::unix::net::UnixStream) {
let mut rest = Vec::new();
std::io::Read::read_to_end(&mut s, &mut rest).expect("the client closes");
assert!(rest.is_empty(), "the client sent {} bytes", rest.len());
}
#[test]
fn an_incoming_attach_below_the_session_version_never_connects() {
let (session, _founding_peer) = v2_initiator();
let added = session.add_incoming_connection_android13plus_transport(
|| -> Result<Box<dyn RpcTransport>> { panic!("the attach must not connect") },
PROTOCOL_V1,
&[7u8; 32],
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
assert_eq!(added, Err(StatusCode::BadType));
assert!(!session.inner.shared.lifecycle.is_torn_down());
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
}
#[test]
fn an_incoming_attach_on_another_transport_kind_is_refused_before_its_header() {
let (session, _founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_secs(5),
|t| Box::new(NoFds(t)),
expect_no_header,
);
assert_eq!(added, Err(StatusCode::BadType));
assert!(!session.inner.shared.lifecycle.is_torn_down());
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
}
#[test]
fn an_incoming_attach_whose_deadline_expires_before_cci_ends_the_session() {
use crate::rpc::wire_android13::server_accept_deferred_init;
let (session, _founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_millis(200),
|t| Box::new(t),
|mut s| {
server_accept_deferred_init(&mut s, PROTOCOL_V2).expect("server header");
hold_until_client_closes(s);
},
);
assert_eq!(added, Err(StatusCode::TimedOut));
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"the client cannot tell whether the server pooled it, so the session must end"
);
assert_eq!(session.inner.slot_count(), 0, "the dead pool is empty");
}
#[test]
fn an_incoming_attach_closed_before_cci_leaves_the_session_up() {
use super::super::transport::UnixTransport;
use crate::rpc::wire_android13::server_accept_deferred_init;
let (session, founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_secs(5),
|t| Box::new(t),
|mut s| {
server_accept_deferred_init(&mut s, PROTOCOL_V2).expect("server header");
},
);
assert_eq!(added, Err(StatusCode::DeadObject));
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"a server that closed before \"cci\" holds no slot, so the session goes on"
);
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
let server = RpcSession::with_profile(
Box::new(UnixTransport::from_stream(founding_peer).expect("transport")),
AddressSpace::Acceptor,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("server session");
let serving = std::thread::spawn(move || server.serve_blocking());
session.set_timeout(Some(Duration::from_secs(5)));
let reply = session.inner.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
);
assert!(
reply.is_ok(),
"the founding connection still serves: {reply:?}"
);
session.close_session();
let _end = serving.join().expect("server loop");
}
#[test]
fn an_android13plus_server_refuses_get_fd_mode() {
use super::super::transport::UnixTransport;
let (a, b) = UnixTransport::pair().expect("socketpair");
let v1 = || WireProfile::Android13Plus(Android13PlusCodec::android14_15());
let server = RpcSession::with_profile(Box::new(a), AddressSpace::Acceptor, v1())
.expect("server session");
server.set_supported_fd_modes(&[FileDescriptorTransportMode::Unix]);
let server_inner = Arc::clone(&server.inner);
let serving = std::thread::spawn(move || server.serve_blocking());
let client = RpcSession::with_profile(Box::new(b), AddressSpace::Initiator, v1())
.expect("client session");
client.set_timeout(Some(Duration::from_secs(5)));
let mut req = Parcel::new();
req.write(&1i32).expect("want Unix");
let reply = client.inner.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetFdMode.code(),
&req,
0,
);
assert_eq!(reply.err(), Some(StatusCode::UnknownTransaction));
assert_eq!(server_inner.fd_mode(), FileDescriptorTransportMode::None);
assert_eq!(
client.negotiate_fd_transport(FileDescriptorTransportMode::Unix),
Err(StatusCode::InvalidOperation)
);
client.close_session();
let _end = serving.join().expect("server loop");
}
#[test]
fn android13plus_negotiate_fd_transport_sends_nothing() {
use super::super::transport::UnixTransport;
use std::io::Read;
let (ours, mut peer) = std::os::unix::net::UnixStream::pair().expect("socketpair");
let v1 = WireProfile::Android13Plus(Android13PlusCodec::android14_15());
let t = UnixTransport::from_stream(ours).expect("transport");
let client = RpcSession::with_profile(Box::new(t), AddressSpace::Initiator, v1)
.expect("client session");
client.set_timeout(Some(Duration::from_millis(200)));
assert_eq!(
client.negotiate_fd_transport(FileDescriptorTransportMode::None),
Ok(FileDescriptorTransportMode::None)
);
assert_eq!(
client.negotiate_fd_transport(FileDescriptorTransportMode::Unix),
Err(StatusCode::InvalidOperation)
);
assert_eq!(
client.fd_transport_mode(),
FileDescriptorTransportMode::None
);
peer.set_nonblocking(true).expect("nonblocking");
let pending = peer.read(&mut [0u8; 1]);
assert!(
matches!(&pending, Err(e) if e.kind() == std::io::ErrorKind::WouldBlock),
"{pending:?}"
);
client.close_session();
}
#[test]
fn r34_client_opens_with_the_preamble_and_an_unprefixed_get_root() {
let (session, mut peer) = r34_client_with_raw_peer();
session.set_timeout(Some(Duration::from_secs(5)));
let call = std::thread::spawn(move || session.get_root().map(|_| ()));
let mut got = [0u8; 4 + 16 + 64];
std::io::Read::read_exact(&mut peer, &mut got).expect("the client's first bytes");
let mut want = vec![0xff; 4]; want.extend_from_slice(&[0, 0, 0, 0]); want.extend_from_slice(&[64, 0, 0, 0]); want.extend_from_slice(&[0; 8]); want.extend_from_slice(&[0; 32]); want.extend_from_slice(&[0, 0, 0, 0]); want.extend_from_slice(&[0; 4 + 8 + 16]); assert_eq!(got.to_vec(), want);
drop(peer);
assert!(call.join().expect("caller").is_err(), "the peer closed");
}
#[test]
fn r34_acceptor_takes_the_preamble_then_unprefixed_messages() {
use super::super::transport::UnixTransport;
let (a, peer) = UnixTransport::pair().expect("socketpair");
let server = RpcSession::new(Box::new(a), AddressSpace::Acceptor).expect("session");
let serving = std::thread::spawn(move || server.serve_blocking());
let get_root = R34Codec
.encode_transact(&WireTransaction {
address: RpcAddress::zero(),
code: SpecialTransaction::GetRoot.code(),
..WireTransaction::default()
})
.expect("GET_ROOT");
peer.send_raw(&(-1i32).to_le_bytes()).expect("preamble");
peer.send_raw(&get_root).expect("GET_ROOT");
let reply = read_aosp_message(&mut RawTransportIo(&peer)).expect("reply");
match R34Codec.decode_message(&reply).expect("decode") {
WireMessage::Reply(r) => {
assert_eq!(r.status, 0);
assert_eq!(r.data, [0, 0, 0, 0, 1, 0, 0, 0]);
}
other => panic!("expected a REPLY, got {other:?}"),
}
drop(peer);
assert_eq!(
serving.join().expect("serve").reason,
EndReason::EndOfStream
);
}
#[test]
fn r34_client_reads_an_aosp12_root_reply() {
let addr = [0x5a_u8; RPC_ADDR_LEN];
for (stability, accepted) in [(0x0c00_0001_i32, true), (0x0c00_0000, false)] {
let (session, mut peer) = r34_client_with_raw_peer();
session.set_timeout(Some(Duration::from_secs(5)));
peer.set_read_timeout(Some(Duration::from_secs(5)))
.expect("peer read timeout");
let inner = Arc::clone(&session.inner);
let call = std::thread::spawn(move || {
let mut reply = inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetRoot.code(),
&Parcel::new(),
0,
)?
.ok_or(StatusCode::UnexpectedNull)?;
let root = reply.read::<SIBinder>();
let at_end = reply.data_position() == reply.data_size();
root.map(|_| at_end)
});
assert_eq!(
read_r34_session_preamble(&mut peer).expect("preamble"),
RPC_SESSION_ID_NEW
);
read_aosp_message(&mut peer).expect("GET_ROOT");
let mut data = 1i32.to_le_bytes().to_vec();
data.extend_from_slice(&addr);
data.extend_from_slice(&stability.to_le_bytes());
let reply = R34Codec
.encode_reply(&WireReply {
status: 0,
data,
..WireReply::default()
})
.expect("reply");
std::io::Write::write_all(&mut peer, &reply).expect("reply");
let got = call.join().expect("caller");
if accepted {
assert_eq!(got, Ok(true), "a proxy, and the cursor past the stability");
} else {
assert_eq!(got, Err(StatusCode::BadType), "stability {stability:#x}");
let release = read_aosp_message(&mut peer).expect("the refused root's release");
match R34Codec.decode_message(&release).expect("decode") {
WireMessage::DecStrong(a, 1) => assert_eq!(*a.as_wire_bytes(), addr),
other => panic!("expected a DEC_STRONG, got {other:?}"),
}
}
}
}
#[test]
fn a_bare_r34_acceptor_has_no_session_id() {
use super::super::transport::MemTransport;
let (a, b) = MemTransport::pair();
let server = RpcSession::new(Box::new(a), AddressSpace::Acceptor).expect("server");
let serving = std::thread::spawn(move || server.serve_blocking());
let client = RpcSession::new(Box::new(b), AddressSpace::Initiator).expect("client");
client.set_timeout(Some(Duration::from_secs(5)));
assert_eq!(client.get_session_id(), Err(StatusCode::UnknownTransaction));
client.close_session();
assert!(
serving.join().expect("serve").is_clean(),
"the client closed"
);
}
#[test]
fn r34_acceptor_refuses_any_preamble_but_new() {
use super::super::transport::UnixTransport;
for first in [5i32.to_le_bytes(), 80u32.to_le_bytes()] {
let (a, peer) = UnixTransport::pair().expect("socketpair");
let server = RpcSession::new(Box::new(a), AddressSpace::Acceptor).expect("session");
peer.send_raw(&first).expect("preamble");
let end = server.serve_blocking();
assert!(
matches!(end.reason, EndReason::Frame(_)),
"refused, got {:?}",
end.reason
);
assert_eq!(peer.recv_raw(&mut [0u8; 4]).expect("closed"), 0);
}
}
#[test]
fn r34_get_fd_mode_unknown_to_the_peer_negotiates_none() {
use super::super::transport::UnixTransport;
let (a, peer) = UnixTransport::pair().expect("socketpair");
let client = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
client.set_timeout(Some(Duration::from_secs(5)));
let answering = std::thread::spawn(move || {
let mut io = RawTransportIo(&peer);
let id = read_r34_session_preamble(&mut io).expect("preamble");
assert_eq!(id, RPC_SESSION_ID_NEW);
let frame = read_aosp_message(&mut io).expect("GET_FD_MODE");
match R34Codec.decode_message(&frame).expect("decode") {
WireMessage::Transact(t) => {
assert_eq!(t.code, SpecialTransaction::GetFdMode.code())
}
other => panic!("expected GET_FD_MODE, got {other:?}"),
}
let reply = WireReply {
status: StatusCode::UnknownTransaction.into(),
..WireReply::default()
};
let reply = R34Codec.encode_reply(&reply).expect("encode");
peer.send_raw(&reply).expect("reply");
peer
});
assert_eq!(
client.negotiate_fd_transport(FileDescriptorTransportMode::Unix),
Ok(FileDescriptorTransportMode::None)
);
assert_eq!(
client.fd_transport_mode(),
FileDescriptorTransportMode::None
);
let _peer = answering.join().expect("peer");
client.close_session();
}
#[test]
fn an_incoming_attach_cut_inside_cci_ends_the_session() {
use super::super::transport::UnixTransport;
type Wrap = fn(UnixTransport) -> Box<dyn RpcTransport>;
let wraps: [(&str, Wrap); 2] = [
("EOF", |t| Box::new(t)),
("no close_notify", |t| Box::new(NoCloseNotify(t))),
];
for (close, wrap) in wraps {
let (session, _founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_secs(5),
wrap,
cut_inside_cci,
);
assert!(added.is_err(), "{close}: {added:?}");
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"{close} after one \"cci\" byte: the server pooled it, so the session must end"
);
assert_eq!(
session.inner.slot_count(),
0,
"{close}: the dead pool is empty"
);
}
}
#[test]
fn an_incoming_attach_uncleanly_closed_before_cci_leaves_the_session_up() {
use crate::rpc::wire_android13::server_accept_deferred_init;
let (session, _founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_secs(5),
|t| Box::new(NoCloseNotify(t)),
|mut s| {
server_accept_deferred_init(&mut s, PROTOCOL_V2).expect("server header");
},
);
assert!(added.is_err(), "{added:?}");
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"a server that closed before any \"cci\" byte holds no slot, so the session goes on"
);
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
}
#[test]
fn an_incoming_attach_reset_before_cci_leaves_the_session_up() {
#[cfg(any(target_os = "linux", target_os = "android"))]
{
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
let (mut writer, reader) = UnixStream::pair().expect("pair");
writer.write_all(b"header").expect("write");
drop(reader);
let premise = writer.read(&mut [0u8; 1]).map_err(|e| e.kind());
assert_eq!(premise, Err(std::io::ErrorKind::ConnectionReset));
}
let (session, _founding_peer) = v2_initiator();
let added = incoming_attach_against(
&session,
PROTOCOL_V2,
Duration::from_secs(5),
|t| Box::new(t),
|s| {
let mut probe = [0u8; 1];
rustix::net::recv(&s, &mut probe[..], rustix::net::RecvFlags::PEEK).expect("peek");
},
);
assert_eq!(added, Err(StatusCode::DeadObject));
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"a reset before any \"cci\" byte may be a refusal, so the session goes on"
);
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
}
fn outgoing_attach_against(
session: &RpcSession,
deadline: Duration,
wrap: impl FnOnce(super::super::transport::UnixTransport) -> Box<dyn RpcTransport>,
server: impl FnOnce(std::os::unix::net::UnixStream) + Send + 'static,
) -> Result<u64> {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
let (attach_fd, server_fd) = unix_socketpair_fd();
let server_side = armed_server_side(server_fd);
let server = std::thread::spawn(move || server(server_side));
let attach = wrap(UnixTransport::from_stream(UnixStream::from(attach_fd)).expect("unix"));
let added = session.add_outgoing_connection_android13plus_transport(
move || Ok(attach),
PROTOCOL_V2,
&[7u8; 32],
FileDescriptorTransportMode::None,
Some(deadline),
);
server.join().expect("server half");
added
}
fn read_probe(s: &mut std::os::unix::net::UnixStream) -> Android13PlusCodec {
use crate::rpc::wire_android13::server_accept;
let (codec, ..) = server_accept(s, PROTOCOL_V2).expect("server handshake");
read_aosp_message(s).expect("probe");
codec
}
fn answer_probe(s: &mut std::os::unix::net::UnixStream, id: &[u8]) {
let codec = read_probe(s);
let mut p = Parcel::new();
p.write(id).expect("id");
let reply = WireReply {
status: 0,
data: p.rpc_data_bytes().to_vec(),
object_positions: Vec::new(),
};
write_aosp_message(s, &codec.encode_reply(&reply).expect("encode")).expect("reply");
}
struct FailsAt {
t: super::super::transport::UnixTransport,
send: usize,
arm: usize,
sends: AtomicUsize,
arms: AtomicUsize,
err: fn() -> RpcError,
}
impl FailsAt {
fn new(t: super::super::transport::UnixTransport, send: usize, arm: usize) -> Self {
FailsAt {
t,
send,
arm,
sends: AtomicUsize::new(0),
arms: AtomicUsize::new(0),
err: Self::injected,
}
}
fn failing_with(mut self, err: fn() -> RpcError) -> Self {
self.err = err;
self
}
fn injected() -> RpcError {
RpcError::Io(std::io::Error::other("injected"))
}
}
impl RpcTransport for FailsAt {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.t.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.t.recv_frame()
}
fn peer_identity(&self) -> PeerIdentity {
self.t.peer_identity()
}
fn describe(&self) -> &str {
"fails-at"
}
fn supports_fd_passing(&self) -> bool {
self.t.supports_fd_passing()
}
fn set_read_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
if timeout.is_some() && self.arms.fetch_add(1, Ordering::SeqCst) == self.arm {
return Err(Self::injected());
}
self.t.set_read_timeout(timeout)
}
fn set_write_timeout(&self, timeout: Option<Duration>) -> RpcResult<()> {
self.t.set_write_timeout(timeout)
}
fn shutdown(&self) -> RpcResult<()> {
self.t.shutdown()
}
fn send_raw(&self, buf: &[u8]) -> RpcResult<()> {
if self.sends.fetch_add(1, Ordering::SeqCst) == self.send {
return Err((self.err)());
}
self.t.send_raw(buf)
}
fn recv_raw(&self, buf: &mut [u8]) -> RpcResult<usize> {
self.t.recv_raw(buf)
}
}
fn assert_attach_outcome(session: &RpcSession, added: Result<u64>, ends: bool, why: &str) {
assert!(added.is_err(), "{why}: {added:?}");
assert_eq!(
session.inner.shared.lifecycle.is_torn_down(),
ends,
"{why}: the session must {}",
if ends { "end" } else { "go on" }
);
assert_eq!(
session.inner.slot_count(),
usize::from(!ends),
"{why}: pool size"
);
}
#[test]
fn an_outgoing_attach_below_the_session_version_never_connects() {
let (session, _founding_peer) = v2_initiator();
let added = session.add_outgoing_connection_android13plus_transport(
|| -> Result<Box<dyn RpcTransport>> { panic!("the attach must not connect") },
PROTOCOL_V1,
&[7u8; 32],
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
assert_eq!(added, Err(StatusCode::BadType));
assert_attach_outcome(&session, added, false, "below the session version");
}
#[test]
fn an_outgoing_attach_whose_header_write_fails_leaves_the_session_up() {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(
&session,
Duration::from_secs(5),
|t| Box::new(FailsAt::new(t, 0, usize::MAX)),
hold_until_client_closes,
);
assert_attach_outcome(&session, added, false, "header write failed");
}
#[test]
fn an_outgoing_attach_failing_past_its_header_ends_the_session() {
use super::super::transport::UnixTransport;
type Wrap = fn(UnixTransport) -> Box<dyn RpcTransport>;
let cases: [(&str, Wrap, Duration); 3] = [
(
"\"cci\" write",
|t| Box::new(FailsAt::new(t, 1, usize::MAX)),
Duration::from_secs(5),
),
(
"probe deadline arm",
|t| Box::new(FailsAt::new(t, usize::MAX, 1)),
Duration::from_secs(5),
),
(
"probe deadline expiry",
|t| Box::new(t),
Duration::from_millis(200),
),
];
for (why, wrap, deadline) in cases {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(&session, deadline, wrap, |mut s| {
let _ = crate::rpc::wire_android13::server_accept(&mut s, PROTOCOL_V2);
hold_until_client_closes(s);
});
assert_attach_outcome(&session, added, true, why);
}
}
#[test]
fn an_outgoing_attach_closed_before_the_reply_ends_the_session() {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
type Wrap = fn(UnixTransport) -> Box<dyn RpcTransport>;
type Server = fn(UnixStream);
fn closes_after_the_probe(mut s: UnixStream) {
read_probe(&mut s);
}
fn holds_after_the_header(mut s: UnixStream) {
let _ = crate::rpc::wire_android13::server_accept(&mut s, PROTOCOL_V2);
hold_until_client_closes(s);
}
let cases: [(&str, Wrap, Server); 4] = [
("EOF", |t| Box::new(t), closes_after_the_probe),
(
"no close_notify",
|t| Box::new(NoCloseNotify(t)),
closes_after_the_probe,
),
(
"close at the \"cci\" write",
|t| Box::new(FailsAt::new(t, 1, usize::MAX).failing_with(|| RpcError::EndOfStream)),
holds_after_the_header,
),
(
"close at the probe write",
|t| Box::new(FailsAt::new(t, 2, usize::MAX).failing_with(|| RpcError::EndOfStream)),
holds_after_the_header,
),
];
for (why, wrap, server) in cases {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(&session, Duration::from_secs(5), wrap, server);
assert_attach_outcome(&session, added, true, why);
}
}
#[test]
fn an_outgoing_attach_cut_inside_the_reply_ends_the_session() {
use super::super::transport::UnixTransport;
type Wrap = fn(UnixTransport) -> Box<dyn RpcTransport>;
let wraps: [(&str, Wrap); 2] = [
("EOF", |t| Box::new(t)),
("no close_notify", |t| Box::new(NoCloseNotify(t))),
];
for (why, wrap) in wraps {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(&session, Duration::from_secs(5), wrap, |mut s| {
read_probe(&mut s);
std::io::Write::write_all(&mut s, &[1u8]).expect("reply byte");
});
assert_attach_outcome(&session, added, true, why);
}
}
#[test]
fn an_outgoing_attach_refused_after_the_reply_ends_the_session() {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(
&session,
Duration::from_secs(5),
|t| Box::new(t),
|mut s| {
answer_probe(&mut s, &[8u8; 32]);
hold_until_client_closes(s);
},
);
assert_attach_outcome(&session, added, true, "another session's id");
}
#[test]
fn an_outgoing_attach_on_another_transport_kind_is_refused_before_its_header() {
let (session, _founding_peer) = v2_initiator();
let added = outgoing_attach_against(
&session,
Duration::from_secs(5),
|t| Box::new(NoFds(t)),
expect_no_header,
);
assert_eq!(added, Err(StatusCode::BadType));
assert_attach_outcome(&session, added, false, "another transport kind");
}
#[test]
fn an_outgoing_attach_echoing_the_client_local_id_never_connects() {
let (session, _founding_peer) = v2_initiator();
let added = session.add_outgoing_connection_android13plus_transport(
|| -> Result<Box<dyn RpcTransport>> { panic!("the attach must not connect") },
PROTOCOL_V2,
&session.session_id(),
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
assert_eq!(added, Err(StatusCode::BadValue));
assert_attach_outcome(&session, added, false, "client-local id");
}
#[test]
fn an_attach_naming_another_server_session_never_connects() {
let (session, _founding_peer) = v2_initiator();
let added = session.add_outgoing_connection_android13plus_transport(
|| -> Result<Box<dyn RpcTransport>> { panic!("the attach must not connect") },
PROTOCOL_V2,
&[9u8; 32],
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
assert_eq!(added, Err(StatusCode::BadValue));
assert_attach_outcome(&session, added, false, "foreign id, outgoing");
let added = session.add_incoming_connection_android13plus_transport(
|| -> Result<Box<dyn RpcTransport>> { panic!("the attach must not connect") },
PROTOCOL_V2,
&[9u8; 32],
FileDescriptorTransportMode::None,
Some(Duration::from_secs(5)),
);
assert_eq!(added, Err(StatusCode::BadValue));
assert_attach_outcome(&session, added, false, "foreign id, incoming");
}
#[test]
fn callback_slot_refused_on_torn_down_session() {
use crate::rpc::transport::MemTransport;
let (t0, _p0) = MemTransport::pair();
let codec = Android13PlusCodec::android14_15();
let session = RpcSession::from_android13plus(Box::new(t0), codec, FD_MODE_NONE, false)
.expect("build session");
session.close_session();
let (t, _p) = MemTransport::pair();
assert_eq!(
session.inner.add_slot_inner_capped(Box::new(t), 2),
Err(StatusCode::DeadObject),
"no slot may be pushed onto a torn-down session"
);
assert_eq!(session.inner.slot_count(), 0, "the dead pool stays empty");
}
#[test]
fn a_failed_send_ends_the_whole_session() {
use super::super::transport::UnixTransport;
let (a, pa) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
let (b, _pb) = UnixTransport::pair().expect("socketpair");
session
.inner
.add_outgoing_slot(Box::new(b))
.expect("fan-out slot");
drop(pa);
let err = session
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
)
.expect_err("the send must fail");
assert_eq!(err, StatusCode::DeadObject);
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"the other outgoing connection does not keep the session up"
);
assert_eq!(session.inner.slot_count(), 0);
}
#[test]
fn liveness_follows_the_session_values_on_every_slot() {
#[derive(Default)]
struct Seen {
send: Mutex<Vec<Option<Duration>>>,
live: Mutex<Vec<Option<Duration>>>,
}
impl Seen {
fn last(&self) -> (Option<Duration>, Option<Duration>) {
let send = *self
.send
.lock()
.unwrap()
.last()
.expect("send deadline armed");
let live = *self.live.lock().unwrap().last().expect("liveness armed");
(send, live)
}
}
struct Recorder(Arc<Seen>);
impl RpcTransport for Recorder {
fn send_frame(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
Err(RpcError::EndOfStream)
}
fn peer_identity(&self) -> PeerIdentity {
PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"recorder"
}
fn send_raw(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn set_write_timeout(&self, t: Option<Duration>) -> RpcResult<()> {
self.0.send.lock().unwrap().push(t);
Ok(())
}
fn set_liveness(&self, t: Option<Duration>) -> RpcResult<()> {
self.0.live.lock().unwrap().push(t);
Ok(())
}
fn shutdown(&self) -> RpcResult<()> {
Ok(())
}
}
let secs = |n| Some(Duration::from_secs(n));
let founding = Arc::new(Seen::default());
let session = RpcSession::new(
Box::new(Recorder(Arc::clone(&founding))),
AddressSpace::Initiator,
)
.expect("session");
assert_eq!(founding.last(), (None, None), "keepalive with no timeout");
session.set_timeout(secs(8));
assert_eq!(founding.last(), (secs(8), secs(8)));
session.set_serve_read_deadline(secs(3));
assert_eq!(founding.last(), (secs(3), secs(8)));
let added = Arc::new(Seen::default());
session
.inner
.add_outgoing_slot(Box::new(Recorder(Arc::clone(&added))))
.expect("fan-out slot");
assert_eq!(added.last(), (secs(3), secs(8)), "a slot added later");
session.set_timeout(None);
assert_eq!(founding.last(), (secs(3), None));
assert_eq!(added.last(), (secs(3), None));
}
#[test]
fn a_slot_armed_across_a_set_timeout_ends_with_the_stored_value() {
#[derive(Default)]
struct Gate {
send: Mutex<Vec<Option<Duration>>>,
entered: Mutex<Option<mpsc::Sender<()>>>,
release: Mutex<Option<mpsc::Receiver<()>>>,
}
struct Held(Arc<Gate>);
impl RpcTransport for Held {
fn send_frame(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
Err(RpcError::EndOfStream)
}
fn peer_identity(&self) -> PeerIdentity {
PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"held"
}
fn send_raw(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn set_write_timeout(&self, t: Option<Duration>) -> RpcResult<()> {
let entered = self.0.entered.lock().unwrap().take();
if let Some(entered) = entered {
let _ = entered.send(());
let release = self.0.release.lock().unwrap().take();
let _ = release.expect("release").recv();
}
self.0.send.lock().unwrap().push(t);
Ok(())
}
fn shutdown(&self) -> RpcResult<()> {
Ok(())
}
}
let secs = |n| Some(Duration::from_secs(n));
let session = RpcSession::new(
Box::new(Held(Arc::new(Gate::default()))),
AddressSpace::Initiator,
)
.expect("session");
session.set_timeout(secs(8));
let (entered_tx, entered) = mpsc::channel();
let (release, release_rx) = mpsc::channel();
let added = Arc::new(Gate {
entered: Mutex::new(Some(entered_tx)),
release: Mutex::new(Some(release_rx)),
..Gate::default()
});
let push = {
let inner = Arc::clone(&session.inner);
let slot = Box::new(Held(Arc::clone(&added)));
std::thread::spawn(move || inner.add_outgoing_slot(slot).map(|_| ()))
};
entered
.recv_timeout(Duration::from_secs(10))
.expect("the push must arm its slot");
let (stored_tx, stored) = mpsc::channel();
{
let session = session.clone();
std::thread::spawn(move || {
session.set_timeout(secs(3));
let _ = stored_tx.send(());
});
}
assert!(
stored.recv_timeout(Duration::from_millis(300)).is_err(),
"the store must wait for the slot being armed"
);
release.send(()).expect("release");
push.join().expect("push").expect("fan-out slot");
stored
.recv_timeout(Duration::from_secs(10))
.expect("set_timeout must finish");
assert_eq!(added.send.lock().unwrap().last(), Some(&secs(3)));
}
#[test]
fn a_set_timeout_during_teardown_leaves_the_shutdown_send_bound() {
const CLOSE_BOUND: Option<Duration> = Some(Duration::from_millis(500));
#[derive(Default)]
struct Gate {
send: Mutex<Vec<Option<Duration>>>,
at_write: Mutex<Option<Option<Duration>>>,
entered: Mutex<Option<mpsc::Sender<()>>>,
release: Mutex<Option<mpsc::Receiver<()>>>,
}
struct Closing(Arc<Gate>);
impl RpcTransport for Closing {
fn send_frame(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
Err(RpcError::EndOfStream)
}
fn peer_identity(&self) -> PeerIdentity {
PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"closing"
}
fn send_raw(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn set_write_timeout(&self, t: Option<Duration>) -> RpcResult<()> {
self.0.send.lock().unwrap().push(t);
Ok(())
}
fn shutdown(&self) -> RpcResult<()> {
self.set_write_timeout(CLOSE_BOUND)?;
let entered = self.0.entered.lock().unwrap().take();
if let Some(entered) = entered {
let _ = entered.send(());
let release = self.0.release.lock().unwrap().take();
let _ = release.expect("release").recv();
}
let now = *self.0.send.lock().unwrap().last().expect("bound set");
*self.0.at_write.lock().unwrap() = Some(now);
Ok(())
}
}
let (entered_tx, entered) = mpsc::channel();
let (release, release_rx) = mpsc::channel();
let gate = Arc::new(Gate {
entered: Mutex::new(Some(entered_tx)),
release: Mutex::new(Some(release_rx)),
..Gate::default()
});
let session = RpcSession::new(
Box::new(Closing(Arc::clone(&gate))),
AddressSpace::Initiator,
)
.expect("session");
let teardown = {
let inner = Arc::clone(&session.inner);
std::thread::spawn(move || inner.fail_session())
};
entered
.recv_timeout(Duration::from_secs(10))
.expect("teardown must shut the transport down");
let (stored_tx, stored) = mpsc::channel();
{
let session = session.clone();
std::thread::spawn(move || {
session.set_timeout(None);
let _ = stored_tx.send(());
});
}
let early = stored.recv_timeout(Duration::from_millis(300)).is_ok();
release.send(()).expect("release");
teardown.join().expect("teardown");
if !early {
stored
.recv_timeout(Duration::from_secs(10))
.expect("set_timeout must finish");
}
assert_eq!(
*gate.at_write.lock().unwrap(),
Some(CLOSE_BOUND),
"the closing write must run under shutdown's own bound"
);
}
#[test]
fn a_send_the_peer_stops_reading_ends_the_session() {
use super::super::transport::UnixTransport;
let (a, _unread) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
session.set_timeout(Some(Duration::from_millis(300)));
let mut data = Parcel::new();
data.write(&vec![0u8; 8 << 20]).expect("payload");
let (tx, rx) = mpsc::channel();
let inner = Arc::clone(&session.inner);
std::thread::spawn(move || {
let r = inner.client_transact(RpcAddress::zero(), 1, &data, FLAG_ONEWAY);
let _ = tx.send(r.map(|_| ()));
});
let sent = rx
.recv_timeout(Duration::from_secs(10))
.expect("the send must give up");
assert_eq!(
sent,
Err(StatusCode::TimedOut),
"a send deadline, mid-frame"
);
assert!(session.inner.shared.lifecycle.is_torn_down());
}
#[test]
fn a_request_or_reply_read_while_a_send_waits_ends_the_session() {
use super::super::transport::UnixTransport;
let txn = R34Codec
.encode_transact_ref(WireTransactionRef {
address: &RpcAddress::zero(),
code: 1,
flags: FLAG_ONEWAY,
async_number: 0,
data: &[],
object_positions: &[],
})
.expect("TRANSACT");
let reply = R34Codec
.encode_reply_ref(WireReplyRef {
status: 0,
data: &[],
object_positions: &[],
})
.expect("REPLY");
for (frame, want) in [(txn, StatusCode::BadType), (reply, StatusCode::DeadObject)] {
let (a, peer) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
peer.send_raw(&frame).expect("peer frame");
let sent = oneway_past_the_buffer(&session).expect("the send must stop at the frame");
assert_eq!(sent, Err(want));
assert!(session.inner.shared.lifecycle.is_torn_down());
}
}
#[test]
fn an_r34_drained_header_is_refused_before_its_body() {
for (command, body_size, want) in [
(0, 1000, StatusCode::BadType),
(1, 1000, StatusCode::DeadObject),
(2, 16, StatusCode::BadValue),
(2, A13_DEC_STRONG_LEN as u32, StatusCode::BadValue),
(9, RPC_ADDR_LEN as u32, StatusCode::DeadObject),
] {
let (session, mut peer) = r34_client_with_raw_peer();
std::io::Write::write_all(&mut peer, &a13_header(command, body_size)).expect("header");
let sent = oneway_past_the_buffer(&session)
.unwrap_or_else(|| panic!("command {command}: the send waited for the body"));
assert_eq!(sent, Err(want), "command {command}, body {body_size}");
assert!(session.inner.shared.lifecycle.is_torn_down());
}
}
fn r34_client_with_raw_peer() -> (RpcSession, std::os::unix::net::UnixStream) {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
let (client_fd, peer_fd) = unix_socketpair_fd();
let session = RpcSession::new(
Box::new(UnixTransport::from_stream(UnixStream::from(client_fd)).expect("transport")),
AddressSpace::Initiator,
)
.expect("session");
(session, UnixStream::from(peer_fd))
}
fn oneway_past_the_buffer(session: &RpcSession) -> Option<Result<()>> {
let mut data = Parcel::new();
data.write(&vec![0u8; 8 << 20]).expect("payload");
let (tx, rx) = mpsc::channel();
let inner = Arc::clone(&session.inner);
std::thread::spawn(move || {
let r = inner.client_transact(RpcAddress::zero(), 1, &data, FLAG_ONEWAY);
let _ = tx.send(r.map(|_| ()));
});
let sent = rx.recv_timeout(Duration::from_secs(10)).ok();
if sent.is_none() {
session.close_session();
}
sent
}
fn a13_client_with_raw_peer() -> (RpcSession, std::os::unix::net::UnixStream) {
use super::super::transport::UnixTransport;
use std::os::unix::net::UnixStream;
let (client_fd, peer_fd) = unix_socketpair_fd();
let session = RpcSession::with_profile(
Box::new(UnixTransport::from_stream(UnixStream::from(client_fd)).expect("transport")),
AddressSpace::Initiator,
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2")),
)
.expect("session");
(session, UnixStream::from(peer_fd))
}
fn a13_header(command: u32, body_size: u32) -> [u8; 16] {
let mut h = [0u8; 16];
h[..4].copy_from_slice(&command.to_le_bytes());
h[4..8].copy_from_slice(&body_size.to_le_bytes());
h
}
#[test]
fn a_drained_header_is_refused_before_its_body() {
for (command, body_size, want) in [
(0, 1000, StatusCode::BadType),
(1, 1000, StatusCode::DeadObject),
(2, 1000, StatusCode::BadValue),
(9, 1000, StatusCode::DeadObject),
] {
let (session, mut peer) = a13_client_with_raw_peer();
std::io::Write::write_all(&mut peer, &a13_header(command, body_size)).expect("header");
let sent = oneway_past_the_buffer(&session)
.unwrap_or_else(|| panic!("command {command}: the send waited for the body"));
assert_eq!(sent, Err(want), "command {command}");
assert!(session.inner.shared.lifecycle.is_torn_down());
}
}
#[test]
fn a_peer_that_stalls_inside_a_drained_message_fails_the_send_at_its_deadline() {
let (session, mut peer) = a13_client_with_raw_peer();
session.set_timeout(Some(Duration::from_millis(300)));
let len = crate::rpc::wire_android13::A13_DEC_STRONG_LEN as u32;
let half = &a13_header(2, len)[..8];
std::io::Write::write_all(&mut peer, half).expect("half a header");
let sent = oneway_past_the_buffer(&session).expect("the send must end at its deadline");
assert_eq!(sent, Err(StatusCode::TimedOut));
assert!(session.inner.shared.lifecycle.is_torn_down());
}
#[test]
fn a_failed_send_writes_no_drained_release_after_its_cut_frame() {
let (session, mut peer) = a13_client_with_raw_peer();
let codec = Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2");
let mut counter = 0;
let pay = codec
.encode_dec_strong(&RpcAddress::unique(&mut counter, AddressSpace::Acceptor), 1)
.expect("a frame")
.0;
let Ok(WireMessage::DecStrong(addr, _)) = codec.decode_message(&pay) else {
panic!("a DEC_STRONG");
};
{
let mut st = session.inner.shared.state.lock().expect("state");
st.on_proxy_leaving(addr);
assert_eq!(st.release_proxy(&addr, std::ptr::null()), 0);
}
peer.set_read_timeout(Some(Duration::from_secs(5)))
.expect("peer timeout");
std::io::Write::write_all(&mut peer, &pay).expect("payment");
std::io::Write::write_all(&mut peer, &a13_header(0, 1000)).expect("TRANSACT header");
let sent = oneway_past_the_buffer(&session).expect("the send must end at the TRANSACT");
assert_eq!(sent, Err(StatusCode::BadType));
assert!(session.inner.shared.lifecycle.is_torn_down());
let mut written = Vec::new();
std::io::Read::read_to_end(&mut peer, &mut written).expect("to the session's end");
let (release, _) = codec.encode_dec_strong(&addr, 1).expect("a frame");
assert!(
!written.windows(release.len()).any(|w| w == release),
"the release followed the cut frame"
);
}
#[test]
fn a_peer_that_closes_while_a_send_waits_is_a_dead_object() {
use super::super::transport::UnixTransport;
for timeout in [None, Some(Duration::from_secs(5))] {
let (a, peer) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
session.set_timeout(timeout);
let closer = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
drop(peer);
});
let sent = oneway_past_the_buffer(&session).expect("the send must end at the close");
closer.join().expect("closer");
assert_eq!(sent, Err(StatusCode::DeadObject), "timeout {timeout:?}");
assert!(session.inner.shared.lifecycle.is_torn_down());
}
}
#[test]
fn a_deadline_guard_restores_the_deadline_it_replaced() {
#[derive(Default)]
struct Rec {
read: Mutex<Option<Duration>>,
closed: AtomicBool,
}
impl RpcTransport for Rec {
fn send_frame(&self, _: &[u8]) -> RpcResult<()> {
Ok(())
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
Err(RpcError::EndOfStream)
}
fn peer_identity(&self) -> PeerIdentity {
PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"rec"
}
fn shutdown(&self) -> RpcResult<()> {
Ok(())
}
fn set_read_timeout(&self, t: Option<Duration>) -> RpcResult<()> {
if self.closed.load(Ordering::SeqCst) {
return Err(RpcError::Io(std::io::Error::from_raw_os_error(22)));
}
*self.read.lock().unwrap() = t;
Ok(())
}
fn peer_closed(&self) -> Option<bool> {
Some(self.closed.load(Ordering::SeqCst))
}
}
let t = Rec::default();
let now = || *t.read.lock().unwrap();
let ms = Duration::from_millis;
let baseline = Some(ms(900));
{
let _reply = ReplyDeadlineGuard::arm(&t, Some(ms(100)), baseline).expect("outer");
{
let _drain = ReplyDeadlineGuard::arm(&t, Some(ms(200)), baseline).expect("inner");
assert_eq!(now(), Some(ms(200)));
}
assert_eq!(
now(),
Some(ms(100)),
"the outer reply wait keeps its deadline"
);
{
let _nested = NestedDeadlineGuard::lift(&t, Some(ms(100))).expect("lift");
assert_eq!(now(), None);
drop(ReplyDeadlineGuard::arm(&t, Some(ms(200)), baseline).expect("inner"));
assert_eq!(now(), None, "the nested dispatch keeps its lifted deadline");
}
assert_eq!(now(), Some(ms(100)));
}
assert_eq!(now(), baseline);
t.closed.store(true, Ordering::SeqCst);
drop(ReplyDeadlineGuard::arm(&t, Some(ms(100)), baseline).expect("closed peer: skipped"));
drop(NestedDeadlineGuard::lift(&t, Some(ms(100))).expect("closed peer: skipped"));
t.closed.store(false, Ordering::SeqCst);
assert_eq!(now(), baseline);
ARMED_READ.with(|a| assert!(a.borrow().is_empty(), "every record is dropped"));
}
#[test]
fn a_peer_that_reads_slowly_is_not_cut_by_the_send_deadline() {
use super::super::transport::UnixTransport;
let (a, peer) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
session.set_timeout(Some(Duration::from_millis(300)));
let payload = 2usize << 20;
let mut data = Parcel::new();
data.write(&vec![0u8; payload]).expect("payload");
let reader = std::thread::spawn(move || {
let mut buf = vec![0u8; 64 << 10];
let mut total = 0usize;
while total < payload {
match peer.recv_raw(&mut buf) {
Ok(0) | Err(_) => break,
Ok(n) => total += n,
}
std::thread::sleep(Duration::from_millis(50));
}
total
});
let sent = session
.inner
.client_transact(RpcAddress::zero(), 1, &data, FLAG_ONEWAY);
assert!(matches!(sent, Ok(None)), "got {sent:?}");
assert!(reader.join().expect("reader") >= payload);
assert!(!session.inner.shared.lifecycle.is_torn_down());
}
#[test]
fn a_loop_with_nothing_to_serve_ends_with_session_ended() {
use super::super::transport::UnixTransport;
use crate::rpc::{EndedBy, StreamState};
let (a, _pa) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
let lifecycle = &session.inner.shared.lifecycle;
let end = session.serve_blocking_on(u64::MAX);
assert_eq!(end.reason, EndReason::SessionEnded);
assert!(
!lifecycle.is_torn_down(),
"a wrong id leaves the session up"
);
session.inner.fail_session();
let end = session.spawn_serve().expect("spawn").join().expect("serve");
assert_eq!(end.reason, EndReason::SessionEnded);
assert_eq!(end.by, EndedBy::NotLocal, "a fault ended it, not this end");
assert_eq!(end.stream, StreamState::Lost);
}
#[test]
fn a_session_ended_under_a_parked_serve_loop_stops_it() {
use super::super::transport::UnixTransport;
use crate::rpc::{EndedBy, StreamState};
let (a, _pa) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
let (b, _pb) = UnixTransport::pair().expect("socketpair");
session
.inner
.add_outgoing_slot(Box::new(b))
.expect("fan-out slot");
let founding = RpcSession::FOUNDING_SLOT_ID;
let mut st = session.inner.conn_state.lock().expect("conn_state");
let slot = st
.slots
.iter_mut()
.find(|s| s.id == founding)
.expect("founding");
slot.exclusive_tid = Some(current_tid());
drop(st);
let (parked_tx, parked) = mpsc::channel();
*session.inner.shared.park_hook.lock().expect("park hook") = Some(parked_tx);
let serve = session.spawn_serve().expect("spawn");
parked
.recv_timeout(Duration::from_secs(10))
.expect("the serve loop must park in `slot_cv`");
session.inner.fail_session();
let (done_tx, done) = mpsc::channel();
std::thread::spawn(move || {
let _ = done_tx.send(serve.join());
});
let end = done
.recv_timeout(Duration::from_secs(10))
.expect("the parked loop must wake")
.expect("serve");
assert_eq!(end.reason, EndReason::SessionEnded);
assert_eq!(end.by, EndedBy::NotLocal);
assert_eq!(end.stream, StreamState::Lost);
assert!(session.inner.shared.lifecycle.is_torn_down());
}
#[test]
fn a_callback_on_a_serve_slot_waits_under_the_idle_value_by_default() {
use super::super::transport::UnixTransport;
let (t, _silent) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::from_android13plus(
Box::new(t),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("session");
let idle = Duration::from_millis(200);
session.set_serve_read_deadline(Some(idle));
let (slot_id, slot) = {
let mut st = session.inner.conn_state.lock().expect("conn_state");
st.slots[0].allow_nested = true;
(st.slots[0].id, Arc::clone(&st.slots[0].transport))
};
slot.set_read_timeout(None).expect("clear");
let inner = Arc::clone(&session.inner);
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let sess_ptr = &*inner as *const RpcSessionInner as usize;
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
let t0 = Instant::now();
let r = inner.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
);
let _ = tx.send((r.map(|_| ()), t0.elapsed()));
});
let (r, waited) = rx
.recv_timeout(Duration::from_secs(10))
.expect("the idle value must bound the reply wait");
assert_eq!(r, Err(StatusCode::TimedOut));
assert!(waited >= idle, "gave up after {waited:?}");
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"a reply timeout ends the session"
);
}
#[test]
fn a_oneway_handlers_callback_waits_under_the_idle_value_by_default() {
use super::super::transport::UnixTransport;
let idle = Duration::from_millis(200);
let (session, slots, peers) = idle_serve_slots(idle, 0);
let (cb, silent) = UnixTransport::pair().expect("socketpair");
let cb_slot = session.inner.add_slot_inner_capped(Box::new(cb), 8);
cb_slot.expect("callback slot");
let inner = Arc::clone(&session.inner);
let serve_slot = slots[0];
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let sess_ptr = &*inner as *const RpcSessionInner as usize;
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, serve_slot)));
let t0 = Instant::now();
let r = inner.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
);
let _ = tx.send((r.map(|_| ()), t0.elapsed()));
});
let (r, waited) = rx
.recv_timeout(Duration::from_secs(10))
.expect("the idle value must bound the reply wait");
assert_eq!(r, Err(StatusCode::TimedOut));
assert!(waited >= idle, "gave up after {waited:?}");
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"a reply timeout ends the session"
);
let mut buf = [0u8; 16];
assert!(silent.recv_raw(&mut buf).expect("callback peer") > 0);
assert_eq!(peers[0].recv_raw(&mut buf).expect("serve peer"), 0);
}
#[test]
fn a_session_without_an_idle_value_decides_reply_deadlines_without_the_pool_lock() {
use super::super::transport::UnixTransport;
let (a, _pa) = UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
let inner = Arc::clone(&session.inner);
let held = session.inner.conn_state.lock().expect("conn_state");
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let sess_ptr = &*inner as *const RpcSessionInner as usize;
let founding = RpcSession::FOUNDING_SLOT_ID;
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, founding)));
let got = (
inner.handler_read_deadline(),
inner.slot_baseline_read_deadline(founding),
);
DRIVING.with(|d| d.borrow_mut().pop());
let _ = tx.send(got);
});
let got = rx.recv_timeout(Duration::from_secs(5));
drop(held);
assert_eq!(got, Ok((None, None)), "answered without the pool lock");
}
#[test]
fn a_frame_being_sent_is_activity() {
use super::super::transport::UnixTransport;
struct Parked {
t: UnixTransport,
sends: AtomicUsize,
parked: mpsc::Sender<()>,
release: Mutex<mpsc::Receiver<()>>,
}
impl RpcTransport for Parked {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.t.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.t.recv_frame()
}
fn peer_identity(&self) -> PeerIdentity {
self.t.peer_identity()
}
fn describe(&self) -> &str {
"parked"
}
fn shutdown(&self) -> RpcResult<()> {
self.t.shutdown()
}
fn supports_fd_passing(&self) -> bool {
self.t.supports_fd_passing()
}
fn send_raw(&self, buf: &[u8]) -> RpcResult<()> {
if self.sends.fetch_add(1, Ordering::SeqCst) == 0 {
let _ = self.parked.send(());
let _ = self.release.lock().unwrap().recv();
}
self.t.send_raw(buf)
}
}
let idle = Duration::from_millis(300);
let (session, slots, _peers) = idle_serve_slots(idle, 0);
let (t, _peer) = UnixTransport::pair().expect("socketpair");
let (parked_tx, parked) = mpsc::channel();
let (release, release_rx) = mpsc::channel();
let transport = Parked {
t,
sends: AtomicUsize::new(0),
parked: parked_tx,
release: Mutex::new(release_rx),
};
let cb_slot = session.inner.add_slot_inner_capped(Box::new(transport), 8);
cb_slot.expect("callback slot");
let data = Parcel::new();
let inner = Arc::clone(&session.inner);
let sender = std::thread::spawn(move || {
inner
.client_transact(RpcAddress::zero(), 1, &data, FLAG_ONEWAY)
.map(|_| ())
});
parked
.recv_timeout(Duration::from_secs(10))
.expect("the write must start");
let (waiting_tx, waiting) = mpsc::channel();
*session.inner.shared.serve_wait_hook.lock().expect("hook") = Some(waiting_tx);
let ends = serve_all(&session, &slots);
waiting
.recv_timeout(Duration::from_secs(10))
.expect("the serve loop must start its wait");
std::thread::sleep(3 * idle);
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"evicted while a frame was being written"
);
assert_eq!(
session.inner.shared.open.load(Ordering::SeqCst),
1,
"the write is open"
);
release.send(()).expect("release");
assert_eq!(sender.join().expect("sender"), Ok(()));
assert!(evicted(&ends, slots.len()), "quiet at last, it is idle");
}
#[test]
fn a_frame_being_received_is_activity() {
use super::super::transport::UnixTransport;
struct Parked {
t: UnixTransport,
reads: AtomicUsize,
parked: mpsc::Sender<()>,
release: Mutex<mpsc::Receiver<()>>,
}
impl Parked {
fn len_for_this_read(&self, want: usize) -> usize {
match self.reads.fetch_add(1, Ordering::SeqCst) {
0 => want.min(1),
1 => {
let _ = self.parked.send(());
let _ = self.release.lock().unwrap().recv();
want
}
_ => want,
}
}
}
impl RpcTransport for Parked {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.t.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.t.recv_frame()
}
fn peer_identity(&self) -> PeerIdentity {
self.t.peer_identity()
}
fn describe(&self) -> &str {
"parked"
}
fn shutdown(&self) -> RpcResult<()> {
self.t.shutdown()
}
fn recv_raw(&self, buf: &mut [u8]) -> RpcResult<usize> {
let len = self.len_for_this_read(buf.len());
self.t.recv_raw(&mut buf[..len])
}
fn recv_raw_with_fds(&self, buf: &mut [u8]) -> RpcResult<(usize, Vec<OwnedFd>)> {
let len = self.len_for_this_read(buf.len());
self.t.recv_raw_with_fds(&mut buf[..len])
}
}
for fd_mode in [FD_MODE_NONE, FD_MODE_UNIX] {
let (t, peer) = UnixTransport::pair().expect("socketpair");
let (parked_tx, parked) = mpsc::channel();
let (release, release_rx) = mpsc::channel();
let transport = Parked {
t,
reads: AtomicUsize::new(0),
parked: parked_tx,
release: Mutex::new(release_rx),
};
let session = RpcSession::from_android13plus(
Box::new(transport),
Android13PlusCodec::android14_15(),
fd_mode,
true,
)
.expect("session");
let unix = session.inner.fd_mode() == FileDescriptorTransportMode::Unix;
assert_eq!(unix, fd_mode == FD_MODE_UNIX, "the reader under test");
let slot = {
let st = session.inner.conn_state.lock().expect("conn_state");
Arc::clone(&st.slots[0].transport)
};
let mut frame = [0u8; 20];
frame[4] = 4;
peer.send_raw(&frame).expect("send the frame");
let mut seen = session.inner.activity();
let inner = Arc::clone(&session.inner);
let reader = std::thread::spawn(move || inner.recv_msg(&*slot).map(|(f, _)| f.len()));
parked
.recv_timeout(Duration::from_secs(10))
.expect("the second read must start");
assert_eq!(session.inner.shared.open.load(Ordering::SeqCst), 0);
assert!(
session.inner.active_since(&mut seen),
"a frame part-way in, fd mode {fd_mode}"
);
release.send(()).expect("release");
assert_eq!(reader.join().expect("reader").ok(), Some(20));
}
}
fn idle_serve_slots(
idle: Duration,
extra: usize,
) -> (
RpcSession,
Vec<u64>,
Vec<super::super::transport::UnixTransport>,
) {
use super::super::transport::UnixTransport;
let (t, peer) = UnixTransport::pair().expect("socketpair");
t.set_read_timeout(Some(idle)).expect("arm");
let session = RpcSession::from_android13plus(
Box::new(t),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("session");
session.set_serve_read_deadline(Some(idle));
let mut slots = vec![RpcSession::FOUNDING_SLOT_ID];
let mut peers = vec![peer];
for _ in 0..extra {
let (t, peer) = UnixTransport::pair().expect("socketpair");
t.set_read_timeout(Some(idle)).expect("arm");
let slot = session.add_incoming_slot_capped(Box::new(t), 8);
slots.push(slot.expect("attach"));
peers.push(peer);
}
(session, slots, peers)
}
fn serve_all(session: &RpcSession, slots: &[u64]) -> mpsc::Receiver<SessionEnd> {
let (tx, rx) = mpsc::channel();
for &slot in slots {
let (session, tx) = (session.clone(), tx.clone());
std::thread::spawn(move || {
let _ = tx.send(session.serve_blocking_on(slot));
});
}
rx
}
fn evicted(ends: &mpsc::Receiver<SessionEnd>, loops: usize) -> bool {
use crate::rpc::EndedBy;
let ends: Vec<SessionEnd> = (0..loops)
.map(|_| {
ends.recv_timeout(Duration::from_secs(10))
.expect("every loop ends")
})
.collect();
ends.iter().any(|end| {
end.reason == EndReason::Frame(StatusCode::TimedOut) && end.by == EndedBy::Local
})
}
#[test]
fn an_idle_session_ends_between_d_and_2d_after_its_last_activity() {
use super::super::transport::UnixTransport;
let idle = Duration::from_secs(1);
for case in [
"call end",
"callback frame",
"serve attach",
"callback attach",
] {
let (session, slots, _peers) = idle_serve_slots(idle, 0);
let (cb, _cb_peer) = UnixTransport::pair().expect("socketpair");
let cb_slot = session.inner.add_slot_inner_capped(Box::new(cb), 8);
cb_slot.expect("callback slot");
let call = (case == "call end").then(|| OpenCall::enter(&session.inner.shared));
let (joining, _joined_peer) = UnixTransport::pair().expect("socketpair");
let (waiting_tx, waiting) = mpsc::channel();
*session.inner.shared.serve_wait_hook.lock().expect("hook") = Some(waiting_tx);
let ends = serve_all(&session, &slots);
waiting
.recv_timeout(Duration::from_secs(10))
.expect("the serve loop must start its wait");
std::thread::sleep(idle / 2);
let last = Instant::now();
match case {
"call end" => drop(call),
"callback frame" => {
let sent = session.inner.client_transact(
RpcAddress::zero(),
1,
&Parcel::new(),
FLAG_ONEWAY,
);
assert!(matches!(sent, Ok(None)), "{case}: {sent:?}");
}
"serve attach" => {
let joined = session.add_incoming_slot_capped(Box::new(joining), 8);
joined.expect("serve attach");
}
_ => {
let codec = Android13PlusCodec::android14_15();
let joined = session.add_callback_slot_and_init(Box::new(joining), 8, &codec);
joined.expect("callback attach");
}
}
assert!(evicted(&ends, slots.len()), "{case}: an idle eviction");
let quiet = last.elapsed();
assert!(
quiet >= idle && quiet < 2 * idle,
"{case}: ended {quiet:?} after the last activity"
);
}
}
#[test]
fn a_frame_trickled_across_the_idle_period_is_not_idle() {
let idle = Duration::from_millis(400);
let (session, slots, peers) = idle_serve_slots(idle, 1);
let ends = serve_all(&session, &slots);
let (frame, _) = Android13PlusCodec::android14_15()
.encode_dec_strong(&RpcAddress::zero(), 1)
.expect("a frame");
for pair in frame.chunks(2) {
peers[1].send_raw(pair).expect("trickle");
std::thread::sleep(idle / 4);
}
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"evicted while the frame was arriving"
);
assert!(evicted(&ends, slots.len()), "quiet at last, it is idle");
}
#[test]
fn frames_on_another_slot_keep_a_quiet_one_up() {
let idle = Duration::from_millis(400);
let (session, slots, peers) = idle_serve_slots(idle, 1);
let ends = serve_all(&session, &slots);
let codec = Android13PlusCodec::android14_15();
for _ in 0..12 {
let (frame, _) = codec
.encode_dec_strong(&RpcAddress::zero(), 1)
.expect("a frame");
peers[1].send_raw(&frame).expect("a frame");
std::thread::sleep(idle / 4);
}
assert!(
!session.inner.shared.lifecycle.is_torn_down(),
"evicted while frames were arriving"
);
assert!(evicted(&ends, slots.len()), "quiet at last, it is idle");
}
#[test]
fn an_idle_expiry_on_a_slot_the_session_dropped_is_not_an_eviction() {
use super::super::transport::MemTransport;
use crate::rpc::{EndedBy, StreamState};
struct Faulted(MemTransport, Arc<Mutex<Weak<RpcSessionInner>>>);
impl RpcTransport for Faulted {
fn send_frame(&self, buf: &[u8]) -> RpcResult<()> {
self.0.send_frame(buf)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
self.0.recv_frame()
}
fn recv_raw(&self, _: &mut [u8]) -> RpcResult<usize> {
let session = self.1.lock().unwrap().upgrade();
if let Some(inner) = session {
inner.fail_session();
}
Err(RpcError::Timeout)
}
fn peer_identity(&self) -> PeerIdentity {
self.0.peer_identity()
}
fn describe(&self) -> &str {
"faulted"
}
fn shutdown(&self) -> RpcResult<()> {
self.0.shutdown()
}
}
let (t, _peer) = MemTransport::pair();
let cell = Arc::new(Mutex::new(Weak::new()));
let transport = Faulted(t, Arc::clone(&cell));
let session =
RpcSession::new(Box::new(transport), AddressSpace::Acceptor).expect("session");
session.set_serve_read_deadline(Some(Duration::from_secs(60)));
*cell.lock().unwrap() = Arc::downgrade(&session.inner);
let end = session.serve_blocking_on(RpcSession::FOUNDING_SLOT_ID);
assert_eq!(end.reason, EndReason::SessionEnded);
assert_eq!(
end.by,
EndedBy::NotLocal,
"another connection's fault ended it"
);
assert_eq!(end.stream, StreamState::Lost);
}
#[test]
fn the_kernels_etimedout_under_an_armed_deadline_is_not_an_eviction() {
use crate::rpc::{EndedBy, StreamState};
struct Etimedout;
impl std::io::Read for Etimedout {
fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
let etimedout = rustix::io::Errno::TIMEDOUT.raw_os_error();
Err(std::io::Error::from_raw_os_error(etimedout))
}
}
struct Gone;
impl RpcTransport for Gone {
fn send_frame(&self, _: &[u8]) -> RpcResult<()> {
Err(RpcError::EndOfStream)
}
fn recv_frame(&self) -> RpcResult<Vec<u8>> {
crate::rpc::transport::read_frame(&mut Etimedout)
}
fn recv_raw(&self, buf: &mut [u8]) -> RpcResult<usize> {
std::io::Read::read(&mut Etimedout, buf).map_err(RpcError::from)
}
fn peer_identity(&self) -> PeerIdentity {
PeerIdentity::Anonymous
}
fn describe(&self) -> &str {
"etimedout"
}
fn shutdown(&self) -> RpcResult<()> {
Ok(())
}
}
let session = RpcSession::new(Box::new(Gone), AddressSpace::Acceptor).expect("session");
let end = session.serve_blocking_clearing_deadline_after_first();
assert_eq!(end.reason, EndReason::Frame(StatusCode::DeadObject));
assert_eq!(end.by, EndedBy::NotLocal, "the peer's host went away");
assert_eq!(end.stream, StreamState::Lost);
}
#[test]
fn slot_of_a_different_transport_kind_is_refused() {
use crate::rpc::transport::{MemTransport, UnixTransport};
let (t0, _p0) = MemTransport::pair();
let codec = Android13PlusCodec::android14_15();
let session = RpcSession::from_android13plus(Box::new(t0), codec, FD_MODE_NONE, false)
.expect("build session");
let (u, _pu) = UnixTransport::pair().expect("socketpair");
assert_eq!(
session.inner.add_outgoing_slot(Box::new(u)),
Err(StatusCode::BadType)
);
let (u, _pu) = UnixTransport::pair().expect("socketpair");
assert_eq!(
session.inner.add_slot_inner_capped(Box::new(u), 2),
Err(StatusCode::BadType)
);
let (u, _pu) = UnixTransport::pair().expect("socketpair");
assert_eq!(
session.inner.add_incoming_slot_capped(Box::new(u), 2),
Err(StatusCode::BadType)
);
assert_eq!(session.inner.slot_count(), 1, "only the founding slot");
let (m, _pm) = MemTransport::pair();
assert!(
session.inner.add_outgoing_slot(Box::new(m)).is_ok(),
"a slot of the founding kind still joins"
);
}
#[test]
fn proxy_of_another_session_is_refused() {
use crate::rpc::proxy::RpcProxy;
use crate::rpc::transport::MemTransport;
let make = || {
let (t, p) = MemTransport::pair();
let s = RpcSession::from_android13plus(
Box::new(t),
Android13PlusCodec::android14_15(),
FD_MODE_NONE,
false,
)
.expect("build session");
(s, p)
};
let (a, _pa) = make();
let (b, _pb) = make();
let mut counter = 0u64;
let addr = RpcAddress::unique(&mut counter, AddressSpace::Acceptor);
let proxy_of_a = SIBinder::new(Arc::new(RpcProxy::new(addr, a.inner.clone())))
.expect("SIBinder::new(RpcProxy)");
let mut parcel = Parcel::new();
assert_eq!(
b.inner.write_binder(Some(&proxy_of_a), &mut parcel),
Err(StatusCode::InvalidOperation),
"another session's proxy must not be addressed on this wire"
);
let mut parcel = Parcel::new();
a.inner
.write_binder(Some(&proxy_of_a), &mut parcel)
.expect("the owning session writes its own proxy back");
}
struct LocalSvc;
impl crate::Interface for LocalSvc {}
impl crate::Remotable for LocalSvc {
fn descriptor() -> &'static str {
"rsbinder.test.LocalSvc"
}
fn on_transact(
&self,
_: crate::TransactionCode,
_: &mut Parcel,
_: &mut Parcel,
) -> Result<()> {
Ok(())
}
fn on_dump(&self, _w: &mut dyn std::io::Write, _a: &[String]) -> Result<()> {
Ok(())
}
}
#[test]
fn append_from_after_the_session_is_gone_is_dead_object() {
use crate::rpc::transport::MemTransport;
let (t, _peer) = MemTransport::pair();
let session = RpcSession::new(Box::new(t), AddressSpace::Acceptor).expect("session");
let ops = session.inner.parcel_ops();
let inner = Arc::downgrade(&session.inner);
drop(session);
assert!(inner.upgrade().is_none(), "the session's inner is freed");
let mut source = Parcel::new();
source.attach_rpc_ops(ops.clone());
source.write(&7i32).expect("scalar");
let mut dest = Parcel::new();
dest.attach_rpc_ops(ops);
let size = source.data_size();
assert_eq!(
dest.append_from(&mut source, 0, size),
Err(StatusCode::DeadObject)
);
}
#[test]
fn would_block_leaves_the_parcel_owning_its_bump() {
use crate::rpc::transport::MemTransport;
let (t, _peer) = MemTransport::pair();
let session = RpcSession::new(Box::new(t), AddressSpace::Acceptor).expect("session");
let local = crate::Interface::as_binder(&crate::Binder::new(LocalSvc));
let mut d = Parcel::new();
d.configure_rpc(
session.inner.parcel_ops(),
session.inner.fd_mode(),
session.inner.records_fd_positions(),
);
d.write(&local).expect("write a local binder");
assert_eq!(session.local_node_count(), 1, "the write took its bump");
assert_eq!(
session
.inner
.client_transact(RpcAddress::zero(), crate::FIRST_CALL_TRANSACTION, &d, 0)
.err(),
Some(StatusCode::WouldBlock)
);
assert_eq!(
session.local_node_count(),
1,
"a request never sent keeps its bump for the retry with the same parcel"
);
drop(d);
assert_eq!(
session.local_node_count(),
0,
"the unsent parcel's drop gives the bump back"
);
}
fn parcel_session(profile: WireProfile) -> RpcSession {
let (t, _peer) = crate::rpc::transport::MemTransport::pair();
RpcSession::with_profile(Box::new(t), AddressSpace::Acceptor, profile).expect("session")
}
fn a13_v2() -> WireProfile {
WireProfile::Android13Plus(Android13PlusCodec::with_version(PROTOCOL_V2).expect("v2"))
}
#[test]
fn a_null_binder_carries_its_stability_on_every_wire() {
for (profile, words) in [
(a13_v2(), [0i32, 0]),
(WireProfile::R34(R34Codec), [0, 0x0000_0001]),
] {
let session = parcel_session(profile);
let mut p = Parcel::new();
p.attach_rpc_ops(session.inner.parcel_ops());
p.write(&None::<SIBinder>).expect("write null");
assert_eq!(p.data_size(), words.len() * 4);
p.set_data_position(0);
for w in words {
assert_eq!(p.read::<i32>().expect("word"), w);
}
p.set_data_position(0);
assert!(p.read::<Option<SIBinder>>().expect("read null").is_none());
assert_eq!(
p.data_position(),
p.data_size(),
"the stability is consumed"
);
}
}
#[test]
fn a_null_binder_stability_is_checked_as_aosp_does() {
for (profile, stability, want) in [
(a13_v2(), 0x0c, Err(StatusCode::BadType)),
(
WireProfile::R34(R34Codec),
0x0c00_0001,
Err(StatusCode::BadType),
),
(WireProfile::R34(R34Codec), 0, Err(StatusCode::BadType)),
(WireProfile::R34(R34Codec), 0x0000_0002, Ok(())),
] {
let session = parcel_session(profile);
let mut p = Parcel::new();
p.attach_rpc_ops(session.inner.parcel_ops());
p.write(&0i32).expect("present");
p.write(&stability).expect("stability");
p.set_data_position(0);
assert_eq!(
p.read::<Option<SIBinder>>().map(|b| assert!(b.is_none())),
want,
"stability {stability:#x}"
);
}
}
#[test]
fn an_r34_binder_is_followed_by_its_android12_category() {
let session = parcel_session(WireProfile::R34(R34Codec));
let local = crate::Interface::as_binder(&crate::Binder::new(LocalSvc));
let mut p = Parcel::new();
p.attach_rpc_ops(session.inner.parcel_ops());
p.write(&local).expect("write");
assert_eq!(p.data_size(), 4 + RPC_ADDR_LEN + 4);
p.set_data_position(0);
assert_eq!(p.read::<i32>().expect("present"), 1);
p.set_data_position(4 + RPC_ADDR_LEN);
assert_eq!(p.read::<i32>().expect("stability"), 0x0c00_0001);
}
#[test]
fn concurrent_senders_of_one_parcel_are_serialized() {
let d = Parcel::new_data_only();
assert_eq!(d.rpc_begin_send(), Ok(()));
assert_eq!(
d.rpc_begin_send(),
Err(StatusCode::InvalidOperation),
"a second sender is refused while the first is in flight"
);
d.rpc_end_send(false);
assert_eq!(
d.rpc_begin_send(),
Ok(()),
"a failed attempt returns to `NotSent`"
);
d.rpc_end_send(true);
assert_eq!(
d.rpc_begin_send(),
Err(StatusCode::InvalidOperation),
"a sent parcel is never sent again"
);
}
}
#[cfg(test)]
#[path = "ref_accounting_tests.rs"]
mod ref_accounting_tests;