Google Git
Sign in
fuchsia / fuchsia / refs/heads/releases/canary / . / src / lib / fuchsia-async / src / handle / emulated / mod.rs
blob: a120d6361c3f81117328cbb3df7670a8c5148c59 [file] [log] [blame]
// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//! Emulation of Zircon handles for non-Zircon based operating systems.
pub mod channel;
pub mod socket;
use crate::invoke_for_handle_types;
use bitflags::bitflags;
use fuchsia_zircon_status as zx_status;
use fuchsia_zircon_types as zx_types;
use futures::channel::oneshot;
use futures::ready;
use futures::task::noop_waker_ref;
#[cfg(debug_assertions)]
use std::cell::Cell;
use std::collections::{HashMap, VecDeque};
use std::future::{poll_fn, Future};
use std::marker::PhantomData;
use std::mem::MaybeUninit;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::task::{Context, Poll, Waker};
use zx_status::Status;
/// Invalid handle value
const INVALID_HANDLE: u32 = 0;
/// The type of an object.
#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash, Copy, Clone)]
pub struct ObjectType(zx_types::zx_obj_type_t);
macro_rules! define_object_type_constant {
($x:tt, $docname:expr, $name:ident, $zx_name:ident, $availability:ident) => {
#[doc = $docname]
pub const $name: ObjectType = ObjectType(zx_types::$zx_name);
};
}
impl ObjectType {
/// No object.
pub const NONE: ObjectType = ObjectType(0);
invoke_for_handle_types!(define_object_type_constant);
/// Creates an `ObjectType` from the underlying zircon type.
pub const fn from_raw(raw: u32) -> Self {
Self(raw)
}
/// Converts `ObjectType` into the underlying zircon type.
pub const fn into_raw(self) -> u32 {
self.0
}
}
/// A borrowed reference to an underlying handle
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct HandleRef<'a>(u32, std::marker::PhantomData<&'a u32>);
impl<'a> HandleRef<'a> {
/// Returns an invalid handle ref
pub fn invalid() -> Self {
Self(INVALID_HANDLE, std::marker::PhantomData)
}
/// Duplicate this handle. The new handle will have the given `rights`, which must be a subset
/// of the rights the existing handle has.
pub fn duplicate(&self, mut rights: Rights) -> Result<Handle, Status> {
let mut new_entry = HANDLE_TABLE
.lock()
.unwrap()
.get(&self.raw_handle())
.cloned()
.ok_or(Status::BAD_HANDLE)?;
if rights == Rights::SAME_RIGHTS {
rights = new_entry.rights;
}
if !new_entry.rights.contains(rights) {
return Err(Status::INVALID_ARGS);
}
if !new_entry.rights.contains(Rights::DUPLICATE) {
return Err(Status::ACCESS_DENIED);
}
new_entry.rights = rights;
let new_handle = alloc_handle();
let side = new_entry.side;
new_entry.object.lock().unwrap().increment_open_count(side);
let _ = HANDLE_TABLE.lock().unwrap().insert(new_handle, new_entry);
Ok(Handle(new_handle))
}
/// Signal an object
pub fn signal(&self, clear_mask: Signals, set_mask: Signals) -> Result<(), Status> {
if self.is_invalid() {
return Err(zx_status::Status::BAD_HANDLE);
}
clear_mask.validate_user_signals()?;
set_mask.validate_user_signals()?;
let rights = get_hdl_rights(self.raw_handle()).ok_or(zx_status::Status::BAD_HANDLE)?;
if !rights.contains(Rights::SIGNAL) {
Err(Status::ACCESS_DENIED)
} else {
with_handle(self.0, |mut h, side| {
// We just checked for an invalid handle above, so this should never fail.
h.as_hdl_data()
.signal(side, clear_mask, set_mask)
.status_for_peer()
.expect("Handle became invalid while processing signal call");
});
Ok(())
}
}
}
#[derive(Debug, Eq, PartialEq, Hash)]
#[repr(transparent)]
/// A Kernel Object ID
pub struct Koid(u64);
const INVALID_KOID: Koid = Koid(0);
impl Koid {
/// Get the raw u64 value of the koid
pub fn raw_koid(&self) -> u64 {
self.0
}
}
#[derive(Debug, Eq, PartialEq)]
pub struct HandleBasicInfo {
pub koid: Koid,
pub rights: Rights,
pub object_type: ObjectType,
pub related_koid: Koid,
pub reserved: u32,
}
/// A trait to get a reference to the underlying handle of an object.
pub trait AsHandleRef {
/// Get a reference to the handle.
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a>;
/// Return true if this handle is invalid
fn is_invalid(&self) -> bool {
self.as_handle_ref().0 == INVALID_HANDLE
}
/// Interpret the reference as a raw handle (an integer type). Two distinct
/// handles will have different raw values (so it can perhaps be used as a
/// key in a data structure).
fn raw_handle(&self) -> u32 {
self.as_handle_ref().0
}
/// Set and clear userspace-accessible signal bits on an object.
fn signal_handle(&self, clear_mask: Signals, set_mask: Signals) -> Result<(), Status> {
self.as_handle_ref().signal(clear_mask, set_mask)
}
/// Wraps the
/// [zx_object_get_info](https://fuchsia.dev/fuchsia-src/reference/syscalls/object_get_info.md)
/// syscall for the ZX_INFO_HANDLE_BASIC topic.
fn basic_info(&self) -> Result<HandleBasicInfo, zx_status::Status> {
if self.is_invalid() {
return Err(zx_status::Status::BAD_HANDLE);
}
let koids = with_handle(self.raw_handle(), |mut h, side| {
let h = h.as_hdl_data();
h.koids(side)
});
let ty = get_hdl_type(self.raw_handle()).ok_or(zx_status::Status::BAD_HANDLE)?;
let rights = get_hdl_rights(self.raw_handle()).ok_or(zx_status::Status::BAD_HANDLE)?;
Ok(HandleBasicInfo {
koid: Koid(koids.0),
rights,
object_type: ty.object_type(),
related_koid: Koid(koids.1),
reserved: 0,
})
}
/// Returns the koid (kernel object ID) for this handle.
fn get_koid(&self) -> Result<Koid, zx_status::Status> {
let koid = with_handle(self.raw_handle(), |mut h, side| {
let h = h.as_hdl_data();
h.koids(side).0
});
Ok(Koid(koid))
}
}
impl<T: AsHandleRef> AsHandleRef for &T {
fn as_handle_ref(&self) -> HandleRef<'_> {
(*self).as_handle_ref()
}
}
/// A trait implemented by all handles for objects which have a peer.
pub trait Peered: HandleBased {
/// Set and clear userspace-accessible signal bits on the object's peer. Wraps the
/// [zx_object_signal_peer](https://fuchsia.dev/fuchsia-src/reference/syscalls/object_signal.md)
/// syscall.
fn signal_peer(&self, clear_mask: Signals, set_mask: Signals) -> Result<(), Status> {
if self.is_invalid() {
return Err(zx_status::Status::BAD_HANDLE);
}
clear_mask.validate_user_signals()?;
set_mask.validate_user_signals()?;
let rights = get_hdl_rights(self.raw_handle()).ok_or(zx_status::Status::BAD_HANDLE)?;
if !rights.contains(Rights::SIGNAL_PEER) {
Err(Status::ACCESS_DENIED)
} else {
with_handle(self.raw_handle(), |mut h, side| {
h.as_hdl_data().signal(side.opposite(), clear_mask, set_mask).status_for_peer()
})
}
}
}
/// An extension of `AsHandleRef` that adds non-Fuchsia-only operations.
pub trait EmulatedHandleRef: AsHandleRef {
/// Return the type of a handle.
fn object_type(&self) -> ObjectType {
if self.is_invalid() {
ObjectType::NONE
} else {
let ty = get_hdl_type(self.raw_handle()).expect("Bad handle");
ty.object_type()
}
}
/// Return a reference to the other end of a handle.
fn related<'a>(&'a self) -> HandleRef<'a> {
if self.is_invalid() {
HandleRef(INVALID_HANDLE, std::marker::PhantomData)
} else {
let table = HANDLE_TABLE.lock().unwrap();
if let Some((target, side)) =
table.get(&self.raw_handle()).map(|x| (Arc::clone(&x.object), x.side))
{
for (&handle, entry) in table.iter() {
if side == entry.side.opposite() && Arc::ptr_eq(&target, &entry.object) {
return HandleRef(handle, std::marker::PhantomData);
}
}
}
HandleRef(INVALID_HANDLE, std::marker::PhantomData)
}
}
/// Return a "koid" like value.
fn koid_pair(&self) -> (u64, u64) {
if self.is_invalid() {
(INVALID_KOID.0, INVALID_KOID.0)
} else {
with_handle(self.as_handle_ref().0, |mut h, side| h.as_hdl_data().koids(side))
}
}
/// Return true if the handle appears valid (i.e. not `Handle::invalid()`),
/// but does not exist in the table. This should not normally return true;
/// when it does, dropping the handle will cause a panic.
fn is_dangling(&self) -> bool {
if self.is_invalid() {
false
} else {
!HANDLE_TABLE.lock().unwrap().contains_key(&self.raw_handle())
}
}
}
impl<T: AsHandleRef> EmulatedHandleRef for T {}
impl AsHandleRef for HandleRef<'_> {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
/// A trait implemented by all handle-based types.
pub trait HandleBased: AsHandleRef + From<Handle> + Into<Handle> {
/// Duplicate a handle, possibly reducing the rights available.
fn duplicate_handle(&self, rights: Rights) -> Result<Self, Status> {
self.as_handle_ref().duplicate(rights).map(|handle| Self::from(handle))
}
/// Creates an instance of this type from a handle.
///
/// This is a convenience function which simply forwards to the `From` trait.
fn from_handle(handle: Handle) -> Self {
Self::from(handle)
}
/// Converts the value into its inner handle.
///
/// This is a convenience function which simply forwards to the `Into` trait.
fn into_handle(self) -> Handle {
self.into()
}
/// Converts the handle into it's raw representation.
///
/// The caller takes ownership over the raw handle, and must close it.
fn into_raw(self) -> u32 {
self.into_handle().raw_take()
}
/// Returns whether this is an invalid handle.
fn is_invalid_handle(&self) -> bool {
self.is_invalid()
}
}
/// Representation of a handle-like object
#[derive(PartialEq, Eq, Debug, Ord, PartialOrd, Hash)]
#[repr(transparent)]
pub struct Handle(u32);
impl Drop for Handle {
fn drop(&mut self) {
hdl_close(self.0);
}
}
impl AsHandleRef for Handle {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
impl HandleBased for Handle {}
impl Handle {
/// Return an invalid handle
#[inline(always)]
pub const fn invalid() -> Handle {
Handle(INVALID_HANDLE)
}
/// Return true if this handle is invalid
pub fn is_invalid(&self) -> bool {
self.0 == INVALID_HANDLE
}
/// If a raw handle is obtained from some other source, this method converts
/// it into a type-safe owned handle.
///
/// # Safety
///
/// `hdl` must be a valid raw handle. The returned `Handle` takes ownership
/// of the raw handle, so it must not be modified or closed by other means
/// after calling `from_raw`.
pub const unsafe fn from_raw(hdl: u32) -> Handle {
Handle(hdl)
}
/// Take this handle and return a new handle (leaves this handle invalid)
pub fn take(&mut self) -> Handle {
let h = Handle(self.0);
self.0 = INVALID_HANDLE;
h
}
/// Invalidates this handle and returns its raw handle.
///
/// To avoid leaking resources, the returned raw handle should eventually be
/// closed. This can be done by converting it back to a [`Handle`] with
/// [`from_raw`] and dropping it.
///
/// [`from_raw`]: Handle::from_raw
pub fn raw_take(&mut self) -> u32 {
let h = self.0;
self.0 = INVALID_HANDLE;
h
}
/// Create a replacement for a handle, possibly reducing the rights available. This invalidates
/// the original handle. Wraps the
/// [zx_handle_replace](https://fuchsia.dev/fuchsia-src/reference/syscalls/handle_replace.md)
/// syscall.
pub fn replace(self, target_rights: Rights) -> Result<Handle, zx_status::Status> {
if target_rights == Rights::SAME_RIGHTS {
return Ok(self);
}
if self.is_invalid() {
return Err(zx_status::Status::BAD_HANDLE);
}
let mut table = HANDLE_TABLE.lock().unwrap();
let std::collections::hash_map::Entry::Occupied(entry) = table.entry(self.raw_handle())
else {
return Err(zx_status::Status::BAD_HANDLE);
};
if !entry.get().rights.contains(target_rights) {
return Err(zx_status::Status::INVALID_ARGS);
}
let new_handle = alloc_handle();
let mut entry = entry.remove_entry().1;
entry.rights = target_rights;
let _ = table.insert(new_handle, entry);
Ok(Handle(new_handle))
}
}
/// An emulated Zircon Channel
#[derive(PartialEq, Eq, Debug, PartialOrd, Ord, Hash)]
pub struct Channel(u32);
impl From<Handle> for Channel {
fn from(mut hdl: Handle) -> Channel {
let out = Channel(hdl.0);
hdl.0 = INVALID_HANDLE;
out
}
}
impl From<Channel> for Handle {
fn from(mut hdl: Channel) -> Handle {
let out = unsafe { Handle::from_raw(hdl.0) };
hdl.0 = INVALID_HANDLE;
out
}
}
impl HandleBased for Channel {}
impl AsHandleRef for Channel {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
impl Peered for Channel {}
impl Drop for Channel {
fn drop(&mut self) {
hdl_close(self.0);
}
}
/// Enum indicating which protocol is being used to proxy a channel, assuming the channel is
/// proxied.
#[derive(Copy, Clone, PartialEq, Eq, Debug)]
pub enum ChannelProxyProtocol {
/// CSO protocol
Cso,
/// Legacy protocol
Legacy,
}
impl ChannelProxyProtocol {
/// Get a &str representation of the protocol.
pub fn as_str(&self) -> &'static str {
match self {
ChannelProxyProtocol::Cso => "cso",
ChannelProxyProtocol::Legacy => "legacy",
}
}
}
impl Channel {
/// Create a channel, resulting in a pair of `Channel` objects representing both
/// sides of the channel. Messages written into one maybe read from the opposite.
pub fn create() -> (Channel, Channel) {
let rights = Rights::CHANNEL_DEFAULT;
let (left, right, obj) = new_handle_pair(HdlType::Channel, rights);
let mut obj = obj.lock().unwrap();
let KObjectEntry::Channel(obj) = &mut *obj else {
unreachable!("Channel we just allocated wasn't present or wasn't a channel");
};
let mut hdl_ref = HdlRef::Channel(obj);
hdl_ref
.as_hdl_data()
.signal(Side::Left, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
hdl_ref
.as_hdl_data()
.signal(Side::Right, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
(Channel(left), Channel(right))
}
/// Returns true if the channel is closed (i.e. other side was dropped).
pub fn is_closed(&self) -> bool {
assert!(!self.is_invalid());
let Some(object) = HANDLE_TABLE.lock().unwrap().get(&self.0).map(|x| Arc::clone(&x.object))
else {
return true;
};
let object = object.lock().unwrap();
!object.is_open()
}
/// If [`is_closed`] returns true, this may return a string explaining why the handle was closed.
pub fn closed_reason(&self) -> Option<String> {
assert!(!self.is_invalid());
let Some(object) = HANDLE_TABLE.lock().unwrap().get(&self.0).map(|x| Arc::clone(&x.object))
else {
return None;
};
let object = object.lock().unwrap();
let KObjectEntry::Channel(c) = &*object else {
return None;
};
c.closed_reason.clone()
}
/// Close this channel, setting `msg` as a reason for the closure.
pub fn close_with_reason(self, msg: String) {
if self.is_invalid() {
return;
}
let Some(object) = HANDLE_TABLE.lock().unwrap().get(&self.0).map(|x| Arc::clone(&x.object))
else {
return;
};
let mut object = object.lock().unwrap();
let KObjectEntry::Channel(c) = &mut *object else {
return;
};
c.closed_reason = Some(msg)
}
/// Overnet announcing to us what protocol is being used to proxy this channel.
pub fn set_channel_proxy_protocol(&self, proto: ChannelProxyProtocol) {
assert!(!self.is_invalid());
let Some(object) = HANDLE_TABLE.lock().unwrap().get(&self.0).map(|x| Arc::clone(&x.object))
else {
return;
};
let mut object = object.lock().unwrap();
if !object.is_open() {
return;
}
let KObjectEntry::Channel(object) = &mut *object else {
unreachable!("Channel handle wasn't a channel in the handle table!");
};
if let ChannelProxyProtocolState::Waiting(waiters) = std::mem::replace(
&mut object.proxy_protocol_state,
ChannelProxyProtocolState::Set(proto),
) {
for waiter in waiters.into_iter() {
let _ = waiter.send(proto);
}
}
}
/// Receive an announcement from overnet if this channel is proxied via a particular protocol.
/// Note that this will block forever if the channel does not become the endpoint of an Overnet
/// proxy.
pub async fn get_channel_proxy_protocol(&self) -> Option<ChannelProxyProtocol> {
assert!(!self.is_invalid());
let Some(object) = HANDLE_TABLE.lock().unwrap().get(&self.0).map(|x| Arc::clone(&x.object))
else {
return None;
};
// this seemingly redundant block lets us avoid holding the object lock
// across the receiver await, potentially causing a deadlock.
let receiver = {
let mut object = object.lock().unwrap();
if !object.is_open() {
return None;
}
let KObjectEntry::Channel(object) = &mut *object else {
unreachable!("Channel handle wasn't a channel in the handle table!");
};
match &mut object.proxy_protocol_state {
ChannelProxyProtocolState::Set(state) => {
return Some(*state);
}
ChannelProxyProtocolState::Unset => {
let (sender, receiver) = oneshot::channel();
object.proxy_protocol_state = ChannelProxyProtocolState::Waiting(vec![sender]);
receiver
}
ChannelProxyProtocolState::Waiting(waiters) => {
let (sender, receiver) = oneshot::channel();
waiters.push(sender);
receiver
}
}
};
receiver.await.ok()
}
/// Read a message from a channel.
pub fn read(&self, buf: &mut MessageBuf) -> Result<(), zx_status::Status> {
let (bytes, handles) = buf.split_mut();
self.read_split(bytes, handles)
}
/// Read a message from a channel into a separate byte vector and handle vector.
pub fn read_split(
&self,
bytes: &mut Vec<u8>,
handles: &mut Vec<Handle>,
) -> Result<(), zx_status::Status> {
match self.poll_read(&mut Context::from_waker(noop_waker_ref()), bytes, handles) {
Poll::Ready(r) => r,
Poll::Pending => Err(zx_status::Status::SHOULD_WAIT),
}
}
fn poll_read(
&self,
cx: &mut Context<'_>,
bytes: &mut Vec<u8>,
handles: &mut Vec<Handle>,
) -> Poll<Result<(), zx_status::Status>> {
with_handle(self.0, |h, side| {
if let HdlRef::Channel(obj) = h {
if let Some(mut msg) = obj.q.side_mut(side.opposite()).pop_front() {
std::mem::swap(bytes, &mut msg.bytes);
std::mem::swap(handles, &mut msg.handles);
if obj.q.side_mut(side.opposite()).is_empty() {
obj.signal(side, Signals::OBJECT_READABLE, Signals::NONE)
.status_for_self()?;
}
Poll::Ready(Ok(()))
} else if obj.is_open() {
obj.wakers.side_mut(side).pending_readable(cx)
} else {
Poll::Ready(Err(zx_status::Status::PEER_CLOSED))
}
} else {
unreachable!();
}
})
}
/// Reads a message from a channel into a fixed size buffer. If either `buf` or `handles` is
/// not large enough to hold the message, then `Err((buf_len, handles_len))` is returned. The
/// caller is then expected to invoke the read function again, albeit with a resized buffer
/// large enough to fit the output values.
///
/// If there are any general errors that happen during read, then `Ok(Err(_), (0, 0))` is
/// returned.
///
/// On success, `Ok(Ok(()), (buf_len, handles_len))` is returned. As with zx_channel_read, of
/// which this function is an analogue, there are no partial reads.
///
/// It is important to remember to check the `Ok(_)` result for potential errors, like
/// `PEER_CLOSED`, for example.
pub fn read_raw(
&self,
buf: &mut [u8],
handles: &mut [MaybeUninit<Handle>],
) -> Result<(Result<(), zx_status::Status>, usize, usize), (usize, usize)> {
match self.poll_read_raw(&mut Context::from_waker(noop_waker_ref()), buf, handles) {
Poll::Ready(r) => r,
Poll::Pending => Ok((Err(zx_status::Status::SHOULD_WAIT), 0, 0)),
}
}
fn poll_read_raw(
&self,
cx: &mut Context<'_>,
buf: &mut [u8],
handles: &mut [MaybeUninit<Handle>],
) -> Poll<Result<(Result<(), zx_status::Status>, usize, usize), (usize, usize)>> {
with_handle(self.0, |h, side| {
let HdlRef::Channel(obj) = h else { unreachable!() };
let read_side = obj.q.side_mut(side.opposite());
if let Some(msg) = read_side.front() {
let msg_bytes_len = msg.bytes.len();
let msg_handles_len = msg.handles.len();
if msg_bytes_len > buf.len() || msg_handles_len > handles.len() {
Poll::Ready(Err((msg_bytes_len, msg_handles_len)))
} else {
let msg = read_side.pop_front().unwrap();
buf[..msg_bytes_len].clone_from_slice(msg.bytes.as_slice());
msg.handles
.into_iter()
.enumerate()
.for_each(|(i, hdl)| handles[i] = MaybeUninit::new(hdl));
if read_side.is_empty() {
match obj
.signal(side, Signals::OBJECT_READABLE, Signals::NONE)
.status_for_self()
{
Err(e) => {
return Poll::Ready(Ok((Err(e), msg_bytes_len, msg_handles_len)))
}
Ok(_) => {}
}
}
Poll::Ready(Ok((Ok(()), msg_bytes_len, msg_handles_len)))
}
} else if obj.is_open() {
obj.wakers.side_mut(side).pending_readable(cx)
} else {
Poll::Ready(Ok((Err(zx_status::Status::PEER_CLOSED), 0, 0)))
}
})
}
/// Read a message from a channel.
pub fn read_etc(&self, buf: &mut MessageBufEtc) -> Result<(), zx_status::Status> {
let (bytes, handles) = buf.split_mut();
self.read_etc_split(bytes, handles)
}
/// Read a message from a channel into a separate byte vector and handle vector.
pub fn read_etc_split(
&self,
bytes: &mut Vec<u8>,
handles: &mut Vec<HandleInfo>,
) -> Result<(), zx_status::Status> {
match self.poll_read_etc(&mut Context::from_waker(noop_waker_ref()), bytes, handles) {
Poll::Ready(r) => r,
Poll::Pending => Err(zx_status::Status::SHOULD_WAIT),
}
}
fn poll_read_etc(
&self,
cx: &mut Context<'_>,
bytes: &mut Vec<u8>,
handle_infos: &mut Vec<HandleInfo>,
) -> Poll<Result<(), zx_status::Status>> {
let mut handles = Vec::new();
ready!(self.poll_read(cx, bytes, &mut handles))?;
handle_infos.clear();
handle_infos.extend(handles.into_iter().map(|handle| {
let h_raw = handle.raw_handle();
let ty = get_hdl_type(h_raw).expect("Bad handle");
let rights = get_hdl_rights(h_raw).expect("Bad handle");
HandleInfo { handle, object_type: ty.object_type(), rights }
}));
Poll::Ready(Ok(()))
}
/// Write a message to a channel.
pub fn write(&self, bytes: &[u8], handles: &mut [Handle]) -> Result<(), zx_status::Status> {
let bytes_vec = bytes.to_vec();
let mut handles_vec = Vec::with_capacity(handles.len());
for i in 0..handles.len() {
handles_vec.push(std::mem::replace(&mut handles[i], Handle::invalid()));
}
with_handle(self.0, |h, side| {
if let HdlRef::Channel(obj) = h {
if !obj.is_open() {
return Err(zx_status::Status::PEER_CLOSED);
}
check_write_shutdown()?;
obj.q
.side_mut(side)
.push_back(ChannelMessage { bytes: bytes_vec, handles: handles_vec });
obj.signal(side.opposite(), Signals::NONE, Signals::OBJECT_READABLE)
.status_for_peer()?;
Ok(())
} else {
unreachable!();
}
})
}
/// Write a message to a channel.
pub fn write_etc<'a>(
&self,
bytes: &[u8],
handles: &mut [HandleDisposition<'a>],
) -> Result<(), zx_status::Status> {
let bytes_vec = bytes.to_vec();
let mut handles_vec = Vec::with_capacity(handles.len());
for hd in handles {
let op: HandleOp<'a> =
std::mem::replace(&mut hd.handle_op, HandleOp::Move(Handle::invalid()));
if let HandleOp::Move(handle) = op {
let ty = get_hdl_type(handle.raw_handle()).ok_or(zx_status::Status::BAD_HANDLE)?;
if ty.object_type() != handle.object_type()
&& handle.object_type() != ObjectType::NONE
{
return Err(zx_status::Status::INVALID_ARGS);
}
handles_vec.push(handle.replace(hd.rights)?);
} else {
panic!("unimplemented HandleOp");
}
}
with_handle(self.0, |h, side| {
if let HdlRef::Channel(obj) = h {
if !obj.is_open() {
// Move the handles outside this closure before dropping them. If any are
// channels in the same shard as this channel, dropping them will attempt to
// re-acquire the lock held by with_handle.
return Err((handles_vec, zx_status::Status::PEER_CLOSED));
}
if let Err(e) = check_write_shutdown() {
return Err((handles_vec, e));
}
obj.q
.side_mut(side)
.push_back(ChannelMessage { bytes: bytes_vec, handles: handles_vec });
obj.signal(side.opposite(), Signals::NONE, Signals::OBJECT_READABLE)
.expect("Signalling readable should never fail here");
Ok(())
} else {
unreachable!();
}
})
.map_err(|(_handles_to_drop, err)| err)
}
}
/// Socket options available portable
#[derive(Clone, Copy)]
pub enum SocketOpts {
/// A bytestream style socket
STREAM,
/// A datagram style socket
DATAGRAM,
}
/// Emulation of a Zircon Socket.
#[derive(PartialEq, Eq, Debug, PartialOrd, Ord, Hash)]
pub struct Socket(u32);
impl From<Handle> for Socket {
fn from(mut hdl: Handle) -> Socket {
let out = Socket(hdl.0);
hdl.0 = INVALID_HANDLE;
out
}
}
impl From<Socket> for Handle {
fn from(mut hdl: Socket) -> Handle {
let out = unsafe { Handle::from_raw(hdl.0) };
hdl.0 = INVALID_HANDLE;
out
}
}
impl HandleBased for Socket {}
impl AsHandleRef for Socket {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
impl Peered for Socket {}
impl Drop for Socket {
fn drop(&mut self) {
hdl_close(self.0);
}
}
impl Socket {
/// Create a streaming socket.
pub fn create_stream() -> (Socket, Socket) {
let rights = Rights::SOCKET_DEFAULT;
let (left, right, obj) = new_handle_pair(HdlType::StreamSocket, rights);
let mut obj = obj.lock().unwrap();
let KObjectEntry::StreamSocket(obj) = &mut *obj else {
unreachable!("Channel we just allocated wasn't present or wasn't a channel");
};
let mut hdl_ref = HdlRef::StreamSocket(obj);
hdl_ref
.as_hdl_data()
.signal(Side::Left, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
hdl_ref
.as_hdl_data()
.signal(Side::Right, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
(Socket(left), Socket(right))
}
/// Create a datagram socket.
pub fn create_datagram() -> (Socket, Socket) {
let rights = Rights::SOCKET_DEFAULT;
let (left, right, obj) = new_handle_pair(HdlType::DatagramSocket, rights);
let mut obj = obj.lock().unwrap();
let KObjectEntry::DatagramSocket(obj) = &mut *obj else {
unreachable!("Channel we just allocated wasn't present or wasn't a channel");
};
let mut hdl_ref = HdlRef::DatagramSocket(obj);
hdl_ref
.as_hdl_data()
.signal(Side::Left, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
hdl_ref
.as_hdl_data()
.signal(Side::Right, Signals::NONE, Signals::OBJECT_WRITABLE)
.expect("Handle wasn't open immediately after creation");
(Socket(left), Socket(right))
}
/// Write the given bytes into the socket.
/// Return value (on success) is number of bytes actually written.
pub fn write(&self, bytes: &[u8]) -> Result<usize, zx_status::Status> {
with_handle(self.0, |h, side| {
match h {
HdlRef::StreamSocket(obj) => {
if !obj.is_open() {
return Err(zx_status::Status::PEER_CLOSED);
}
check_write_shutdown()?;
obj.q.side_mut(side).extend(bytes);
obj.signal(side.opposite(), Signals::NONE, Signals::OBJECT_READABLE)
.status_for_peer()?;
}
HdlRef::DatagramSocket(obj) => {
if !obj.is_open() {
return Err(zx_status::Status::PEER_CLOSED);
}
check_write_shutdown()?;
obj.q.side_mut(side).push_back(bytes.to_vec());
obj.signal(side.opposite(), Signals::NONE, Signals::OBJECT_READABLE)
.status_for_peer()?;
}
_ => panic!("Non socket passed to Socket::write"),
}
Ok(bytes.len())
})
}
/// Return how many bytes are buffered in the socket
pub fn outstanding_read_bytes(&self) -> Result<usize, zx_status::Status> {
let (len, open) = with_handle(self.0, |h, side| match h {
HdlRef::StreamSocket(obj) => (obj.q.side(side.opposite()).len(), obj.is_open()),
HdlRef::DatagramSocket(obj) => (
obj.q.side(side.opposite()).front().map(|frame| frame.len()).unwrap_or(0),
obj.is_open(),
),
_ => panic!("Non socket passed to Socket::outstanding_read_bytes"),
});
if len > 0 {
return Ok(len);
}
if !open {
return Err(zx_status::Status::PEER_CLOSED);
}
Ok(0)
}
fn poll_read(
&self,
bytes: &mut [u8],
ctx: &mut Context<'_>,
) -> Poll<Result<usize, zx_status::Status>> {
with_handle(self.0, |h, side| match h {
HdlRef::StreamSocket(obj) => {
if bytes.is_empty() {
if obj.is_open() {
return Poll::Ready(Ok(0));
} else {
return Poll::Ready(Err(zx_status::Status::PEER_CLOSED));
}
}
let read = obj.q.side_mut(side.opposite());
let copy_bytes = std::cmp::min(bytes.len(), read.len());
if copy_bytes == 0 {
if obj.is_open() {
return obj.wakers.side_mut(side).pending_readable(ctx);
} else {
return Poll::Ready(Err(zx_status::Status::PEER_CLOSED));
}
}
for (i, b) in read.drain(..copy_bytes).enumerate() {
bytes[i] = b;
}
if read.is_empty() {
obj.signal(side, Signals::OBJECT_READABLE, Signals::NONE).status_for_self()?;
}
Poll::Ready(Ok(copy_bytes))
}
HdlRef::DatagramSocket(obj) => {
if let Some(frame) = obj.q.side_mut(side.opposite()).pop_front() {
let n = std::cmp::min(bytes.len(), frame.len());
bytes[..n].clone_from_slice(&frame[..n]);
if obj.q.side_mut(side.opposite()).is_empty() {
obj.signal(side, Signals::OBJECT_READABLE, Signals::NONE)
.status_for_self()?;
}
Poll::Ready(Ok(n))
} else if !obj.is_open() {
Poll::Ready(Err(zx_status::Status::PEER_CLOSED))
} else {
obj.wakers.side_mut(side).pending_readable(ctx)
}
}
_ => panic!("Non socket passed to Socket::read"),
})
}
/// Read bytes from the socket.
/// Return value (on success) is number of bytes actually read.
pub fn read(&self, bytes: &mut [u8]) -> Result<usize, zx_status::Status> {
match self.poll_read(bytes, &mut Context::from_waker(noop_waker_ref())) {
Poll::Ready(r) => r,
Poll::Pending => Err(zx_status::Status::SHOULD_WAIT),
}
}
/// Get an OnSignals that returns when this handle's peer closes.
pub fn on_closed(&self) -> on_signals::OnSignalsRef<'_> {
on_signals::OnSignalsRef::new(self, Signals::OBJECT_PEER_CLOSED)
}
}
/// Emulation of Zircon EventPair.
#[derive(PartialEq, Eq, Debug, PartialOrd, Ord, Hash)]
pub struct EventPair(u32);
impl From<Handle> for EventPair {
fn from(mut hdl: Handle) -> EventPair {
let out = EventPair(hdl.0);
hdl.0 = INVALID_HANDLE;
out
}
}
impl From<EventPair> for Handle {
fn from(mut hdl: EventPair) -> Handle {
let out = unsafe { Handle::from_raw(hdl.0) };
hdl.0 = INVALID_HANDLE;
out
}
}
impl HandleBased for EventPair {}
impl AsHandleRef for EventPair {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
impl Peered for EventPair {}
impl Drop for EventPair {
fn drop(&mut self) {
hdl_close(self.0);
}
}
impl EventPair {
/// Create an event pair.
pub fn create() -> (Self, Self) {
Self::try_create().unwrap()
}
/// Create an event pair.
pub fn try_create() -> Result<(EventPair, EventPair), Status> {
let rights = Rights::EVENTPAIR_DEFAULT;
let (left, right, _) = new_handle_pair(HdlType::EventPair, rights);
Ok((EventPair(left), EventPair(right)))
}
}
/// Emulation of Zircon Event.
#[derive(PartialEq, Eq, Debug, PartialOrd, Ord, Hash)]
pub struct Event(u32);
impl From<Handle> for Event {
fn from(mut hdl: Handle) -> Event {
let out = Event(hdl.0);
hdl.0 = INVALID_HANDLE;
out
}
}
impl From<Event> for Handle {
fn from(mut hdl: Event) -> Handle {
let out = unsafe { Handle::from_raw(hdl.0) };
hdl.0 = INVALID_HANDLE;
out
}
}
impl HandleBased for Event {}
impl AsHandleRef for Event {
fn as_handle_ref<'a>(&'a self) -> HandleRef<'a> {
HandleRef(self.0, std::marker::PhantomData)
}
}
impl Drop for Event {
fn drop(&mut self) {
hdl_close(self.0);
}
}
impl Event {
/// Create an event .
pub fn create() -> Event {
Event(new_handle(HdlType::Event, Rights::EVENT_DEFAULT).0)
}
}
/// A buffer for _receiving_ messages from a channel.
///
/// A `MessageBuf` is essentially a byte buffer and a vector of
/// handles, but move semantics for "taking" handles requires special handling.
///
/// Note that for sending messages to a channel, the caller manages the buffers,
/// using a plain byte slice and `Vec<Handle>`.
#[derive(Debug, Default)]
pub struct MessageBuf {
bytes: Vec<u8>,
handles: Vec<Handle>,
}
impl MessageBuf {
/// Create a new, empty, message buffer.
pub fn new() -> Self {
Default::default()
}
/// Create a new non-empty message buffer.
pub fn new_with(v: Vec<u8>, h: Vec<Handle>) -> Self {
Self { bytes: v, handles: h }
}
/// Splits apart the message buf into a vector of bytes and a vector of handles.
pub fn split_mut(&mut self) -> (&mut Vec<u8>, &mut Vec<Handle>) {
(&mut self.bytes, &mut self.handles)
}
/// Splits apart the message buf into a vector of bytes and a vector of handles.
pub fn split(self) -> (Vec<u8>, Vec<Handle>) {
(self.bytes, self.handles)
}
/// Ensure that the buffer has the capacity to hold at least `n_bytes` bytes.
pub fn ensure_capacity_bytes(&mut self, n_bytes: usize) {
ensure_capacity(&mut self.bytes, n_bytes);
}
/// Ensure that the buffer has the capacity to hold at least `n_handles` handles.
pub fn ensure_capacity_handles(&mut self, n_handles: usize) {
ensure_capacity(&mut self.handles, n_handles);
}
/// Ensure that at least n_bytes bytes are initialized (0 fill).
pub fn ensure_initialized_bytes(&mut self, n_bytes: usize) {
if n_bytes <= self.bytes.len() {
return;
}
self.bytes.resize(n_bytes, 0);
}
/// Get a reference to the bytes of the message buffer, as a `&[u8]` slice.
pub fn bytes(&self) -> &[u8] {
self.bytes.as_slice()
}
/// The number of handles in the message buffer. Note this counts the number
/// available when the message was received; `take_handle` does not affect
/// the count.
pub fn n_handles(&self) -> usize {
self.handles.len()
}
/// Take the handle at the specified index from the message buffer. If the
/// method is called again with the same index, it will return `None`, as
/// will happen if the index exceeds the number of handles available.
pub fn take_handle(&mut self, index: usize) -> Option<Handle> {
self.handles.get_mut(index).and_then(|handle| {
if handle.is_invalid() {
None
} else {
Some(std::mem::replace(handle, Handle::invalid()))
}
})
}
/// Clear the bytes and handles contained in the buf. This will drop any
/// contained handles, resulting in their resources being freed.
pub fn clear(&mut self) {
self.bytes.clear();
self.handles.clear();
}
}
/// A buffer for _receiving_ messages from a channel.
///
/// This differs from `MessageBuf` in that it holds `HandleInfo` with
/// extended handle information.
///
/// A `MessageBufEtc` is essentially a byte buffer and a vector of handle
/// infos, but move semantics for "taking" handles requires special handling.
///
/// Note that for sending messages to a channel, the caller manages the buffers,
/// using a plain byte slice and `Vec<HandleDisposition>`.
#[derive(Debug, Default)]
pub struct MessageBufEtc {
bytes: Vec<u8>,
handle_infos: Vec<HandleInfo>,
}
impl MessageBufEtc {
/// Create a new, empty, message buffer.
pub fn new() -> Self {
Default::default()
}
/// Create a new non-empty message buffer.
pub fn new_with(v: Vec<u8>, h: Vec<HandleInfo>) -> Self {
Self { bytes: v, handle_infos: h }
}
/// Splits apart the message buf into a vector of bytes and a vector of handle infos.
pub fn split_mut(&mut self) -> (&mut Vec<u8>, &mut Vec<HandleInfo>) {
(&mut self.bytes, &mut self.handle_infos)
}
/// Splits apart the message buf into a vector of bytes and a vector of handle infos.
pub fn split(self) -> (Vec<u8>, Vec<HandleInfo>) {
(self.bytes, self.handle_infos)
}
/// Ensure that the buffer has the capacity to hold at least `n_bytes` bytes.
pub fn ensure_capacity_bytes(&mut self, n_bytes: usize) {
ensure_capacity(&mut self.bytes, n_bytes);
}
/// Ensure that the buffer has the capacity to hold at least `n_handles` handle infos.
pub fn ensure_capacity_handle_infos(&mut self, n_handle_infos: usize) {
ensure_capacity(&mut self.handle_infos, n_handle_infos);
}
/// Ensure that at least n_bytes bytes are initialized (0 fill).
pub fn ensure_initialized_bytes(&mut self, n_bytes: usize) {
if n_bytes <= self.bytes.len() {
return;
}
self.bytes.resize(n_bytes, 0);
}
/// Get a reference to the bytes of the message buffer, as a `&[u8]` slice.
pub fn bytes(&self) -> &[u8] {
self.bytes.as_slice()
}
/// The number of handles in the message buffer. Note this counts the number
/// available when the message was received; `take_handle` does not affect
/// the count.
pub fn n_handle_infos(&self) -> usize {
self.handle_infos.len()
}
/// Take the handle at the specified index from the message buffer. If the
/// method is called again with the same index, it will return `None`, as
/// will happen if the index exceeds the number of handles available.
pub fn take_handle_info(&mut self, index: usize) -> Option<HandleInfo> {
self.handle_infos.get_mut(index).and_then(|handle_info| {
if handle_info.handle.is_invalid() {
None
} else {
Some(std::mem::replace(
handle_info,
HandleInfo {
handle: Handle::invalid(),
object_type: ObjectType::NONE,
rights: Rights::NONE,
},
))
}
})
}
/// Clear the bytes and handles contained in the buf. This will drop any
/// contained handles, resulting in their resources being freed.
pub fn clear(&mut self) {
self.bytes.clear();
self.handle_infos.clear();
}
}
fn ensure_capacity<T>(vec: &mut Vec<T>, size: usize) {
let len = vec.len();
if size > len {
vec.reserve(size - len);
}
}
pub mod on_signals {
use super::*;
/// Wait for some signals to be raised.
#[must_use = "futures do nothing unless polled"]
pub struct OnSignalsRef<'a> {
h: u32,
koid: u64,
signals: Signals,
_lifetime: PhantomData<&'a ()>,
}
impl Unpin for OnSignalsRef<'_> {}
impl<'a> OnSignalsRef<'a> {
/// Construct a new OnSignalsRef
pub fn new<T: AsHandleRef + 'a>(handle: T, signals: Signals) -> Self {
let handle = handle.as_handle_ref();
let h = handle.0;
with_handle(h, |mut hdl, side| Self {
h,
koid: hdl.as_hdl_data().koids(side).0,
signals,
_lifetime: PhantomData,
})
}
/// This function allows the `OnSignals` object to live for the `'static` lifetime.
///
/// It is functionally a no-op, but callers of this method should note that
/// `OnSignals` will not fire if the handle that was used to create it is dropped or
/// transferred to another process.
pub fn extend_lifetime(self) -> OnSignalsRef<'static> {
OnSignalsRef {
h: self.h,
koid: self.koid,
signals: self.signals,
_lifetime: PhantomData,
}
}
}
impl Future for OnSignalsRef<'_> {
type Output = Result<Signals, Status>;
fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
with_handle(self.h, |mut h, side| {
let h = h.as_hdl_data();
if self.koid != h.koids(side).0 || !h.is_side_open(side) {
if self.signals.contains(Signals::HANDLE_CLOSED) {
Poll::Ready(Signals::HANDLE_CLOSED)
} else {
Poll::Pending
}
} else {
h.poll_signals(ctx, side, self.signals)
}
})
.map(Ok)
}
}
}
// The inner mod is required because bitflags cannot pass the attribute through to the single
// variant, and attributes cannot be applied to macro invocations.
mod inner_signals {
// Part of the code for the NONE cases that are produced by the macro triggers the lint, but as
// a whole, the produced code is still correct.
#![allow(clippy::bad_bit_mask)] // TODO(b/303500202) Remove once addressed in bitflags.
use super::{bitflags, Status};
bitflags! {
/// Signals that can be waited upon.
///
/// See [signals](https://fuchsia.dev/fuchsia-src/concepts/kernel/signals) for more information.
#[repr(transparent)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Signals : u32 {
/// No signals
const NONE = 0x00000000;
/// All object signals
const OBJECT_ALL = 0x00ffffff;
/// All user signals
const USER_ALL = 0xff000000;
/// Object signal 0
const OBJECT_0 = 1 << 0;
/// Object signal 1
const OBJECT_1 = 1 << 1;
/// Object signal 2
const OBJECT_2 = 1 << 2;
/// Object signal 3
const OBJECT_3 = 1 << 3;
/// Object signal 4
const OBJECT_4 = 1 << 4;
/// Object signal 5
const OBJECT_5 = 1 << 5;
/// Object signal 6
const OBJECT_6 = 1 << 6;
/// Object signal 7
const OBJECT_7 = 1 << 7;
/// Object signal 8
const OBJECT_8 = 1 << 8;
/// Object signal 9
const OBJECT_9 = 1 << 9;
/// Object signal 10
const OBJECT_10 = 1 << 10;
/// Object signal 11
const OBJECT_11 = 1 << 11;
/// Object signal 12
const OBJECT_12 = 1 << 12;
/// Object signal 13
const OBJECT_13 = 1 << 13;
/// Object signal 14
const OBJECT_14 = 1 << 14;
/// Object signal 15
const OBJECT_15 = 1 << 15;
/// Object signal 16
const OBJECT_16 = 1 << 16;
/// Object signal 17
const OBJECT_17 = 1 << 17;
/// Object signal 18
const OBJECT_18 = 1 << 18;
/// Object signal 19
const OBJECT_19 = 1 << 19;
/// Object signal 20
const OBJECT_20 = 1 << 20;
/// Object signal 21
const OBJECT_21 = 1 << 21;
/// Object signal 22
const OBJECT_22 = 1 << 22;
/// Handle closed
const HANDLE_CLOSED = 1 << 23;
/// User signal 0
const USER_0 = 1 << 24;
/// User signal 1
const USER_1 = 1 << 25;
/// User signal 2
const USER_2 = 1 << 26;
/// User signal 3
const USER_3 = 1 << 27;
/// User signal 4
const USER_4 = 1 << 28;
/// User signal 5
const USER_5 = 1 << 29;
/// User signal 6
const USER_6 = 1 << 30;
/// User signal 7
const USER_7 = 1 << 31;
/// All user signals
const USER_SIGNALS = Self::USER_0.bits() |
Self::USER_1.bits() |
Self::USER_2.bits() |
Self::USER_3.bits() |
Self::USER_4.bits() |
Self::USER_5.bits() |
Self::USER_6.bits() |
Self::USER_7.bits();
/// Object is readable
const OBJECT_READABLE = Self::OBJECT_0.bits();
/// Object is writable
const OBJECT_WRITABLE = Self::OBJECT_1.bits();
/// Object peer closed
const OBJECT_PEER_CLOSED = Self::OBJECT_2.bits();
/// Channel peer closed
const CHANNEL_PEER_CLOSED = Self::OBJECT_PEER_CLOSED.bits();
}
}
impl Signals {
/// Returns `Status::INVALID_ARGS` if this signal set contains non-user signals.
pub(super) fn validate_user_signals(&self) -> Result<(), Status> {
if Signals::USER_SIGNALS.contains(*self) {
Ok(())
} else {
Err(Status::INVALID_ARGS)
}
}
}
bitflags! {
/// Rights associated with a handle.
///
/// See [rights](https://fuchsia.dev/fuchsia-src/concepts/kernel/rights) for more information.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Rights: u32 {
/// No rights.
const NONE = 0;
/// Duplicate right.
const DUPLICATE = 1 << 0;
/// Transfer right.
const TRANSFER = 1 << 1;
/// Read right.
const READ = 1 << 2;
/// Write right.
const WRITE = 1 << 3;
/// Execute right.
const EXECUTE = 1 << 4;
/// Map right.
const MAP = 1 << 5;
/// Get Property right.
const GET_PROPERTY = 1 << 6;
/// Set Property right.
const SET_PROPERTY = 1 << 7;
/// Enumerate right.
const ENUMERATE = 1 << 8;
/// Destroy right.
const DESTROY = 1 << 9;
/// Set Policy right.
const SET_POLICY = 1 << 10;
/// Get Policy right.
const GET_POLICY = 1 << 11;
/// Signal right.
const SIGNAL = 1 << 12;
/// Signal Peer right.
const SIGNAL_PEER = 1 << 13;
/// Wait right.
const WAIT = 1 << 14;
/// Inspect right.
const INSPECT = 1 << 15;
/// Manage Job right.
const MANAGE_JOB = 1 << 16;
/// Manage Process right.
const MANAGE_PROCESS = 1 << 17;
/// Manage Thread right.
const MANAGE_THREAD = 1 << 18;
/// Apply Profile right.
const APPLY_PROFILE = 1 << 19;
/// Manage Socket right.
const MANAGE_SOCKET = 1 << 20;
/// Same rights.
const SAME_RIGHTS = 1 << 31;
/// A basic set of rights for most things.
const BASIC = Rights::TRANSFER.bits() |
Rights::DUPLICATE.bits() |
Rights::WAIT.bits() |
Rights::INSPECT.bits();
/// IO related rights
const IO = Rights::WRITE.bits() |
Rights::READ.bits();
/// Rights of a new socket.
const SOCKET_DEFAULT = Rights::BASIC.bits() |
Rights::IO.bits() |
Rights::SIGNAL.bits() |
Rights::SIGNAL_PEER.bits();
/// Rights of a new channel.
const CHANNEL_DEFAULT = (Rights::BASIC.bits() & !Rights::DUPLICATE.bits()) |
Rights::IO.bits() |
Rights::SIGNAL.bits() |
Rights::SIGNAL_PEER.bits();
/// Rights of a new event pair.
const EVENTPAIR_DEFAULT =
Rights::TRANSFER.bits() |
Rights::DUPLICATE.bits() |
Rights::IO.bits() |
Rights::SIGNAL.bits() |
Rights::SIGNAL_PEER.bits();
/// Rights of a new event.
const EVENT_DEFAULT = Rights::BASIC.bits() |
Rights::SIGNAL.bits();
}
}
}
pub use inner_signals::{Rights, Signals};
/// Handle operation.
#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub enum HandleOp<'a> {
/// Move the handle.
Move(Handle),
/// Duplicate the handle.
Duplicate(HandleRef<'a>),
}
/// Operation to perform on handles during write.
/// Based on zx_handle_disposition_t, but does not match the same layout.
#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct HandleDisposition<'a> {
/// Handle operation.
pub handle_op: HandleOp<'a>,
/// Object type to check, or NONE to avoid the check.
pub object_type: ObjectType,
/// Rights to send, or SAME_RIGHTS.
/// Rights are checked against existing handle rights to ensure there is no
/// increase in rights.
pub rights: Rights,
/// Result of attempting to write this handle disposition.
pub result: Status,
}
/// HandleInfo represents a handle with additional metadata.
#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub struct HandleInfo {
/// The handle.
pub handle: Handle,
/// The type of object referenced by the handle.
pub object_type: ObjectType,
/// The rights of the handle.
pub rights: Rights,
}
impl HandleInfo {
/// # Safety
///
/// See [`Handle::from_raw`] for requirements about the validity and closing
/// of `raw.handle`.
///
/// `raw.rights` must be a bitwise combination of one or more [`Rights`]
/// with no additional bits set.
///
/// Note that while `raw.ty` _should_ correspond to the type of the handle,
/// that this is not required for safety.
pub const unsafe fn from_raw(raw: zx_types::zx_handle_info_t) -> HandleInfo {
HandleInfo {
handle: Handle::from_raw(raw.handle),
object_type: ObjectType(raw.ty),
rights: Rights::from_bits_retain(raw.rights),
}
}
}
#[derive(Default)]
struct Sided<T> {
left: T,
right: T,
}
impl<T> Sided<T> {
fn side_mut(&mut self, side: Side) -> &mut T {
match side {
Side::Left => &mut self.left,
Side::Right => &mut self.right,
}
}
fn side(&self, side: Side) -> &T {
match side {
Side::Left => &self.left,
Side::Right => &self.right,
}
}
}
struct ChannelMessage {
bytes: Vec<u8>,
handles: Vec<Handle>,
}
#[derive(Debug, Clone, Copy, PartialEq)]
enum HdlType {
Channel,
StreamSocket,
DatagramSocket,
EventPair,
Event,
}
impl HdlType {
fn object_type(&self) -> ObjectType {
match self {
HdlType::Channel => ObjectType::CHANNEL,
HdlType::StreamSocket => ObjectType::SOCKET,
HdlType::DatagramSocket => ObjectType::SOCKET,
HdlType::EventPair => ObjectType::EVENTPAIR,
HdlType::Event => ObjectType::EVENT,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
enum Side {
Left,
Right,
}
impl Side {
fn opposite(self) -> Side {
match self {
Side::Left => Side::Right,
Side::Right => Side::Left,
}
}
}
#[derive(Default)]
struct WakerSlot(Vec<Waker>);
impl WakerSlot {
fn wake(&mut self) {
self.0.drain(..).for_each(|w| w.wake());
}
fn arm(&mut self, ctx: &mut Context<'_>) {
self.0.push(ctx.waker().clone())
}
}
#[derive(Default)]
struct Wakers {
signals: [WakerSlot; 32],
}
impl Wakers {
fn for_signals_in(&mut self, signals: Signals, mut f: impl FnMut(&mut WakerSlot)) {
for i in 0..32 {
if signals.bits() & (1 << i) != 0 {
f(&mut self.signals[i])
}
}
}
fn wake(&mut self, signals: Signals) {
self.for_signals_in(signals, |w| w.wake())
}
fn arm(&mut self, signals: Signals, ctx: &mut Context<'_>) {
self.for_signals_in(signals, |w| w.arm(ctx))
}
fn pending<R>(&mut self, signals: Signals, ctx: &mut Context<'_>) -> Poll<R> {
self.arm(signals, ctx);
Poll::Pending
}
fn pending_readable<R>(&mut self, ctx: &mut Context<'_>) -> Poll<R> {
self.pending(
Signals::OBJECT_READABLE | Signals::HANDLE_CLOSED | Signals::OBJECT_PEER_CLOSED,
ctx,
)
}
}
#[derive(Debug)]
enum SignalError {
HandleInvalid,
}
trait SignalErrorToStatus<T> {
fn status_for_self(self) -> Result<T, Status>;
fn status_for_peer(self) -> Result<T, Status>;
}
impl<T> SignalErrorToStatus<T> for Result<T, SignalError> {
fn status_for_self(self) -> Result<T, Status> {
self.map_err(|e| {
let SignalError::HandleInvalid = e;
Status::BAD_HANDLE
})
}
fn status_for_peer(self) -> Result<T, Status> {
self.map_err(|e| {
let SignalError::HandleInvalid = e;
Status::PEER_CLOSED
})
}
}
trait HdlData {
fn signal(
&mut self,
side: Side,
clear_mask: Signals,
set_mask: Signals,
) -> Result<(), SignalError>;
fn poll_signals(
&mut self,
ctx: &mut Context<'_>,
side: Side,
signals: Signals,
) -> Poll<Signals>;
fn koids(&self, side: Side) -> (u64, u64);
fn is_side_open(&self, side: Side) -> bool;
}
enum ChannelProxyProtocolState {
Unset,
Waiting(Vec<oneshot::Sender<ChannelProxyProtocol>>),
Set(ChannelProxyProtocol),
}
struct KObject<Q> {
two_sided: bool,
proxy_protocol_state: ChannelProxyProtocolState,
q: Sided<Q>,
wakers: Sided<Wakers>,
open_count: Sided<usize>,
koid_left: u64,
signals: Sided<Signals>,
closed_reason: Option<String>,
drain_waker: Option<Waker>,
}
impl<Q> KObject<Q> {
fn is_open(&self) -> bool {
*self.open_count.side(Side::Left) != 0
&& (!self.two_sided || *self.open_count.side(Side::Right) != 0)
}
/// Returns true unless:
/// 1) We are a two-sided handle.
/// 2) One side is closed.
/// 3) The other side hasn't read all the data.
fn is_drained(&self) -> bool {
if !self.two_sided || self.is_open() {
return true;
}
if *self.open_count.side(Side::Left) == 0
&& self.signals.side(Side::Right).contains(Signals::OBJECT_READABLE)
{
return false;
}
if *self.open_count.side(Side::Right) == 0
&& self.signals.side(Side::Left).contains(Signals::OBJECT_READABLE)
{
return false;
}
return true;
}
fn poll_drained(&mut self, ctx: &mut Context<'_>) -> Poll<()> {
if self.is_drained() {
Poll::Ready(())
} else {
self.drain_waker = Some(ctx.waker().clone());
Poll::Pending
}
}
fn do_close(&mut self, side: Side) {
let open_count = self.open_count.side_mut(side);
if *open_count == 0 {
return;
}
*open_count = open_count.saturating_sub(1);
if *open_count == 0 {
self.proxy_protocol_state = ChannelProxyProtocolState::Unset;
self.wakers.side_mut(side).wake(Signals::HANDLE_CLOSED);
if *self.open_count.side(side.opposite()) != 0 {
self.wakers.side_mut(side).wake(Signals::HANDLE_CLOSED);
self.signal(side.opposite(), Signals::empty(), Signals::OBJECT_PEER_CLOSED)
.expect("Close reported other side was open but we could not signal it.");
}
}
}
}
impl<Q> HdlData for KObject<Q> {
fn signal(
&mut self,
side: Side,
clear_mask: Signals,
set_mask: Signals,
) -> Result<(), SignalError> {
if *self.open_count.side(side) == 0 {
return Err(SignalError::HandleInvalid);
}
let was_drained = self.is_drained();
let signals = self.signals.side_mut(side);
signals.remove(clear_mask);
signals.insert(set_mask);
self.wakers.side_mut(side).wake(*signals);
if !was_drained && self.is_drained() {
if let Some(waker) = self.drain_waker.take() {
waker.wake()
}
}
Ok(())
}
fn poll_signals(
&mut self,
ctx: &mut Context<'_>,
side: Side,
signals: Signals,
) -> Poll<Signals> {
let intersection = *self.signals.side(side) & signals;
if !intersection.is_empty() {
Poll::Ready(intersection)
} else {
self.wakers.side_mut(side).pending(signals, ctx)
}
}
fn koids(&self, side: Side) -> (u64, u64) {
if !self.two_sided {
assert_eq!(side, Side::Left);
return (self.koid_left, INVALID_KOID.0);
}
match side {
Side::Left => (self.koid_left, self.koid_left + 1),
Side::Right => (self.koid_left + 1, self.koid_left),
}
}
fn is_side_open(&self, side: Side) -> bool {
*self.open_count.side(side) != 0
}
}
enum KObjectEntry {
Channel(KObject<VecDeque<ChannelMessage>>),
StreamSocket(KObject<VecDeque<u8>>),
DatagramSocket(KObject<VecDeque<Vec<u8>>>),
EventPair(KObject<()>),
Event(KObject<()>),
}
impl KObjectEntry {
fn is_open(&self) -> bool {
match self {
KObjectEntry::Channel(k) => k.is_open(),
KObjectEntry::StreamSocket(k) => k.is_open(),
KObjectEntry::DatagramSocket(k) => k.is_open(),
KObjectEntry::EventPair(k) => k.is_open(),
KObjectEntry::Event(k) => k.is_open(),
}
}
fn increment_open_count(&mut self, side: Side) {
match self {
KObjectEntry::Channel(k) => *k.open_count.side_mut(side) += 1,
KObjectEntry::StreamSocket(k) => *k.open_count.side_mut(side) += 1,
KObjectEntry::DatagramSocket(k) => *k.open_count.side_mut(side) += 1,
KObjectEntry::EventPair(k) => *k.open_count.side_mut(side) += 1,
KObjectEntry::Event(k) => *k.open_count.side_mut(side) += 1,
}
}
fn do_close(&mut self, side: Side) {
match self {
KObjectEntry::Channel(k) => k.do_close(side),
KObjectEntry::StreamSocket(k) => k.do_close(side),
KObjectEntry::DatagramSocket(k) => k.do_close(side),
KObjectEntry::EventPair(k) => k.do_close(side),
KObjectEntry::Event(k) => k.do_close(side),
}
}
}
#[derive(Clone)]
struct HandleTableEntry {
object: Arc<Mutex<KObjectEntry>>,
rights: Rights,
side: Side,
}
/// Allocate two new handles which point to the peered sides of a single object. The `hdl_type` must
/// be a handle type that refers to a paired object, e.g. it should not be `HdlType::Event`.
fn new_handle_pair(hdl_type: HdlType, rights: Rights) -> (u32, u32, Arc<Mutex<KObjectEntry>>) {
fn new_kobject<T: Default>() -> KObject<T> {
KObject {
two_sided: true,
proxy_protocol_state: ChannelProxyProtocolState::Unset,
q: Default::default(),
wakers: Default::default(),
open_count: Sided { left: 1, right: 1 },
koid_left: NEXT_KOID.fetch_add(2, Ordering::Relaxed),
signals: Sided { left: Signals::empty(), right: Signals::empty() },
closed_reason: None,
drain_waker: None,
}
}
let kobject_entry = Arc::new(Mutex::new(match hdl_type {
HdlType::Channel => KObjectEntry::Channel(new_kobject()),
HdlType::StreamSocket => KObjectEntry::StreamSocket(new_kobject()),
HdlType::DatagramSocket => KObjectEntry::DatagramSocket(new_kobject()),
HdlType::EventPair => KObjectEntry::EventPair(new_kobject()),
HdlType::Event => panic!("Can't create a paired handle for the unpaired Event handle type"),
}));
let left = HandleTableEntry { object: Arc::clone(&kobject_entry), rights, side: Side::Left };
let right = HandleTableEntry { object: Arc::clone(&kobject_entry), rights, side: Side::Right };
let left_handle = alloc_handle();
let right_handle = alloc_handle();
let mut handle_table = HANDLE_TABLE.lock().unwrap();
let _ = handle_table.insert(left_handle, left);
let _ = handle_table.insert(right_handle, right);
(left_handle, right_handle, kobject_entry)
}
/// Allocate a new handle which points to a single object. The `hdl_type` must be a handle type that
/// does not refer to a paired object, i.e. it must be `HdlType::Event`.
fn new_handle(hdl_type: HdlType, rights: Rights) -> (u32, Arc<Mutex<KObjectEntry>>) {
fn new_kobject<T: Default>() -> KObject<T> {
KObject {
two_sided: false,
proxy_protocol_state: ChannelProxyProtocolState::Unset,
q: Default::default(),
wakers: Default::default(),
open_count: Sided { left: 1, right: 0 },
koid_left: NEXT_KOID.fetch_add(2, Ordering::Relaxed),
signals: Sided { left: Signals::empty(), right: Signals::empty() },
closed_reason: None,
drain_waker: None,
}
}
let kobject_entry = Arc::new(Mutex::new(match hdl_type {
HdlType::Event => KObjectEntry::Event(new_kobject()),
_ => panic!("Cannot create single handle for paired object"),
}));
let left = HandleTableEntry { object: Arc::clone(&kobject_entry), rights, side: Side::Left };
let left_handle = alloc_handle();
let mut handle_table = HANDLE_TABLE.lock().unwrap();
let _ = handle_table.insert(left_handle, left);
(left_handle, kobject_entry)
}
fn alloc_handle() -> u32 {
FREE_HANDLES
.lock()
.unwrap()
.pop_front()
.unwrap_or_else(|| NEXT_HANDLE.fetch_add(1, Ordering::Relaxed))
}
enum HdlRef<'a> {
Channel(&'a mut KObject<VecDeque<ChannelMessage>>),
StreamSocket(&'a mut KObject<VecDeque<u8>>),
DatagramSocket(&'a mut KObject<VecDeque<Vec<u8>>>),
EventPair(&'a mut KObject<()>),
Event(&'a mut KObject<()>),
}
impl<'a> HdlRef<'a> {
fn as_hdl_data<'b>(&'b mut self) -> &'b mut dyn HdlData {
match self {
HdlRef::Channel(hdl) => *hdl,
HdlRef::StreamSocket(hdl) => *hdl,
HdlRef::DatagramSocket(hdl) => *hdl,
HdlRef::EventPair(hdl) => *hdl,
HdlRef::Event(hdl) => *hdl,
}
}
}
#[cfg(debug_assertions)]
std::thread_local! {
static IN_WITH_HANDLE: Cell<bool> = Cell::new(false);
}
fn with_handle<R>(handle: u32, f: impl FnOnce(HdlRef<'_>, Side) -> R) -> R {
#[cfg(debug_assertions)]
IN_WITH_HANDLE.with(|iwh| assert_eq!(iwh.replace(true), false));
let (side, object) = {
let handle_table = HANDLE_TABLE.lock().unwrap();
let entry = handle_table.get(&handle).expect("Tried to use dangling handle");
(entry.side, Arc::clone(&entry.object))
};
let mut object = object.lock().unwrap();
let r = match &mut *object {
KObjectEntry::Channel(o) => f(HdlRef::Channel(&mut *o), side),
KObjectEntry::StreamSocket(o) => f(HdlRef::StreamSocket(&mut *o), side),
KObjectEntry::DatagramSocket(o) => f(HdlRef::DatagramSocket(&mut *o), side),
KObjectEntry::EventPair(o) => f(HdlRef::EventPair(&mut *o), side),
KObjectEntry::Event(o) => f(HdlRef::Event(&mut *o), side),
};
#[cfg(debug_assertions)]
IN_WITH_HANDLE.with(|iwh| assert_eq!(iwh.replace(false), true));
r
}
lazy_static::lazy_static! {
static ref HANDLE_TABLE: Mutex<HashMap<u32, HandleTableEntry>> = Default::default();
static ref FREE_HANDLES: Mutex<VecDeque<u32>> = Default::default();
}
static NEXT_KOID: AtomicU64 = AtomicU64::new(1);
static NEXT_HANDLE: AtomicU32 = AtomicU32::new(1);
static SHUTTING_DOWN: AtomicBool = AtomicBool::new(false);
/// Flush all data buffered in emulated handles, then force them to close. If you want to preserve
/// the property that writing to a handle and then closing it means all the data got to the other
/// side, and your handles are being serviced by Overnet rather than a local process, you should
/// probably call this before the process exits. Note that this guarantees the data has left the
/// emulated handle layer, not that it's made it over the network. Also this could in theory block
/// forever so you should time out around it.
pub async fn shut_down_handles() {
SHUTTING_DOWN.store(true, Ordering::Release);
poll_fn(|ctx| {
let handle_table = HANDLE_TABLE.lock().unwrap();
for v in handle_table.values() {
let mut object = v.object.lock().unwrap();
match &mut *object {
KObjectEntry::Channel(o) => ready!(o.poll_drained(ctx)),
KObjectEntry::StreamSocket(o) => ready!(o.poll_drained(ctx)),
KObjectEntry::DatagramSocket(o) => ready!(o.poll_drained(ctx)),
KObjectEntry::EventPair(o) => ready!(o.poll_drained(ctx)),
KObjectEntry::Event(o) => ready!(o.poll_drained(ctx)),
}
}
Poll::Ready(())
})
.await;
let handle_table = HANDLE_TABLE.lock().unwrap();
for v in handle_table.values() {
let mut object = v.object.lock().unwrap();
object.do_close(Side::Left);
object.do_close(Side::Right);
}
}
fn check_write_shutdown() -> Result<(), zx_status::Status> {
if SHUTTING_DOWN.load(Ordering::Acquire) {
Err(zx_status::Status::SHOULD_WAIT)
} else {
Ok(())
}
}
fn get_hdl_type(handle: u32) -> Option<HdlType> {
let object = {
let table = HANDLE_TABLE.lock().unwrap();
Arc::clone(&table.get(&handle)?.object)
};
let object = object.lock().unwrap();
match &*object {
KObjectEntry::Channel(_) => Some(HdlType::Channel),
KObjectEntry::StreamSocket(_) => Some(HdlType::StreamSocket),
KObjectEntry::DatagramSocket(_) => Some(HdlType::DatagramSocket),
KObjectEntry::EventPair(_) => Some(HdlType::EventPair),
KObjectEntry::Event(_) => Some(HdlType::Event),
}
}
fn get_hdl_rights(handle: u32) -> Option<Rights> {
let table = HANDLE_TABLE.lock().unwrap();
table.get(&handle).map(|x| x.rights)
}
/// Close the handle: no action if hdl==INVALID_HANDLE
fn hdl_close(hdl: u32) {
if hdl == INVALID_HANDLE {
return;
}
let Some(entry) = HANDLE_TABLE.lock().unwrap().remove(&hdl) else {
return;
};
entry.object.lock().unwrap().do_close(entry.side);
FREE_HANDLES.lock().unwrap().push_back(hdl);
}
#[cfg(test)]
mod test {
use super::*;
use fuchsia_zircon_status as zx_status;
use futures::FutureExt;
use std::mem::ManuallyDrop;
/// Returns a "handle" which mimics a closed handle.
///
/// This handle is not a real handle, and will cause panics most places it is used. It should
/// be impossible to get this kind of handle safely, but some tests want to assert what happens
/// if you try to use a closed handle.
///
/// The result is wrapped in a ManuallyDrop because you should avoid dropping the resulting
/// handle because that too will trigger a panic.
#[deny(unsafe_op_in_unsafe_fn)]
unsafe fn mimic_closed_handle() -> std::mem::ManuallyDrop<Handle> {
let raw_unallocated_handle = alloc_handle();
let unallocated_handle = unsafe { Handle::from_raw(raw_unallocated_handle) };
std::mem::ManuallyDrop::new(unallocated_handle)
}
#[test]
fn channel_create_not_closed() {
let (a, b) = Channel::create();
assert_eq!(a.is_closed(), false);
assert!(a.closed_reason().is_none());
assert_eq!(b.is_closed(), false);
assert!(b.closed_reason().is_none());
}
#[test]
fn channel_drop_left_closes_right() {
let (a, b) = Channel::create();
drop(a);
assert_eq!(b.is_closed(), true);
assert!(b.closed_reason().is_none());
}
#[test]
fn channel_drop_right_closes_left() {
let (a, b) = Channel::create();
drop(b);
assert_eq!(a.is_closed(), true);
assert!(a.closed_reason().is_none());
}
#[test]
fn channel_close_message() {
let (a, b) = Channel::create();
assert_eq!(a.is_closed(), false);
assert!(a.closed_reason().is_none());
b.close_with_reason("Testing reason!!".to_owned());
assert_eq!(a.is_closed(), true);
assert_eq!(a.closed_reason().unwrap().as_str(), "Testing reason!!");
}
#[test]
fn channel_read_raw() {
const UNINIT: MaybeUninit<Handle> = MaybeUninit::<Handle>::uninit();
let (a, b) = Channel::create();
let (c, d) = Channel::create();
let mut buf: [u8; 2] = [0, 0];
let mut handles = [UNINIT; 2];
assert_eq!(
b.read_raw(&mut buf, &mut handles).ok().unwrap(),
(Err(Status::SHOULD_WAIT), 0, 0)
);
d.write(&[4, 5, 6], &mut vec![]).unwrap();
a.write(&[1, 2, 3], &mut vec![c.into(), d.into()]).unwrap();
// Should err even though handle length is the same.
let (b_len, h_len) = b.read_raw(&mut buf[..], &mut handles[..]).err().unwrap();
assert_eq!(b_len, 3);
assert_eq!(h_len, 2);
let mut buf = [0, 0, 0];
assert_eq!(b.read_raw(&mut buf, &mut handles).ok().unwrap(), (Ok(()), 3, 2));
assert_eq!(buf, [1, 2, 3]);
assert_eq!(handles.len(), 2);
let mut handles_iter = handles.into_iter().take(2);
let c: Channel = unsafe { handles_iter.next().unwrap().assume_init() }.into();
let d: Channel = unsafe { handles_iter.next().unwrap().assume_init() }.into();
let mut handles = [UNINIT; 0];
assert_eq!(c.read_raw(&mut buf, &mut handles).ok().unwrap(), (Ok(()), 3, 0));
assert_eq!(buf, [4, 5, 6]);
b.write(&[1, 2], &mut vec![c.into(), d.into()]).unwrap();
// Checks that having an incorrect handle buffer size also fails.
assert_eq!(a.read_raw(&mut buf, &mut handles).err().unwrap(), (2, 2));
let mut handles = [UNINIT; 2];
// Verifies that copying into an "oversized" buffer does not fail (copy_to_slice panics
// if the slices are not the same size).
let _ = a.read_raw(&mut buf, &mut handles).unwrap();
let mut handles_iter = handles.into_iter().take(2);
let c: Channel = unsafe { handles_iter.next().unwrap().assume_init() }.into();
let d: Channel = unsafe { handles_iter.next().unwrap().assume_init() }.into();
// Verifies that the passed channels weren't closed after being moved.
c.write(&[6, 7, 8], &mut vec![]).unwrap();
let mut handles = [UNINIT; 0];
assert_eq!(d.read_raw(&mut buf, &mut handles).unwrap(), (Ok(()), 3, 0));
assert_eq!(buf, [6, 7, 8]);
}
#[test]
fn channel_write_read() {
let (a, b) = Channel::create();
let (c, d) = Channel::create();
let mut incoming = MessageBuf::new();
assert_eq!(b.read(&mut incoming).err().unwrap(), Status::SHOULD_WAIT);
d.write(&[4, 5, 6], &mut vec![]).unwrap();
a.write(&[1, 2, 3], &mut vec![c.into(), d.into()]).unwrap();
b.read(&mut incoming).unwrap();
assert_eq!(incoming.bytes(), &[1, 2, 3]);
assert_eq!(incoming.n_handles(), 2);
let c: Channel = incoming.take_handle(0).unwrap().into();
let d: Channel = incoming.take_handle(1).unwrap().into();
c.read(&mut incoming).unwrap();
drop(d);
assert_eq!(incoming.bytes(), &[4, 5, 6]);
assert_eq!(incoming.n_handles(), 0);
}
#[test]
fn channel_write_etc_read_etc() {
let (a, b) = Channel::create();
let (c, d) = Channel::create();
let mut incoming = MessageBufEtc::new();
assert_eq!(b.read_etc(&mut incoming).err().unwrap(), Status::SHOULD_WAIT);
d.write(&[4, 5, 6], &mut vec![]).unwrap();
let mut hds = vec![
HandleDisposition {
handle_op: HandleOp::Move(c.into()),
object_type: ObjectType::CHANNEL,
rights: Rights::SAME_RIGHTS,
result: Status::OK,
},
HandleDisposition {
handle_op: HandleOp::Move(d.into()),
object_type: ObjectType::CHANNEL,
rights: Rights::TRANSFER | Rights::READ,
result: Status::OK,
},
];
a.write_etc(&[1, 2, 3], &mut hds).unwrap();
b.read_etc(&mut incoming).unwrap();
assert_eq!(incoming.bytes(), &[1, 2, 3]);
assert_eq!(incoming.n_handle_infos(), 2);
let mut c_handle_info = incoming.take_handle_info(0).unwrap();
assert_eq!(c_handle_info.object_type, ObjectType::CHANNEL);
assert_eq!(c_handle_info.rights, Rights::CHANNEL_DEFAULT);
let mut d_handle_info = incoming.take_handle_info(1).unwrap();
assert_eq!(d_handle_info.object_type, ObjectType::CHANNEL);
assert_eq!(d_handle_info.rights, Rights::TRANSFER | Rights::READ);
let c: Channel = std::mem::replace(&mut c_handle_info.handle, Handle::invalid()).into();
let d: Channel = std::mem::replace(&mut d_handle_info.handle, Handle::invalid()).into();
c.read_etc(&mut incoming).unwrap();
drop(d);
assert_eq!(incoming.bytes(), &[4, 5, 6]);
assert_eq!(incoming.n_handle_infos(), 0);
}
#[test]
fn mixed_channel_write_read_etc() {
let (a, b) = Channel::create();
let (c, _) = Channel::create();
a.write(&[1, 2, 3], &mut [c.into()]).unwrap();
let mut buf = MessageBufEtc::new();
b.read_etc(&mut buf).unwrap();
assert_eq!(buf.bytes(), &[1, 2, 3]);
assert_eq!(buf.n_handle_infos(), 1);
let hi = &buf.handle_infos[0];
assert_eq!(hi.object_type, ObjectType::CHANNEL);
assert_eq!(hi.rights, Rights::CHANNEL_DEFAULT);
assert_ne!(hi.handle, Handle::invalid());
}
#[test]
fn mixed_channel_write_etc_read() {
let (a, b) = Channel::create();
let (c, _) = Channel::create();
let hd = HandleDisposition {
handle_op: HandleOp::Move(c.into()),
object_type: ObjectType::NONE,
rights: Rights::SAME_RIGHTS,
result: Status::OK,
};
a.write_etc(&[1, 2, 3], &mut [hd]).unwrap();
let mut buf = MessageBuf::new();
b.read(&mut buf).unwrap();
assert_eq!(buf.bytes(), &[1, 2, 3]);
assert_eq!(buf.n_handles(), 1);
assert_ne!(buf.handles[0], Handle::invalid());
}
#[test]
fn socket_write_read() {
let (a, b) = Socket::create_stream();
a.write(&[1, 2, 3]).unwrap();
let mut buf = [0u8; 128];
assert_eq!(b.read(&mut buf).unwrap(), 3);
assert_eq!(&buf[0..3], &[1, 2, 3]);
}
#[test]
fn socket_dup_write_read() {
let (a, b) = Socket::create_stream();
let c = a.duplicate_handle(Rights::SAME_RIGHTS).unwrap();
a.write(&[1, 2, 3]).unwrap();
c.write(&[4, 5, 6]).unwrap();
drop(c);
a.write(&[7, 8, 9]).unwrap();
let mut buf = [0u8; 128];
assert_eq!(b.read(&mut buf).unwrap(), 9);
assert_eq!(&buf[0..9], &[1, 2, 3, 4, 5, 6, 7, 8, 9]);
}
#[test]
fn socket_write_dup_read() {
let (a, b) = Socket::create_stream();
let c = b.duplicate_handle(Rights::SAME_RIGHTS).unwrap();
a.write(&[1, 2, 3, 4, 5, 6]).unwrap();
let mut buf = [0u8; 3];
assert_eq!(b.read(&mut buf).unwrap(), 3);
assert_eq!(&buf[0..3], &[1, 2, 3]);
drop(b);
assert_eq!(c.read(&mut buf).unwrap(), 3);
assert_eq!(&buf[0..3], &[4, 5, 6]);
}
#[test]
fn socket_dup_requires_right() {
let (_a, b) = Socket::create_stream();
let c = b.duplicate_handle(Rights::SOCKET_DEFAULT & !Rights::DUPLICATE).unwrap();
assert!(matches!(c.duplicate_handle(Rights::SAME_RIGHTS), Err(Status::ACCESS_DENIED)));
}
#[test]
fn channel_basic() {
let (p1, p2) = Channel::create();
let mut empty = vec![];
assert!(p1.write(b"hello", &mut empty).is_ok());
let mut buf = MessageBuf::new();
assert!(p2.read(&mut buf).is_ok());
assert_eq!(buf.bytes(), b"hello");
}
#[test]
fn channel_send_handle() {
let hello_length: usize = 5;
// Create a pair of channels and a pair of sockets.
let (p1, p2) = Channel::create();
let (s1, s2) = Socket::create_stream();
// Send one socket down the channel
let mut handles_to_send: Vec<Handle> = vec![s1.into_handle()];
assert!(p1.write(b"", &mut handles_to_send).is_ok());
// The handle vector should only contain invalid handles.
for handle in handles_to_send {
assert!(handle.is_invalid());
}
// Read the handle from the receiving channel.
let mut buf = MessageBuf::new();
assert!(p2.read(&mut buf).is_ok());
assert_eq!(buf.n_handles(), 1);
// Take the handle from the buffer.
let received_handle = buf.take_handle(0).unwrap();
// Should not affect number of handles.
assert_eq!(buf.n_handles(), 1);
// Trying to take it again should fail.
assert!(buf.take_handle(0).is_none());
// Now to test that we got the right handle, try writing something to it...
let received_socket = Socket::from(received_handle);
assert!(received_socket.write(b"hello").is_ok());
// ... and reading it back from the original VMO.
let mut read_vec = vec![0; hello_length];
assert!(s2.read(&mut read_vec).is_ok());
assert_eq!(read_vec, b"hello");
}
#[test]
fn socket_basic() {
let (s1, s2) = Socket::create_stream();
// Write two packets and read from other end
assert_eq!(s1.write(b"hello").unwrap(), 5);
assert_eq!(s1.write(b"world").unwrap(), 5);
let mut read_vec = vec![0; 11];
assert_eq!(s2.read(&mut read_vec).unwrap(), 10);
assert_eq!(&read_vec[0..10], b"helloworld");
// Try reading when there is nothing to read.
assert_eq!(s2.read(&mut read_vec), Err(Status::SHOULD_WAIT));
}
#[cfg(not(target_os = "fuchsia"))]
#[test]
fn object_type_is_correct() {
let (c1, c2) = Channel::create();
let (s1, s2) = Socket::create_stream();
assert_eq!(c1.into_handle().object_type(), ObjectType::CHANNEL);
assert_eq!(c2.into_handle().object_type(), ObjectType::CHANNEL);
assert_eq!(s1.into_handle().object_type(), ObjectType::SOCKET);
assert_eq!(s2.into_handle().object_type(), ObjectType::SOCKET);
}
#[cfg(not(target_os = "fuchsia"))]
#[test]
fn invalid_handle_is_not_dangling() {
let h = Handle::invalid();
assert!(!h.is_dangling());
}
#[cfg(not(target_os = "fuchsia"))]
#[test]
fn live_valid_handle_is_not_dangling() {
let (c1, c2) = Channel::create();
assert!(!c1.is_dangling());
assert!(!c2.is_dangling());
}
#[cfg(not(target_os = "fuchsia"))]
#[test]
fn closed_handle_is_dangling() {
let closed_handle = unsafe { mimic_closed_handle() };
assert!(closed_handle.is_dangling());
}
#[test]
fn handle_basic_info_success() {
let (c1, c2) = Channel::create();
let c1_info = c1.basic_info().unwrap();
let c2_info = c2.basic_info().unwrap();
assert_ne!(c1_info.koid, INVALID_KOID);
assert_ne!(c1_info.related_koid, INVALID_KOID);
assert_ne!(c1_info.koid, c1_info.related_koid);
assert_eq!(c1_info.related_koid, c2_info.koid);
assert_eq!(c1_info.koid, c2_info.related_koid);
assert_eq!(c1_info.rights, Rights::CHANNEL_DEFAULT);
assert_eq!(c1_info.object_type, ObjectType::CHANNEL);
assert_eq!(c1_info.reserved, 0);
}
#[test]
fn handle_basic_info_invalid() {
assert_eq!(Handle::invalid().basic_info().unwrap_err(), zx_status::Status::BAD_HANDLE);
// Note non-zero but invalid handles can't be tested because of a
// panic when the handle isn't found in with_handle().
}
#[test]
fn handle_replace_success() {
let (c1, c2) = Channel::create();
let c1_basic_info = c1.basic_info().unwrap();
let c2_basic_info = c2.basic_info().unwrap();
assert_eq!(c1_basic_info.rights, Rights::CHANNEL_DEFAULT);
let new_handle = c1.into_handle().replace(Rights::TRANSFER | Rights::WRITE).unwrap();
let new_c1_basic_info = new_handle.basic_info().unwrap();
assert_eq!(new_c1_basic_info.koid, c1_basic_info.koid);
assert_eq!(new_c1_basic_info.related_koid, c1_basic_info.related_koid);
assert_eq!(new_c1_basic_info.object_type, ObjectType::CHANNEL);
assert_eq!(new_c1_basic_info.rights, Rights::TRANSFER | Rights::WRITE);
let new_c2_basic_info = c2.basic_info().unwrap();
assert_eq!(new_c2_basic_info.koid, c2_basic_info.koid);
assert_eq!(new_c2_basic_info.related_koid, c2_basic_info.related_koid);
assert_eq!(new_c2_basic_info.object_type, c2_basic_info.object_type);
assert_eq!(new_c2_basic_info.rights, c2_basic_info.rights);
}
#[test]
fn handle_replace_invalid() {
assert_eq!(
Handle::invalid().replace(Rights::TRANSFER).unwrap_err(),
zx_status::Status::BAD_HANDLE
);
let closed_handle = unsafe { mimic_closed_handle() };
assert_eq!(
ManuallyDrop::into_inner(closed_handle).replace(Rights::TRANSFER).unwrap_err(),
zx_status::Status::BAD_HANDLE
);
}
#[test]
fn handle_replace_increasing_rights() {
let (c1, _) = Channel::create();
let orig_basic_info = c1.basic_info().unwrap();
assert_eq!(orig_basic_info.rights, Rights::CHANNEL_DEFAULT);
assert_eq!(
c1.into_handle().replace(Rights::DUPLICATE).unwrap_err(),
zx_status::Status::INVALID_ARGS
);
}
#[test]
fn handle_replace_same_rights() {
let (c1, _) = Channel::create();
let orig_basic_info = c1.basic_info().unwrap();
assert_eq!(orig_basic_info.rights, Rights::CHANNEL_DEFAULT);
let orig_raw = c1.raw_handle();
let new_handle = c1.into_handle().replace(Rights::SAME_RIGHTS).unwrap();
assert_eq!(new_handle.raw_handle(), orig_raw);
let new_basic_info = new_handle.basic_info().unwrap();
assert_eq!(new_basic_info.rights, Rights::CHANNEL_DEFAULT);
}
#[test]
fn await_user_signal() {
let (c1, _) = EventPair::create();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::USER_0);
let (waker, count) = futures_test::task::new_count_waker();
let mut ctx = Context::from_waker(&waker);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 0);
c1.signal_handle(Signals::empty(), Signals::USER_1).unwrap();
assert_eq!(count, 0);
c1.signal_handle(Signals::empty(), Signals::USER_0).unwrap();
assert_eq!(count, 1);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::USER_0)));
c1.signal_handle(Signals::USER_0, Signals::empty()).unwrap();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::USER_0);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 1);
c1.signal_handle(Signals::empty(), Signals::USER_0).unwrap();
assert_eq!(count, 2);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::USER_0)));
}
#[test]
fn await_user_signal_peer() {
let (c1, c2) = EventPair::create();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::USER_0);
let (waker, count) = futures_test::task::new_count_waker();
let mut ctx = Context::from_waker(&waker);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 0);
c2.signal_peer(Signals::empty(), Signals::USER_1).unwrap();
assert_eq!(count, 0);
c2.signal_peer(Signals::empty(), Signals::USER_0).unwrap();
assert_eq!(count, 1);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::USER_0)));
c2.signal_peer(Signals::USER_0, Signals::empty()).unwrap();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::USER_0);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 1);
c2.signal_peer(Signals::empty(), Signals::USER_0).unwrap();
assert_eq!(count, 2);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::USER_0)));
}
#[test]
fn await_close_signal() {
let (c1, c2) = EventPair::create();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::OBJECT_PEER_CLOSED);
let (waker, count) = futures_test::task::new_count_waker();
let mut ctx = Context::from_waker(&waker);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 0);
c1.signal_handle(Signals::empty(), Signals::USER_1).unwrap();
assert_eq!(count, 0);
c1.signal_handle(Signals::empty(), Signals::USER_0).unwrap();
assert_eq!(count, 0);
std::mem::drop(c2);
assert_eq!(count, 1);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_PEER_CLOSED)));
}
#[test]
fn user_signal_no_rights() {
let (c1, _) = EventPair::create();
let c1 = c1.into_handle().replace(Rights::EVENTPAIR_DEFAULT & !Rights::SIGNAL).unwrap();
assert_eq!(c1.signal_handle(Signals::empty(), Signals::USER_1), Err(Status::ACCESS_DENIED));
}
#[test]
fn user_signal_peer_no_rights() {
let (c1, _) = EventPair::create();
let c1 =
c1.into_handle().replace(Rights::EVENTPAIR_DEFAULT & !Rights::SIGNAL_PEER).unwrap();
let c1 = EventPair::from(c1);
assert_eq!(c1.signal_peer(Signals::empty(), Signals::USER_1), Err(Status::ACCESS_DENIED));
}
#[test]
fn kernel_signal_denied() {
let (c1, _) = EventPair::create();
assert_eq!(
c1.signal_handle(Signals::empty(), Signals::OBJECT_WRITABLE),
Err(Status::INVALID_ARGS)
);
}
#[test]
fn kernel_signal_peer_denied() {
let (c1, _) = EventPair::create();
assert_eq!(
c1.signal_peer(Signals::empty(), Signals::OBJECT_WRITABLE),
Err(Status::INVALID_ARGS)
);
}
#[test]
fn handles_always_writable() {
let mut ctx = futures_test::task::noop_context();
let (s1, s2) = Socket::create_stream();
let mut on_sig = on_signals::OnSignalsRef::new(&s1, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
let mut on_sig = on_signals::OnSignalsRef::new(&s2, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
let (s1, s2) = Socket::create_datagram();
let mut on_sig = on_signals::OnSignalsRef::new(&s1, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
let mut on_sig = on_signals::OnSignalsRef::new(&s2, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
let (c1, c2) = Channel::create();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
let mut on_sig = on_signals::OnSignalsRef::new(&c2, Signals::OBJECT_WRITABLE);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_WRITABLE)));
}
#[test]
fn read_signal() {
let (c1, c2) = Channel::create();
let mut on_sig = on_signals::OnSignalsRef::new(&c1, Signals::OBJECT_READABLE);
let (waker, count) = futures_test::task::new_count_waker();
let mut ctx = Context::from_waker(&waker);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 0);
assert_eq!(c2.write(b"abc", &mut []), Ok(()));
assert_eq!(count, 1);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::OBJECT_READABLE)));
}
#[test]
fn event_is_one_sided() {
let e = Event::create();
assert_eq!(e.object_type(), ObjectType::EVENT);
assert!(e.related().is_invalid());
assert_eq!(e.koid_pair().1, INVALID_KOID.0);
assert_eq!(e.basic_info().unwrap().related_koid, INVALID_KOID);
}
#[test]
fn event_replace_rights() {
let e = Event::create();
assert_eq!(e.basic_info().unwrap().rights, Rights::EVENT_DEFAULT);
let e = e.into_handle().replace(Rights::TRANSFER).unwrap();
assert_eq!(e.basic_info().unwrap().rights, Rights::TRANSFER);
}
#[test]
fn event_user_signal() {
let e = Event::create();
let mut on_sig = on_signals::OnSignalsRef::new(&e, Signals::USER_0);
let (waker, count) = futures_test::task::new_count_waker();
let mut ctx = Context::from_waker(&waker);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Pending);
assert_eq!(count, 0);
assert_eq!(e.signal_handle(Signals::empty(), Signals::USER_0), Ok(()));
assert_eq!(count, 1);
assert_eq!(on_sig.poll_unpin(&mut ctx), Poll::Ready(Ok(Signals::USER_0)));
}
#[test]
fn mix_one_sided_and_two_sided_handles() {
// Test non-sided, left-side, and right-side handles for odd/even koids.
let e1 = Event::create(); // odd
let (c1, c2) = Channel::create(); // (even, odd)
let e2 = Event::create(); // even
let (p1, p2) = EventPair::create(); // (odd, even)
let e1_info = e1.basic_info().unwrap();
let c1_info = c1.basic_info().unwrap();
let c2_info = c2.basic_info().unwrap();
let e2_info = e2.basic_info().unwrap();
let p1_info = p1.basic_info().unwrap();
let p2_info = p2.basic_info().unwrap();
assert_eq!(e1_info.object_type, ObjectType::EVENT);
assert_eq!(c1_info.object_type, ObjectType::CHANNEL);
assert_eq!(c2_info.object_type, ObjectType::CHANNEL);
assert_eq!(e2_info.object_type, ObjectType::EVENT);
assert_eq!(p1_info.object_type, ObjectType::EVENTPAIR);
assert_eq!(p2_info.object_type, ObjectType::EVENTPAIR);
assert_eq!(e1_info.related_koid, INVALID_KOID);
assert_eq!(c1_info.related_koid, c2_info.koid);
assert_eq!(c2_info.related_koid, c1_info.koid);
assert_eq!(e2_info.related_koid, INVALID_KOID);
assert_eq!(p1_info.related_koid, p2_info.koid);
assert_eq!(p2_info.related_koid, p1_info.koid);
assert_eq!(e1.koid_pair(), (e1_info.koid.0, INVALID_KOID.0));
assert_eq!(c1.koid_pair(), (c1_info.koid.0, c2_info.koid.0));
assert_eq!(c2.koid_pair(), (c2_info.koid.0, c1_info.koid.0));
assert_eq!(e2.koid_pair(), (e2_info.koid.0, INVALID_KOID.0));
assert_eq!(p1.koid_pair(), (p1_info.koid.0, p2_info.koid.0));
assert_eq!(p2.koid_pair(), (p2_info.koid.0, p1_info.koid.0));
assert!(e1.related().is_invalid());
assert_eq!(c1.related(), c2.as_handle_ref());
assert_eq!(c2.related(), c1.as_handle_ref());
assert!(e2.related().is_invalid());
assert_eq!(p1.related(), p2.as_handle_ref());
assert_eq!(p2.related(), p1.as_handle_ref());
}
}