use std::cell::RefCell;
use std::os::fd::{AsFd, OwnedFd};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::{Duration, Instant};
use super::end::{EndReason, ServeStep, SessionEnd};
use super::fd_mode::FileDescriptorTransportMode;
use super::lifecycle::SessionLifecycle;
use crate::binder::{SIBinder, FLAG_ONEWAY, INTERFACE_TRANSACTION, PING_TRANSACTION};
use crate::error::{Result, StatusCode};
use crate::parcel::{Parcel, RpcParcelOps};
use super::address::{AddressSpace, RpcAddress, SpecialTransaction, RPC_ADDR_LEN};
use super::proxy::RpcProxy;
use super::state::RpcState;
use super::transport::{PeerIdentity, RpcTransport};
use super::wire::{R34Codec, WireCodec, WireMessage, WireReply, WireTransaction};
use super::wire_android13::{
client_connect_with_id, read_aosp_message, read_aosp_message_with_fds,
server_accept_deferred_init, write_aosp_message, write_aosp_message_with_fds,
Android13PlusCodec, RawTransportIo, A13_ADDR_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);
enum RpcUnixAddr<'a> {
Path(&'a Path),
#[cfg(any(target_os = "linux", target_os = "android"))]
Abstract(&'a [u8]),
}
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>,
}
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 connect(&self) -> Result<super::transport::UnixTransport> {
match &self.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)
}
}
enum WireProfile {
R34(R34Codec),
Android13Plus(Android13PlusCodec),
}
impl WireProfile {
fn codec(&self) -> &dyn WireCodec {
match self {
WireProfile::R34(c) => c,
WireProfile::Android13Plus(c) => c,
}
}
fn aosp_framing(&self) -> bool {
matches!(self, WireProfile::Android13Plus(_))
}
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<()> {
let got: String = reader.read()?;
if got != expected {
log::error!("RPC interface token mismatch: expected '{expected}', got '{got}'");
return Err(StatusCode::BadType);
}
Ok(())
}
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,
}
}
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>,
restore: Option<Duration>,
) -> RpcResult<Self> {
let armed = deadline.is_some();
if let Some(d) = deadline {
transport.set_read_timeout(Some(d))?;
}
Ok(Self {
transport,
armed,
restore,
})
}
}
impl Drop for ReplyDeadlineGuard<'_> {
fn drop(&mut self) {
if self.armed {
let _ = self.transport.set_read_timeout(self.restore);
}
}
}
struct HandshakeDeadline<'a> {
transport: &'a dyn RpcTransport,
armed: bool,
}
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 = deadline.is_some();
if armed {
transport.set_read_timeout(deadline)?;
transport.set_write_timeout(deadline)?;
}
Ok(Self { transport, armed })
}
}
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) {
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}) — the session id is unknown or \
stale, the peer's outgoing-slot cap (`set_max_threads`) is spent, or it is shutting \
down. The id must be the server-minted one from `RpcSession::get_session_id()` (NOT \
`session_id()`, which is a client-local value), and the connection count must stay \
within `RpcSession::negotiate()`"
);
}
fn client_handshake_err(e: RpcError, requesting_new_session: bool) -> StatusCode {
if requesting_new_session {
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 without `?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 may be speaking the r34 (default) profile, or a read \
deadline armed on this connection landed mid-frame. Connect without \
`?profile=android13plus`, or enable the android-13+ wire on the server \
(`RpcServer::set_android13plus`)"
),
RpcError::Timeout => log::error!(
"rsbinder RPC: the android-13+ handshake stalled and a read deadline armed on \
this connection elapsed — that deadline is the caller's own \
(`RpcUnixClientConfig::handshake_timeout`, the 10s \
`RpcSession::from_preconnected_fd` arms, or one set on the transport \
directly), so it may simply be shorter than this peer's legitimate response \
time. 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> {
if deadline.is_some() {
transport.set_read_timeout(None)?;
}
Ok(Self {
transport,
restore: deadline,
})
}
}
impl Drop for NestedDeadlineGuard<'_> {
fn drop(&mut self) {
if let Some(d) = self.restore {
let _ = self.transport.set_read_timeout(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,
ClientRefcount,
Reply,
}
impl ConnUse {
fn is_new_transaction(self) -> bool {
matches!(self, ConnUse::Client)
}
}
type Tid = std::thread::ThreadId;
#[inline]
fn current_tid() -> Tid {
std::thread::current().id()
}
struct ConnSlot {
transport: Arc<dyn RpcTransport>,
exclusive_tid: Option<Tid>,
id: u64,
role: SlotRole,
allow_nested: bool,
stale_replies: u32,
unreadable: bool,
}
struct ConnState {
slots: Vec<ConnSlot>,
next_slot_id: u64,
}
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
}
}
enum StreamPosition {
FrameBoundaryIntact,
StreamPositionUnknown,
}
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 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;
}
}
}
pub(crate) struct SharedSession {
state: Mutex<RpcState>,
root: Mutex<Option<SIBinder>>,
max_threads: AtomicU32,
negotiated: AtomicU32,
timeout: Mutex<Option<Duration>>,
serve_read_deadline: Mutex<Option<Duration>>,
fd_mode: Mutex<crate::rpc::FileDescriptorTransportMode>,
fd_unix_supported: AtomicBool,
rpc_session_id: RpcSessionId,
lifecycle: SessionLifecycle,
ended_locally: AtomicBool,
space: AddressSpace,
}
impl SharedSession {
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
}
}
pub(crate) struct RpcSessionInner {
conn_state: Mutex<ConnState>,
slot_cv: Condvar,
profile: WireProfile,
self_weak: Mutex<Weak<RpcSessionInner>>,
dec_strong_tx: mpsc::Sender<RpcAddress>,
shared: Arc<SharedSession>,
incoming_threads: Mutex<Vec<(u64, std::thread::JoinHandle<()>)>>,
incoming_live: AtomicUsize,
incoming_joined: AtomicUsize,
}
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)
}
}
impl RpcSessionInner {
fn find_conn(&self) -> Result<ConnGuard<'_>> {
self.find_conn_impl(ConnUse::Client)
}
fn find_conn_lenient(&self) -> Result<ConnGuard<'_>> {
self.find_conn_impl(ConnUse::ClientRefcount)
}
fn find_conn_impl(&self, use_: ConnUse) -> Result<ConnGuard<'_>> {
let refcount = use_ == ConnUse::ClientRefcount;
let mut wait_until: Option<Instant> = None;
let tid = current_tid();
let sess_ptr = self as *const RpcSessionInner as usize;
let pinned = DRIVING.with(|d| {
d.borrow()
.iter()
.rev()
.find_map(|&(sp, sid)| if sp == sess_ptr { Some(sid) } else { None })
});
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.unreadable => return Err(StatusCode::DeadObject),
Some(s)
if s.role == SlotRole::Outgoing
|| s.allow_nested
|| !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 && !s.unreadable);
if let Some(idx) = pick {
let s = &mut st.slots[idx];
s.exclusive_tid = Some(tid);
let slot_id = s.id;
let transport = Arc::clone(&s.transport);
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
return Ok(ConnGuard {
inner: self,
slot_id,
transport,
reentrant: false,
});
}
if !st
.slots
.iter()
.any(|s| s.role == SlotRole::Outgoing && !s.unreadable)
{
if refcount {
log::debug!(
"RPC: no outgoing connection for a DEC_STRONG; skipping (the node is \
released at session end, or promptly once the peer opens an incoming \
connection)"
);
} else if self.shared.space() == AddressSpace::Initiator {
log::error!(
"RPC: this client has lost every outgoing connection (a reply \
deadline or a desynced reply retires the slot it rode); the \
surviving callback connections are server→client only, so \
nothing would read a request written on them"
);
} else 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 call (from inside a twoway handler) \
can transact here; use `?profile=android13plus` for callbacks \
outside a handler"
);
} else {
log::error!(
"RPC: session has no outgoing connection — a non-nested call (from \
another thread, or a oneway outside a handler) needs the peer to open \
incoming connections (RpcUnixClientConfig::incoming_connections / \
ARpcSession_setMaxIncomingThreads); refusing instead of waiting forever"
);
}
return Err(StatusCode::WouldBlock);
}
let deadline = *self.shared.timeout.lock().expect("timeout poisoned");
st = match deadline {
Some(d) => {
let at = *wait_until.get_or_insert_with(|| Instant::now() + d);
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
}
None => self.slot_cv.wait(st).expect("slot_cv poisoned"),
};
}
}
fn try_find_conn(&self) -> Option<ConnGuard<'_>> {
let tid = current_tid();
let sess_ptr = self as *const RpcSessionInner as usize;
if let Some(slot_id) = DRIVING.with(|d| {
d.borrow()
.iter()
.rev()
.find_map(|&(sp, sid)| if sp == sess_ptr { Some(sid) } else { None })
}) {
let st = self.conn_state.lock().expect("conn_state poisoned");
let slot = st.slots.iter().find(|s| s.id == slot_id)?;
if slot.unreadable {
return None;
}
let transport = Arc::clone(&slot.transport);
drop(st);
return Some(ConnGuard {
inner: self,
slot_id,
transport,
reentrant: true,
});
}
let mut st = self.conn_state.lock().expect("conn_state poisoned");
let idx = st.slots.iter().position(|s| {
s.exclusive_tid.is_none() && s.role == SlotRole::Outgoing && !s.unreadable
})?;
let s = &mut st.slots[idx];
s.exclusive_tid = Some(tid);
let slot_id = s.id;
let transport = Arc::clone(&s.transport);
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
Some(ConnGuard {
inner: self,
slot_id,
transport,
reentrant: false,
})
}
fn find_conn_for_reaper(&self) -> Option<ConnGuard<'_>> {
let tid = current_tid();
let sess_ptr = self as *const RpcSessionInner as usize;
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 && !s.unreadable);
if let Some(idx) = pick {
let s = &mut st.slots[idx];
s.exclusive_tid = Some(tid);
let slot_id = s.id;
let transport = Arc::clone(&s.transport);
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, slot_id)));
return Some(ConnGuard {
inner: self,
slot_id,
transport,
reentrant: false,
});
}
if !st
.slots
.iter()
.any(|s| s.role == SlotRole::Outgoing && !s.unreadable)
{
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) {
slot.exclusive_tid = Some(tid);
let transport = Arc::clone(&slot.transport);
DRIVING.with(|d| d.borrow_mut().push((sess_ptr, want_slot_id)));
return Ok(ConnGuard {
inner: self,
slot_id: want_slot_id,
transport,
reentrant: false,
});
}
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 clear_handshake_timeouts(&self) {
let transports: Vec<Arc<dyn RpcTransport>> = {
let st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.iter().map(|s| Arc::clone(&s.transport)).collect()
};
for t in transports {
let _ = t.set_read_timeout(None);
let _ = t.set_write_timeout(None);
}
}
pub(crate) fn parcel_ops(&self) -> Arc<dyn RpcParcelOps> {
Arc::new(SessionParcelOps(
self.self_weak.lock().expect("self_weak").clone(),
))
}
fn add_slot_inner(&self, transport: Box<dyn RpcTransport>, role: SlotRole) -> Option<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 None;
}
let id = st.next_slot_id;
st.next_slot_id += 1;
st.slots.push(ConnSlot {
transport,
exclusive_tid: None,
id,
role,
allow_nested: false,
stale_replies: 0,
unreadable: false,
});
drop(st);
self.slot_cv.notify_all();
Some(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 !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;
st.slots.push(ConnSlot {
transport,
exclusive_tid: None,
id,
role: SlotRole::Incoming,
allow_nested: false,
stale_replies: 0,
unreadable: false,
});
drop(st);
self.slot_cv.notify_all();
Ok(id)
}
fn add_outgoing_slot(&self, transport: Box<dyn RpcTransport>) -> Option<u64> {
self.add_slot_inner(transport, SlotRole::Outgoing)
}
fn add_slot_inner_capped(
&self,
transport: Box<dyn RpcTransport>,
cap: usize,
claimed: bool,
) -> Option<u64> {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if self.shared.lifecycle.is_torn_down() {
return None;
}
if st
.slots
.iter()
.filter(|s| s.role == SlotRole::Outgoing)
.count()
>= cap
{
return None;
}
let transport: Arc<dyn RpcTransport> = Arc::from(transport);
let id = st.next_slot_id;
st.next_slot_id += 1;
st.slots.push(ConnSlot {
transport,
exclusive_tid: if claimed { Some(current_tid()) } else { None },
id,
role: SlotRole::Outgoing,
allow_nested: false,
stale_replies: 0,
unreadable: false,
});
drop(st);
self.slot_cv.notify_all();
Some(id)
}
fn note_stale_reply(&self, slot_id: u64) {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == slot_id) {
s.stale_replies += 1;
}
}
fn mark_slot_unreadable(&self, slot_id: u64) {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == slot_id) {
s.unreadable = true;
}
}
fn slot_unreadable(&self, slot_id: u64) -> bool {
let st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.iter().any(|s| s.id == slot_id && s.unreadable)
}
fn take_stale_reply(&self, slot_id: u64) -> bool {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
match st.slots.iter_mut().find(|s| s.id == slot_id) {
Some(s) if s.stale_replies > 0 => {
s.stale_replies -= 1;
true
}
_ => false,
}
}
fn remove_slot(&self, slot_id: u64) {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.retain(|s| s.id != slot_id);
let empty = st.slots.is_empty();
let no_outgoing = !empty
&& self.shared.space() == AddressSpace::Initiator
&& !st.slots.iter().any(|s| s.role == SlotRole::Outgoing);
drop(st);
self.slot_cv.notify_all();
if (empty || no_outgoing) && self.shared.lifecycle.try_drop_sole_connection() {
self.on_session_dead();
}
}
fn retire_after_failed_send(&self, conn: &ConnGuard<'_>, e: &RpcError) {
if !matches!(e, RpcError::EndOfStream | RpcError::Io(_)) {
return;
}
if conn.reentrant {
self.mark_slot_unreadable(conn.slot_id);
log::error!(
"RPC: a frame failed part-way through its send on a slot another frame owns; \
what the peer reads next is unknowable, so this connection is marked \
unreadable and shut down rather than left for that frame to write on"
);
if let Err(e) = conn.transport().shutdown() {
log::warn!("RPC: shutting the desynced connection down failed: {e}");
}
} else {
self.remove_slot(conn.slot_id);
}
}
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) {
self.shutdown_all_transports();
{
let mut st = self.conn_state.lock().expect("conn_state poisoned");
st.slots.clear();
}
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 = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.clear_local();
drop(locals);
drop(root);
}
pub(crate) fn fd_mode(&self) -> FileDescriptorTransportMode {
*self.shared.fd_mode.lock().expect("fd_mode poisoned")
}
pub(crate) fn records_fd_positions(&self) -> bool {
self.profile.records_fd_positions()
}
fn send_msg(
&self,
transport: &dyn RpcTransport,
frame: &[u8],
fds: &[OwnedFd],
) -> RpcResult<()> {
if self.profile.aosp_framing() {
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);
}
debug_assert!(
fds.is_empty(),
"non-Unix android-13+ session must not carry fds"
);
let _ = fds; return write_aosp_message_with_fds(transport, frame, &[]);
}
if self.fd_mode() == FileDescriptorTransportMode::Unix {
let borrowed: Vec<_> = fds.iter().map(|f| f.as_fd()).collect();
transport.send_frame_with_fds(frame, &borrowed)
} else {
transport.send_frame(frame)
}
}
fn recv_msg(&self, transport: &dyn RpcTransport) -> RpcResult<(Vec<u8>, Vec<OwnedFd>)> {
if self.profile.aosp_framing() {
if self.fd_mode() == FileDescriptorTransportMode::Unix {
return read_aosp_message_with_fds(transport);
}
let mut io = RawTransportIo(transport);
let frame = read_aosp_message(&mut io)?;
return Ok((frame, Vec::new()));
}
if self.fd_mode() == FileDescriptorTransportMode::Unix {
transport.recv_frame_with_fds()
} else {
Ok((transport.recv_frame()?, Vec::new()))
}
}
fn self_weak(&self) -> Weak<RpcSessionInner> {
self.self_weak.lock().expect("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()
}
fn is_client_incoming_slot(&self, slot_id: u64) -> bool {
self.shared.space() == AddressSpace::Initiator
&& self
.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.any(|s| s.id == slot_id && s.role == SlotRole::Incoming)
}
fn slot_baseline_read_deadline(&self, slot_id: u64) -> Option<Duration> {
let serve_driven = self
.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.any(|s| s.id == slot_id && s.role == SlotRole::Incoming);
if !serve_driven {
return None;
}
*self
.shared
.serve_read_deadline
.lock()
.expect("serve_read_deadline poisoned")
}
fn incoming_slot_count(&self) -> usize {
self.conn_state
.lock()
.expect("conn_state poisoned")
.slots
.iter()
.filter(|s| s.role == SlotRole::Incoming)
.count()
}
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 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),
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);
}
parcel.rpc_pin_binder(b.clone());
rp.address()
} else if (**b).is_remote() {
log::error!("RPC: cannot send a kernel binder proxy over sockets");
return Err(StatusCode::InvalidOperation);
} else {
let addr = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.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);
}
if matches!(self.profile, WireProfile::Android13Plus(_)) {
let rep: i32 = b.stability().into();
parcel.write(&rep)?;
}
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 {
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)?;
if matches!(self.profile, WireProfile::Android13Plus(_)) {
let _stability: i32 = parcel.read()?;
}
if let Some(local) = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.lookup_local(&addr)
{
return Ok(Some(local));
}
let strong = self.self_weak().upgrade().ok_or(StatusCode::DeadObject)?;
let (sib, excess) = {
let mut st = self.shared.state.lock().expect("rpc state poisoned");
st.remote_proxy(addr, || {
SIBinder::new(Arc::new(RpcProxy::new(addr, strong)))
.expect("SIBinder::new(RpcProxy)")
})?
};
if excess {
let _ = self.send_dec_strong(addr);
}
Ok(Some(sib))
}
fn rollback_outgoing(&self, data: &Parcel, oneway_addr: Option<(RpcAddress, u64)>) {
if oneway_addr.is_none() && data.rpc_leaving_addrs().is_empty() {
return;
}
let released: Vec<SIBinder> = {
let mut state = self.shared.state.lock().expect("rpc state poisoned");
if let Some((addr, consumed)) = oneway_addr {
state.cancel_send_async_number(addr, consumed);
}
data.rpc_leaving_addrs()
.iter()
.filter_map(|addr| state.cancel_binder_leaving(addr))
.collect()
};
drop(released);
}
pub(crate) fn client_transact(
&self,
addr: RpcAddress,
code: u32,
data: &Parcel,
flags: u32,
) -> Result<Option<Parcel>> {
let oneway = (flags & FLAG_ONEWAY) != 0;
let conn = self.find_conn()?;
let transport = conn.transport();
let async_number = if oneway {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.next_send_async_number(addr)
} else {
0
};
let txn = WireTransaction {
address: addr,
code,
flags,
async_number,
data: data.rpc_data_bytes().to_vec(),
object_positions: data.rpc_object_positions().to_vec(),
};
let rollback = || self.rollback_outgoing(data, oneway.then_some((addr, async_number)));
let frame = match self.profile.codec().encode_transact(&txn) {
Ok(frame) => frame,
Err(e) => {
rollback();
return Err(e.into());
}
};
if let Err(e) = self.send_msg(transport, &frame, data.rpc_out_fds()) {
rollback();
self.retire_after_failed_send(&conn, &e);
return Err(e.into());
}
if oneway {
return Ok(None);
}
let deadline = *self.shared.timeout.lock().expect("timeout poisoned");
let record_stream_position = |desync: StreamPosition| {
if !conn.reentrant {
self.remove_slot(conn.slot_id);
return;
}
match desync {
StreamPosition::FrameBoundaryIntact => self.note_stale_reply(conn.slot_id),
StreamPosition::StreamPositionUnknown => {
self.mark_slot_unreadable(conn.slot_id);
log::error!(
"RPC: a nested call lost track of the stream; what the peer sends next \
is unknowable, so this connection is marked unreadable and shut down \
rather than left for the outer call to read"
);
if let Err(e) = transport.shutdown() {
log::warn!("RPC: shutting the desynced connection down failed: {e}");
}
}
}
};
let restore = self.slot_baseline_read_deadline(conn.slot_id);
let _deadline_guard = match ReplyDeadlineGuard::arm(transport, deadline, restore) {
Ok(g) => g,
Err(e) => {
record_stream_position(StreamPosition::FrameBoundaryIntact);
return Err(e.into());
}
};
loop {
if self.slot_unreadable(conn.slot_id) {
if !conn.reentrant {
self.remove_slot(conn.slot_id);
}
return Err(StatusCode::DeadObject);
}
let (frame, in_fds) = match self.recv_msg(transport) {
Ok(v) => v,
Err(e) => {
let desync = if e.leaves_frame_boundary_intact() {
StreamPosition::FrameBoundaryIntact
} else {
StreamPosition::StreamPositionUnknown
};
record_stream_position(desync);
return Err(e.into());
}
};
let message = match self.profile.codec().decode_message(&frame) {
Ok(m) => m,
Err(e) => {
record_stream_position(StreamPosition::StreamPositionUnknown);
return Err(e.into());
}
};
match message {
WireMessage::Reply(WireReply {
status,
data,
object_positions,
}) => {
if self.take_stale_reply(conn.slot_id) {
log::debug!("RPC: skipped the late reply of an abandoned nested call");
continue;
}
if status != 0 {
return Err(StatusCode::from(status));
}
let mut reply = Parcel::from_vec(data);
reply.configure_rpc(
self.parcel_ops(),
self.fd_mode(),
self.records_fd_positions(),
);
reply.rpc_set_in_fds(in_fds);
reply.rpc_set_object_positions(object_positions);
reply.set_data_position(0);
return Ok(Some(reply));
}
WireMessage::DecStrong(a, amount) => {
let released = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.dec_strong_local(&a, amount);
drop(released);
}
WireMessage::Transact(t) => {
let _restore = match NestedDeadlineGuard::lift(transport, deadline) {
Ok(g) => g,
Err(e) => {
record_stream_position(StreamPosition::FrameBoundaryIntact);
return Err(e.into());
}
};
let peer = transport.peer_identity();
if let Err(e) = self.dispatch_transact(t, in_fds, peer) {
record_stream_position(StreamPosition::FrameBoundaryIntact);
return Err(e);
}
}
}
}
}
pub(crate) fn queue_dec_strong(&self, addr: RpcAddress) {
if self.shared.lifecycle.is_torn_down() {
return;
}
if let Some(conn) = self.try_find_conn() {
let frame = self.profile.codec().encode_dec_strong(&addr);
if let Err(e) = self.send_msg(conn.transport(), &frame, &[]) {
self.retire_after_failed_send(&conn, &e);
}
return;
}
let _ = self.dec_strong_tx.send(addr);
}
pub(crate) fn send_dec_strong(&self, addr: RpcAddress) -> Result<()> {
if self.shared.lifecycle.is_torn_down() {
return Ok(());
}
let conn = self.find_conn_lenient()?;
let frame = self.profile.codec().encode_dec_strong(&addr);
if let Err(e) = self.send_msg(conn.transport(), &frame, &[]) {
self.retire_after_failed_send(&conn, &e);
return Err(e.into());
}
Ok(())
}
pub(crate) fn forget_remote_if(&self, addr: &RpcAddress, who: *const ()) {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.forget_remote_if(addr, who);
}
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<()> {
let frame = self.profile.codec().encode_reply(&WireReply {
status,
data: data.to_vec(),
object_positions: object_positions.to_vec(),
})?;
let conn = self.find_conn_impl(ConnUse::Reply)?;
if let Err(e) = self.send_msg(conn.transport(), &frame, fds) {
self.retire_after_failed_send(&conn, &e);
return Err(e.into());
}
Ok(())
}
fn dispatch_transact(
&self,
t: WireTransaction,
in_fds: Vec<OwnedFd>,
peer: PeerIdentity,
) -> Result<()> {
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 mut next = {
let mut state = self.shared.state.lock().expect("rpc state poisoned");
match state.dispatch_async_or_enqueue(addr, wire_async, t, in_fds) {
super::state::AsyncDecision::Dispatch(t, fds) => 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
);
None
}
super::state::AsyncDecision::Terminate(num_pending) => {
log::error!(
"RPC: {num_pending} pending oneway transactions on {addr:?}; \
flushing backlog and tearing down connection"
);
return Err(StatusCode::FailedTransaction);
}
}
};
while let Some((t, fds)) = next {
self.execute_dispatched(t, fds, true, Arc::clone(&peer))?;
next = {
let mut state = self.shared.state.lock().expect("rpc state poisoned");
state.advance_and_pop_async(addr)
};
}
Ok(())
}
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 {
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(
0,
reply.rpc_data_bytes(),
reply.rpc_object_positions(),
&[],
);
}
PING_TRANSACTION => {
return self.send_reply(0, &[], &[], &[]);
}
_ => {}
}
}
let mut reader = Parcel::from_vec(t.data);
reader.configure_rpc(
self.parcel_ops(),
self.fd_mode(),
self.records_fd_positions(),
);
reader.rpc_set_in_fds(in_fds);
reader.rpc_set_object_positions(t.object_positions);
reader.set_data_position(0);
let mut reply = Parcel::new();
reply.configure_rpc(
self.parcel_ops(),
self.fd_mode(),
self.records_fd_positions(),
);
let _nested = AllowNestedGuard::arm(self, !oneway);
let result = consume_rpc_interface_token(&mut reader, target.descriptor()).and_then(|()| {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _calling = crate::thread_state::RpcCallingGuard::install(Arc::clone(&peer));
target.rpc_transact(t.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 {}: {msg}", t.code);
Err(crate::StatusCode::Unknown)
})
});
if oneway {
if let Err(e) = result {
log::error!("oneway RPC transaction failed (dropped): {e:?}");
}
self.rollback_outgoing(&reply, None);
return Ok(());
}
match result {
Ok(()) => {
let sent = self.send_reply(
0,
reply.rpc_data_bytes(),
reply.rpc_object_positions(),
reply.rpc_out_fds(),
);
if sent.is_err() {
self.rollback_outgoing(&reply, None);
}
sent
}
Err(e) => {
self.rollback_outgoing(&reply, None);
self.send_reply(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::Retired);
};
if self.slot_unreadable(slot_id) {
return ServeStep::Ended(EndReason::Unreadable);
}
let transport = conn.transport();
let (frame, in_fds) = match self.recv_msg(transport) {
Ok(f) => f,
Err(RpcError::EndOfStream) => return ServeStep::Ended(EndReason::EndOfStream),
Err(RpcError::UncleanEndOfStream) => {
return ServeStep::Ended(EndReason::UncleanEndOfStream)
}
Err(RpcError::DeadlineMidFrame) => {
return ServeStep::Ended(EndReason::DeadlineMidFrame)
}
Err(e) => return ServeStep::Ended(EndReason::Frame(e.into())),
};
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) => {
let released = self
.shared
.state
.lock()
.expect("rpc state poisoned")
.dec_strong_local(&a, amount);
drop(released);
ServeStep::Continue
}
WireMessage::Reply(_) => {
if self.take_stale_reply(slot_id) {
log::debug!("RPC: skipped the late reply of an abandoned nested call");
return ServeStep::Continue;
}
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());
let write_res = match &root {
Some(b) => reply.write(b),
None => reply.write(&0i32),
};
if let Err(e) = write_res {
self.rollback_outgoing(&reply, None);
return Err(e);
}
let sent =
self.send_reply(0, reply.rpc_data_bytes(), reply.rpc_object_positions(), &[]);
if sent.is_err() {
self.rollback_outgoing(&reply, None);
}
sent
}
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();
reply.write(&self.shared.rpc_session_id.as_bytes()[..])?;
self.send_reply(0, reply.rpc_data_bytes(), &[], &[])
}
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 fd_version_ok = match self.profile.wire_version() {
None => true,
Some(v) => v >= PROTOCOL_V1,
};
let agreed = if want_unix
&& fd_version_ok
&& 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.fd_mode.lock().expect("fd_mode poisoned") = agreed;
Ok(())
}
None => self.send_reply(StatusCode::UnknownTransaction.into(), &[], &[], &[]),
}
}
}
#[derive(Clone)]
pub struct RpcSession {
inner: Arc<RpcSessionInner>,
}
impl RpcSession {
pub fn new(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: Mutex::new(None),
fd_mode: Mutex::new(FileDescriptorTransportMode::None),
fd_unix_supported: AtomicBool::new(false),
rpc_session_id: gen_rpc_session_id()?,
lifecycle: SessionLifecycle::new(),
ended_locally: AtomicBool::new(false),
space,
}))
}
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 founding = ConnSlot {
transport: Arc::from(transport),
exclusive_tid: None,
id: 1,
role: founding_role,
allow_nested: false,
stale_replies: 0,
unreadable: false,
};
let (dec_strong_tx, dec_strong_rx) = mpsc::channel();
let inner = Arc::new(RpcSessionInner {
conn_state: Mutex::new(ConnState {
slots: vec![founding],
next_slot_id: 2,
}),
slot_cv: Condvar::new(),
profile,
self_weak: Mutex::new(Weak::new()),
shared,
dec_strong_tx,
incoming_threads: Mutex::new(Vec::new()),
incoming_live: AtomicUsize::new(0),
incoming_joined: AtomicUsize::new(0),
});
*inner.self_weak.lock().expect("self_weak") = Arc::downgrade(&inner);
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 slot_id = self
.inner
.add_slot_inner_capped(transport, cap, true)
.ok_or(StatusCode::FailedTransaction)?;
let transport = {
let st = self.inner.conn_state.lock().expect("conn_state poisoned");
st.slots
.iter()
.find(|s| s.id == slot_id)
.map(|s| Arc::clone(&s.transport))
.ok_or(StatusCode::DeadObject)?
};
let sent = {
let mut io = RawTransportIo(&*transport);
super::wire_android13::server_write_connection_init(&mut io, codec)
};
{
let mut st = self.inner.conn_state.lock().expect("conn_state poisoned");
if let Some(s) = st.slots.iter_mut().find(|s| s.id == slot_id) {
s.exclusive_tid = None;
}
}
self.inner.slot_cv.notify_all();
match sent {
Ok(()) => Ok(slot_id),
Err(e) => {
if let Err(e) = transport.shutdown() {
log::warn!("RPC: shutting a failed callback connection down failed: {e}");
}
self.inner.remove_slot(slot_id);
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, false)
.ok_or(StatusCode::FailedTransaction)
}
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> {
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
.fd_mode
.lock()
.expect("fd_mode poisoned") = 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_with_id(transport, max_version, fd_mode, &[])
}
pub fn connect_android13plus_fd_with_id(
transport: Box<dyn RpcTransport>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
session_id: &[u8],
) -> Result<RpcSession> {
Self::connect_android13plus_fd_with_id_hs(transport, max_version, fd_mode, session_id, None)
}
pub(crate) fn connect_android13plus_fd_with_id_hs(
transport: Box<dyn RpcTransport>,
max_version: u32,
fd_mode: FileDescriptorTransportMode,
session_id: &[u8],
handshake_timeout: Option<Duration>,
) -> Result<RpcSession> {
if !(session_id.is_empty() || session_id.len() == 32) {
return Err(StatusCode::BadValue);
}
let want_unix = fd_mode == FileDescriptorTransportMode::Unix;
let hdr_fd_mode = if want_unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let codec = {
let _hs = HandshakeDeadline::arm(transport.as_ref(), handshake_timeout)
.map_err(StatusCode::from)?;
let mut io = RawTransportIo(transport.as_ref());
client_connect_with_id(&mut io, max_version, false, hdr_fd_mode, session_id)
.map_err(|e| client_handshake_err(e, session_id.is_empty()))?
};
if !session_id.is_empty() {
let _hs = HandshakeDeadline::arm(transport.as_ref(), handshake_timeout)
.map_err(StatusCode::from)?;
if let Err(e) = confirm_attach(transport.as_ref(), &codec, session_id) {
log_attach_refused(&e);
return Err(StatusCode::from(e));
}
}
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
.fd_mode
.lock()
.expect("fd_mode poisoned") = 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)?;
reply.read::<Vec<u8>>()
}
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> {
let want_unix = want == FileDescriptorTransportMode::Unix;
let mut req = Parcel::new();
req.write(&(if want_unix { 1i32 } else { 0i32 }))?;
let mut reply = self
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetFdMode.code(),
&req,
0,
)?
.ok_or(StatusCode::UnexpectedNull)?;
let agreed = if reply.read::<i32>()? == 1 {
FileDescriptorTransportMode::Unix
} else {
FileDescriptorTransportMode::None
};
*self.inner.shared.fd_mode.lock().expect("fd_mode poisoned") = agreed;
Ok(agreed)
}
pub fn fd_transport_mode(&self) -> FileDescriptorTransportMode {
self.inner.fd_mode()
}
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 serve_blocking_on(&self, slot_id: u64) -> SessionEnd {
self.serve_blocking_on_inner(slot_id, false, false)
}
pub fn serve_blocking_clearing_deadline_after_first(&self) -> SessionEnd {
self.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, true, true)
}
pub(crate) fn serve_blocking_clearing_admission_deadline(&self, armed: bool) -> SessionEnd {
self.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, true, armed)
}
fn serve_blocking_on_inner(
&self,
slot_id: u64,
clear_deadline_after_first: bool,
admission_deadline_armed: bool,
) -> SessionEnd {
let client_incoming = self.inner.is_client_incoming_slot(slot_id);
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;
loop {
match self.inner.serve_once_on_slot(slot_id) {
ServeStep::Continue => {
if first {
self.inner.clear_slot_read_timeout(slot_id);
first = false;
deadline_armed = false;
}
}
ServeStep::Ended(reason) => break (reason, deadline_armed),
}
}
};
let end = SessionEnd::new(
reason,
self.inner.shared.ended_locally.load(Ordering::SeqCst),
deadline_armed,
);
if !client_incoming && self.inner.shared.lifecycle.drop_connection() {
self.inner.on_session_dead();
}
self.inner.remove_slot(slot_id);
if client_incoming
&& self.inner.incoming_slot_count() == 0
&& self.inner.shared.lifecycle.try_drop_sole_connection()
{
self.inner.on_session_dead();
}
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 set_timeout(&self, timeout: Option<Duration>) {
*self.inner.shared.timeout.lock().expect("timeout poisoned") = reject_zero_deadline(
timeout,
"RpcSession::set_timeout: a zero duration is not a valid deadline; ignoring",
);
}
pub(crate) fn set_serve_read_deadline(&self, deadline: Option<Duration>) {
*self
.inner
.shared
.serve_read_deadline
.lock()
.expect("serve_read_deadline poisoned") = deadline;
}
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)
}
pub fn setup_unix_client_android13plus_with_id(
path: impl AsRef<std::path::Path>,
max_version: u32,
session_id: &[u8],
) -> Result<RpcSession> {
Self::setup_unix_client_android13plus_with_config(
RpcUnixClientConfig::path(path.as_ref(), max_version).session_id(session_id),
)
}
pub fn setup_unix_client_android13plus_with_config(
config: RpcUnixClientConfig,
) -> Result<RpcSession> {
reject_zero_handshake_timeout(
config.handshake_timeout,
"RpcUnixClientConfig::handshake_timeout",
)?;
let local = config.outgoing_connections.max(1);
let incoming = config.incoming_connections;
if (local > 1 || incoming > 0) && !config.session_id.is_empty() {
return Err(StatusCode::BadValue);
}
let session = RpcSession::connect_android13plus_fd_with_id_hs(
Box::new(config.connect()?),
config.max_version,
config.fd_mode.unwrap_or(FileDescriptorTransportMode::None),
config.session_id,
config.handshake_timeout,
)?;
if config.timeout.is_some() {
session.set_timeout(config.timeout);
}
if local == 1 && incoming == 0 {
return Ok(session);
}
let 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(
|| config.connect(),
config.max_version,
&session_id,
fd_mode,
config.handshake_timeout,
)?;
}
for _ in 0..incoming {
session.add_incoming_connection_android13plus_transport(
|| config.connect(),
config.max_version,
&session_id,
fd_mode,
config.handshake_timeout,
)?;
}
Ok(())
};
if let Err(e) = build() {
session.close_session();
return Err(e);
}
Ok(session)
}
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_android13plus_with_config(
RpcUnixClientConfig::path(path.as_ref(), max_version).session_id(session_id),
)
}
pub fn add_outgoing_connection_android13plus_with_config(
&self,
config: RpcUnixClientConfig,
) -> Result<u64> {
reject_zero_handshake_timeout(
config.handshake_timeout,
"RpcUnixClientConfig::handshake_timeout",
)?;
if config.outgoing_connections.max(1) != 1
|| config.incoming_connections != 0
|| config.session_id.len() != 32
{
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);
}
self.add_outgoing_connection_android13plus_transport(
|| config.connect(),
config.max_version,
config.session_id,
fd_mode,
config.handshake_timeout,
)
}
fn add_outgoing_connection_android13plus_transport(
&self,
connect: impl FnOnce() -> Result<super::transport::UnixTransport>,
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),
};
let effective_max = max_version.min(session_version);
let hdr_fd_mode = if fd_mode == FileDescriptorTransportMode::Unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let t = connect()?;
let codec = {
let _hs = HandshakeDeadline::arm(&t, handshake_timeout).map_err(StatusCode::from)?;
let mut io = RawTransportIo(&t);
client_connect_with_id(&mut io, effective_max, false, hdr_fd_mode, session_id)
.map_err(StatusCode::from)?
};
if codec.version() != session_version {
log::error!(
"android-13+ RPC: this attach negotiated wire v{} but the session runs v{} — \
a caller-supplied `max_version` below the session's negotiated version can \
never attach; pass `RpcSession::wire_protocol_version()`",
codec.version(),
session_version
);
return Err(StatusCode::BadType);
}
{
let probe_deadline =
handshake_timeout.or(*self.inner.shared.timeout.lock().expect("timeout poisoned"));
let _hs = HandshakeDeadline::arm(&t, probe_deadline).map_err(StatusCode::from)?;
if let Err(e) = confirm_attach(&t, &codec, session_id) {
log_attach_refused(&e);
return Err(StatusCode::from(e));
}
}
self.inner
.add_outgoing_slot(Box::new(t))
.ok_or(StatusCode::DeadObject)
}
pub fn add_incoming_connection_android13plus_with_config(
&self,
config: RpcUnixClientConfig,
) -> Result<u64> {
reject_zero_handshake_timeout(
config.handshake_timeout,
"RpcUnixClientConfig::handshake_timeout",
)?;
if config.outgoing_connections.max(1) != 1
|| config.incoming_connections != 0
|| config.session_id.len() != 32
{
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);
}
self.add_incoming_connection_android13plus_transport(
|| config.connect(),
config.max_version,
config.session_id,
fd_mode,
config.handshake_timeout,
)
}
fn add_incoming_connection_android13plus_transport(
&self,
connect: impl FnOnce() -> Result<super::transport::UnixTransport>,
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);
let hdr_fd_mode = if fd_mode == FileDescriptorTransportMode::Unix {
FD_MODE_UNIX
} else {
FD_MODE_NONE
};
let t = connect()?;
let codec = {
let _hs = HandshakeDeadline::arm(&t, handshake_timeout).map_err(StatusCode::from)?;
let mut io = RawTransportIo(&t);
client_connect_with_id(&mut io, effective_max, true, hdr_fd_mode, session_id)
.map_err(StatusCode::from)?
};
if codec.version() != session_version {
return Err(StatusCode::BadType);
}
let slot_id = self
.inner
.add_slot_inner(Box::new(t), SlotRole::Incoming)
.ok_or(StatusCode::DeadObject)?;
let inner = Arc::clone(&self.inner);
self.inner.incoming_live.fetch_add(1, Ordering::SeqCst);
let spawned = std::thread::Builder::new()
.name(format!("rsbinder-rpc-in-{slot_id}"))
.spawn(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));
Ok(slot_id)
}
Err(e) => {
log::error!("RPC: incoming connection thread spawn failed: {e}");
self.inner.incoming_live.fetch_sub(1, Ordering::SeqCst);
self.inner.remove_slot(slot_id);
Err(StatusCode::from(e))
}
}
}
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_unix_client_android13plus_with_config(
RpcUnixClientConfig::path(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 _ = transport.set_read_timeout(Some(HANDSHAKE_TIMEOUT));
let _ = transport.set_write_timeout(Some(HANDSHAKE_TIMEOUT));
let session = RpcSession::connect_android13plus_fd(
transport,
max_version,
FileDescriptorTransportMode::None,
)?;
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 __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>) {
while let Ok(addr) = 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 frame = inner.profile.codec().encode_dec_strong(&addr);
if let Err(e) = inner.send_msg(conn.transport(), &frame, &[]) {
inner.retire_after_failed_send(&conn, &e);
}
drop(conn);
drop(inner);
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::os::fd::{AsFd, OwnedFd};
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());
}
#[test]
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_unix_client_android13plus_with_config(
RpcUnixClientConfig::path(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_makes_the_slot_unreadable() {
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, "protocol violation");
assert_eq!(
session
.inner
.conn_state
.lock()
.expect("conn_state")
.slots
.len(),
1,
"a reentrant frame must not retire the outer frame's slot"
);
assert!(
session.inner.slot_unreadable(slot_id),
"the borrowed slot must be marked unreadable for its owner"
);
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::Unreadable)
),
"the serve loop must refuse the slot, not read another frame on it"
);
}
#[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 outer frame owns slot {slot_id} and must retire it on the way out"
);
assert!(
session.inner.shared.lifecycle.is_torn_down(),
"the only connection is gone, so the session is dead"
);
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 an_unreadable_slot_is_never_selected_for_writing() {
let (a, _b) = super::super::transport::UnixTransport::pair().expect("socketpair");
let session = RpcSession::new(Box::new(a), AddressSpace::Initiator).expect("session");
let slot_id = session.inner.conn_state.lock().expect("conn_state").slots[0].id;
session.inner.mark_slot_unreadable(slot_id);
assert!(
matches!(session.inner.find_conn(), Err(StatusCode::WouldBlock)),
"the scan must not hand out the only outgoing slot once it is unreadable"
);
assert!(
session.inner.try_find_conn().is_none(),
"the non-blocking selector must decline it too"
);
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;
assert!(
matches!(session.inner.find_conn(), Err(StatusCode::DeadObject)),
"the owning frame must be refused before it writes"
);
assert!(session.inner.try_find_conn().is_none());
}
#[test]
fn an_unreadable_slot_is_refused_before_a_request_is_sent() {
use crate::rpc::wire_android13::write_aosp_message;
use std::io::Read;
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 slot_id = {
let st = session.inner.conn_state.lock().expect("conn_state");
st.slots[0].id
};
let reply = session
.inner
.profile
.codec()
.encode_reply(&WireReply {
status: 0,
data: Vec::new(),
object_positions: Vec::new(),
})
.expect("encode a REPLY frame");
write_aosp_message(&mut peer, &reply).expect("queue the reply frame");
session.inner.mark_slot_unreadable(slot_id);
assert_eq!(
session
.inner
.client_transact(
RpcAddress::zero(),
SpecialTransaction::GetSessionId.code(),
&Parcel::new(),
0,
)
.expect_err("an unreadable slot must not be used again"),
StatusCode::WouldBlock,
"the session's only outgoing slot is unreadable: refused at selection"
);
peer.set_read_timeout(Some(Duration::from_millis(100)))
.expect("read timeout");
let mut buf = [0u8; 16];
assert!(
matches!(peer.read(&mut buf), Err(e) if super::super::transport::is_timeout(&e)),
"no request may be written on an unreadable slot"
);
}
#[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.find_conn_lenient(),
Err(StatusCode::WouldBlock)
),
"a DEC_STRONG must not claim the free serve-driven slot"
);
assert!(
session.inner.try_find_conn().is_none(),
"the non-blocking fast path must not claim it either"
);
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!(
session.inner.find_conn_lenient().is_ok(),
"with an outgoing slot the DEC_STRONG goes out normally"
);
}
#[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();
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 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!(
session
.inner
.add_slot_inner_capped(Box::new(t), 2, false)
.is_none(),
"no slot may be pushed onto a torn-down session"
);
assert_eq!(session.inner.slot_count(), 0, "the dead pool stays empty");
}
#[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");
}
}