use std::cell::RefCell;
use std::os::fd::{AsFd, OwnedFd};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::Duration;
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,
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,
fd_mode: Option<FileDescriptorTransportMode>,
}
impl<'a> RpcUnixClientConfig<'a> {
fn new(addr: RpcUnixAddr<'a>, max_version: u32) -> Self {
Self {
addr,
max_version,
session_id: &[],
outgoing_connections: 1,
fd_mode: 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 fd_mode(mut self, mode: FileDescriptorTransportMode) -> Self {
self.fd_mode = Some(mode);
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) {
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))
}
struct ReplyDeadlineGuard<'a> {
transport: &'a dyn RpcTransport,
armed: bool,
}
impl<'a> ReplyDeadlineGuard<'a> {
fn arm(transport: &'a dyn RpcTransport, deadline: Option<Duration>) -> RpcResult<Self> {
let armed = deadline.is_some();
if let Some(d) = deadline {
transport.set_read_timeout(Some(d))?;
}
Ok(Self { transport, armed })
}
}
impl Drop for ReplyDeadlineGuard<'_> {
fn drop(&mut self) {
if self.armed {
let _ = self.transport.set_read_timeout(None);
}
}
}
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()) };
}
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,
}
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
}
}
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();
}
}
}
pub(crate) struct SharedSession {
state: Mutex<RpcState>,
root: Mutex<Option<SIBinder>>,
max_threads: AtomicU32,
negotiated: AtomicU32,
timeout: Mutex<Option<Duration>>,
fd_mode: Mutex<crate::rpc::FileDescriptorTransportMode>,
fd_unix_supported: AtomicBool,
rpc_session_id: RpcSessionId,
lifecycle: SessionLifecycle,
}
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 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>,
}
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<'_>> {
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 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))
.expect("DRIVING slot present")
};
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);
}
if let Some(s) = st
.slots
.iter_mut()
.find(|s| s.exclusive_tid == Some(tid) || s.exclusive_tid.is_none())
{
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,
});
}
st = 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;
let mut st = self.conn_state.lock().expect("conn_state poisoned");
let s = st
.slots
.iter_mut()
.find(|s| s.exclusive_tid == Some(tid) || s.exclusive_tid.is_none())?;
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;
}
if let Some(s) = st
.slots
.iter_mut()
.find(|s| s.exclusive_tid == Some(tid) || s.exclusive_tid.is_none())
{
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,
});
}
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");
st.slots
.iter()
.find(|s| s.id == want_slot_id)
.map(|s| Arc::clone(&s.transport))
.expect("pinned slot present")
};
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>) -> u64 {
let transport: Arc<dyn RpcTransport> = Arc::from(transport);
let mut st = self.conn_state.lock().expect("conn_state poisoned");
let id = st.next_slot_id;
st.next_slot_id += 1;
st.slots.push(ConnSlot {
transport,
exclusive_tid: None,
id,
});
drop(st);
self.slot_cv.notify_all();
id
}
fn add_outgoing_slot(&self, transport: Box<dyn RpcTransport>) -> u64 {
self.add_slot_inner(transport)
}
fn add_slot_inner_capped(&self, transport: Box<dyn RpcTransport>, cap: usize) -> Option<u64> {
let mut st = self.conn_state.lock().expect("conn_state poisoned");
if st.slots.len() >= 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: None,
id,
});
drop(st);
self.slot_cv.notify_all();
Some(id)
}
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);
drop(st);
self.slot_cv.notify_all();
}
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; let mut io = RawTransportIo(transport);
return write_aosp_message(&mut io, 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()
}
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) {
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>() {
rp.address()
} 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 weak = self.self_weak();
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, weak))).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)>) {
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);
}
for &addr in data.rpc_leaving_addrs() {
state.cancel_binder_leaving(&addr);
}
}
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();
return Err(e.into());
}
if oneway {
return Ok(None);
}
let deadline = *self.shared.timeout.lock().expect("timeout poisoned");
let _deadline_guard = ReplyDeadlineGuard::arm(transport, deadline)?;
loop {
let (frame, in_fds) = self.recv_msg(transport)?;
match self.profile.codec().decode_message(&frame)? {
WireMessage::Reply(WireReply {
status,
data,
object_positions,
}) => {
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) => {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.dec_strong_local(&a);
}
WireMessage::Transact(t) => {
let _restore = NestedDeadlineGuard::lift(transport, deadline)?;
let peer = transport.peer_identity();
self.dispatch_transact(t, in_fds, peer)?;
}
}
}
}
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);
let _ = self.send_msg(conn.transport(), &frame, &[]);
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()?;
let frame = self.profile.codec().encode_dec_strong(&addr);
self.send_msg(conn.transport(), &frame, &[])?;
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()?;
Ok(self.send_msg(conn.transport(), &frame, fds)?)
}
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 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:?}");
}
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.send_reply(e.into(), &[], &[], &[]),
}
}
fn serve_once_on_slot(&self, slot_id: u64) -> Result<bool> {
let conn = self.find_conn_pinned(slot_id)?;
let transport = conn.transport();
let (frame, in_fds) = match self.recv_msg(transport) {
Ok(f) => f,
Err(RpcError::PeerClosed) => return Ok(false),
Err(e) => return Err(e.into()),
};
match self.profile.codec().decode_message(&frame)? {
WireMessage::Transact(t) => {
let peer = transport.peer_identity();
self.dispatch_transact(t, in_fds, peer)?;
Ok(true)
}
WireMessage::DecStrong(a) => {
self.shared
.state
.lock()
.expect("rpc state poisoned")
.dec_strong_local(&a);
Ok(true)
}
WireMessage::Reply(_) => {
log::warn!("RPC server received an unexpected REPLY; ignoring");
Ok(true)
}
}
}
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());
match &root {
Some(b) => reply.write(b)?,
None => reply.write(&0i32)?,
}
self.send_reply(0, reply.rpc_data_bytes(), reply.rpc_object_positions(), &[])
}
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) => {
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.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),
fd_mode: Mutex::new(FileDescriptorTransportMode::None),
fd_unix_supported: AtomicBool::new(false),
rpc_session_id: gen_rpc_session_id()?,
lifecycle: SessionLifecycle::new(),
}))
}
fn with_shared(
transport: Box<dyn RpcTransport>,
profile: WireProfile,
shared: Arc<SharedSession>,
) -> RpcSession {
let founding = ConnSlot {
transport: Arc::from(transport),
exclusive_tid: None,
id: 1,
};
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,
});
*inner.self_weak.lock().expect("self_weak") = Arc::downgrade(&inner);
let weak_for_reaper = Arc::downgrade(&inner);
let _ = std::thread::Builder::new()
.name("rsbinder-rpc-reaper".into())
.spawn(move || reaper_loop(weak_for_reaper, dec_strong_rx));
RpcSession { inner }
}
pub(crate) const FOUNDING_SLOT_ID: u64 = 1;
pub(crate) fn add_incoming_slot(&self, transport: Box<dyn RpcTransport>) -> Result<u64> {
if !self.inner.shared.try_bump_live_conns() {
return Err(StatusCode::DeadObject);
}
Ok(self.inner.add_slot_inner(transport))
}
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)
.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,
) -> Result<Android13PlusAccept> {
let (codec, client_fd_mode, client_id, incoming) = {
let mut io = RawTransportIo(transport.as_ref());
server_accept(&mut io, server_max_version).map_err(StatusCode::from)?
};
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,
shared: Option<Arc<SharedSession>>,
) -> RpcResult<RpcSession> {
let negotiated = codec.version();
let shared = match shared {
Some(s) => s,
None => 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> {
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 mut io = RawTransportIo(transport.as_ref());
client_connect_with_id(&mut io, max_version, false, hdr_fd_mode, session_id)
.map_err(StatusCode::from)?
};
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)?;
if incoming {
return Err(StatusCode::BadType);
}
Self::from_android13plus(transport, codec, client_fd_mode, server_fd_unix, None)
.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) {
*self.inner.shared.root.lock().expect("root poisoned") = Some(binder);
}
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>()
.map_err(|_| StatusCode::UnexpectedNull)
}
pub fn serve_blocking(&self) -> Result<()> {
self.serve_blocking_on(Self::FOUNDING_SLOT_ID)
}
pub fn serve_blocking_on(&self, slot_id: u64) -> Result<()> {
self.serve_blocking_on_inner(slot_id, false)
}
pub fn serve_blocking_clearing_deadline_after_first(&self) -> Result<()> {
self.serve_blocking_on_inner(Self::FOUNDING_SLOT_ID, true)
}
fn serve_blocking_on_inner(
&self,
slot_id: u64,
clear_deadline_after_first: bool,
) -> Result<()> {
let result = {
let mut r = Ok(());
let mut first = clear_deadline_after_first;
loop {
match self.inner.serve_once_on_slot(slot_id) {
Ok(cont) => {
if first {
self.inner.clear_slot_read_timeout(slot_id);
first = false;
}
if cont {
continue;
}
break;
}
Err(e) => {
r = Err(e);
break;
}
}
}
r
};
if self.inner.shared.lifecycle.drop_connection() {
self.inner.send_session_obituaries();
self.inner.shared.lifecycle.mark_dead();
}
self.inner.remove_slot(slot_id);
result
}
pub(crate) fn set_max_threads(&self, n: u32) {
self.inner
.shared
.max_threads
.store(n.max(1), Ordering::SeqCst);
}
pub fn set_timeout(&self, timeout: Option<Duration>) {
*self.inner.shared.timeout.lock().expect("timeout poisoned") = timeout;
}
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> {
let local = config.outgoing_connections.max(1);
if local > 1 && !config.session_id.is_empty() {
return Err(StatusCode::BadValue);
}
let session = RpcSession::connect_android13plus_fd_with_id(
Box::new(config.connect()?),
config.max_version,
config.fd_mode.unwrap_or(FileDescriptorTransportMode::None),
config.session_id,
)?;
if local == 1 {
return Ok(session);
}
let negotiated = session.negotiate(local)?;
if negotiated <= 1 {
return Ok(session);
}
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,
)?;
}
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> {
if config.outgoing_connections.max(1) != 1 || 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,
)
}
fn add_outgoing_connection_android13plus_transport(
&self,
connect: impl FnOnce() -> Result<super::transport::UnixTransport>,
max_version: u32,
session_id: &[u8],
fd_mode: FileDescriptorTransportMode,
) -> 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 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 {
return Err(StatusCode::BadType);
}
Ok(self.inner.add_outgoing_slot(Box::new(t)))
}
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)
}
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);
let _ = inner.send_msg(conn.transport(), &frame, &[]);
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 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);
});
std::thread::sleep(std::time::Duration::from_millis(50));
let observed = rustix::fs::fcntl_getfl(observer.as_fd()).expect("getfl observer");
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 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,
None,
)
.expect("build session");
let base = session.inner.slot_count();
let cap = base + 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(),
cap,
"pool never exceeds the cap"
);
assert_eq!(admitted, cap - base, "exactly cap-base slots admitted");
assert!(refused >= 1, "attaches past the cap are refused");
}
}