diff options
Diffstat (limited to '')
29 files changed, 5636 insertions, 0 deletions
diff --git a/third_party/rust/audioipc2-client/.cargo-checksum.json b/third_party/rust/audioipc2-client/.cargo-checksum.json new file mode 100644 index 0000000000..c33c0e16a1 --- /dev/null +++ b/third_party/rust/audioipc2-client/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"Cargo.toml":"9a41bc278a815fdcbe8d07c1d32f5cedb17dec3375993351b942567b4a644f9e","cbindgen.toml":"fb6abe1671497f432a06e40b1db7ed7cd2cceecbd9a2382193ad7534e8855e34","src/context.rs":"93dc43233603409dbb0c5de20c4e838c88bb0711df65cd5afc882f5b5e39f659","src/lib.rs":"c4a6797734489280f6b97dd72c9e51a7bd7be4104592eece3929e29d45cbca4a","src/send_recv.rs":"859abe75b521eb4297c84b30423814b5b87f3c7741ad16fe72189212e123e1ac","src/stream.rs":"3101f4052e35c1600adafedc67aed92d7d5531ff0ff4845de0a5ea3cedef872d"},"package":null}
\ No newline at end of file diff --git a/third_party/rust/audioipc2-client/Cargo.toml b/third_party/rust/audioipc2-client/Cargo.toml new file mode 100644 index 0000000000..5ff80d59d3 --- /dev/null +++ b/third_party/rust/audioipc2-client/Cargo.toml @@ -0,0 +1,34 @@ +# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO +# +# When uploading crates to the registry Cargo will automatically +# "normalize" Cargo.toml files for maximal compatibility +# with all versions of Cargo and also rewrite `path` dependencies +# to registry (e.g., crates.io) dependencies. +# +# If you are reading this file be aware that the original Cargo.toml +# will likely look very different (and much more reasonable). +# See Cargo.toml.orig for the original contents. + +[package] +edition = "2018" +name = "audioipc2-client" +version = "0.5.0" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>", +] +description = "Cubeb Backend for talking to remote cubeb server." +license = "ISC" + +[dependencies] +cubeb-backend = "0.10" +log = "0.4" + +[dependencies.audio_thread_priority] +version = "0.26.1" +features = ["winapi"] +default-features = false + +[dependencies.audioipc] +path = "../audioipc" +package = "audioipc2" diff --git a/third_party/rust/audioipc2-client/cbindgen.toml b/third_party/rust/audioipc2-client/cbindgen.toml new file mode 100644 index 0000000000..b3267889c6 --- /dev/null +++ b/third_party/rust/audioipc2-client/cbindgen.toml @@ -0,0 +1,28 @@ +header = """/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */""" +autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */""" +include_version = true +braces = "SameLine" +line_length = 100 +tab_width = 2 +language = "C++" +namespaces = ["mozilla", "audioipc2"] + +[export] +item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"] + +[parse] +parse_deps = true +include = ["audioipc2"] + +[fn] +args = "Vertical" +rename_args = "GeckoCase" + +[struct] +rename_fields = "GeckoCase" + +[defines] +"windows" = "XP_WIN" +"unix" = "XP_UNIX" diff --git a/third_party/rust/audioipc2-client/src/context.rs b/third_party/rust/audioipc2-client/src/context.rs new file mode 100644 index 0000000000..647d944724 --- /dev/null +++ b/third_party/rust/audioipc2-client/src/context.rs @@ -0,0 +1,383 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::stream; +use crate::{assert_not_in_callback, run_in_callback}; +use crate::{ClientStream, AUDIOIPC_INIT_PARAMS}; +#[cfg(target_os = "linux")] +use audio_thread_priority::get_current_thread_info; +#[cfg(not(target_os = "linux"))] +use audio_thread_priority::promote_current_thread_to_real_time; +use audioipc::ipccore::EventLoopHandle; +use audioipc::{ipccore, rpccore, sys, PlatformHandle}; +use audioipc::{ + messages, messages::DeviceCollectionReq, messages::DeviceCollectionResp, ClientMessage, + ServerMessage, +}; +use cubeb_backend::{ + capi_new, ffi, Context, ContextOps, DeviceCollectionRef, DeviceId, DeviceType, Error, Ops, + Result, Stream, StreamParams, StreamParamsRef, +}; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::{fmt, mem, ptr}; + +struct CubebClient; + +impl rpccore::Client for CubebClient { + type ServerMessage = ServerMessage; + type ClientMessage = ClientMessage; +} + +pub const CLIENT_OPS: Ops = capi_new!(ClientContext, ClientStream); + +// ClientContext's layout *must* match cubeb.c's `struct cubeb` for the +// common fields. +#[repr(C)] +pub struct ClientContext { + _ops: *const Ops, + rpc: rpccore::Proxy<ServerMessage, ClientMessage>, + rpc_thread: ipccore::EventLoopThread, + callback_thread: ipccore::EventLoopThread, + backend_id: CString, + device_collection_rpc: bool, + input_device_callback: Arc<Mutex<DeviceCollectionCallback>>, + output_device_callback: Arc<Mutex<DeviceCollectionCallback>>, +} + +impl ClientContext { + #[doc(hidden)] + pub fn rpc_handle(&self) -> &EventLoopHandle { + self.rpc_thread.handle() + } + + #[doc(hidden)] + pub fn rpc(&self) -> rpccore::Proxy<ServerMessage, ClientMessage> { + self.rpc.clone() + } + + #[doc(hidden)] + pub fn callback_handle(&self) -> &EventLoopHandle { + self.callback_thread.handle() + } +} + +#[cfg(target_os = "linux")] +fn promote_thread(rpc: &rpccore::Proxy<ServerMessage, ClientMessage>) { + match get_current_thread_info() { + Ok(info) => { + let bytes = info.serialize(); + let _ = rpc.call(ServerMessage::PromoteThreadToRealTime(bytes)); + } + Err(_) => { + warn!("Could not remotely promote thread to RT."); + } + } +} + +#[cfg(not(target_os = "linux"))] +fn promote_thread(_rpc: &rpccore::Proxy<ServerMessage, ClientMessage>) { + match promote_current_thread_to_real_time(0, 48000) { + Ok(_) => { + info!("Audio thread promoted to real-time."); + } + Err(_) => { + warn!("Could not promote thread to real-time."); + } + } +} + +fn register_thread(callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>) { + if let Some(func) = callback { + let thr = thread::current(); + let name = CString::new(thr.name().unwrap()).unwrap(); + func(name.as_ptr()); + } +} + +fn unregister_thread(callback: Option<extern "C" fn()>) { + if let Some(func) = callback { + func(); + } +} + +fn promote_and_register_thread( + rpc: &rpccore::Proxy<ServerMessage, ClientMessage>, + callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>, +) { + promote_thread(rpc); + register_thread(callback); +} + +#[derive(Default)] +struct DeviceCollectionCallback { + cb: ffi::cubeb_device_collection_changed_callback, + user_ptr: usize, +} + +struct DeviceCollectionServer { + input_device_callback: Arc<Mutex<DeviceCollectionCallback>>, + output_device_callback: Arc<Mutex<DeviceCollectionCallback>>, +} + +impl rpccore::Server for DeviceCollectionServer { + type ServerMessage = DeviceCollectionReq; + type ClientMessage = DeviceCollectionResp; + + fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage { + match req { + DeviceCollectionReq::DeviceChange(device_type) => { + trace!( + "ctx_thread: DeviceChange Callback: device_type={}", + device_type + ); + + let devtype = cubeb_backend::DeviceType::from_bits_truncate(device_type); + + let (input_cb, input_user_ptr) = { + let dcb = self.input_device_callback.lock().unwrap(); + (dcb.cb, dcb.user_ptr) + }; + let (output_cb, output_user_ptr) = { + let dcb = self.output_device_callback.lock().unwrap(); + (dcb.cb, dcb.user_ptr) + }; + + run_in_callback(|| { + if devtype.contains(cubeb_backend::DeviceType::INPUT) { + unsafe { input_cb.unwrap()(ptr::null_mut(), input_user_ptr as *mut c_void) } + } + if devtype.contains(cubeb_backend::DeviceType::OUTPUT) { + unsafe { + output_cb.unwrap()(ptr::null_mut(), output_user_ptr as *mut c_void) + } + } + }); + + DeviceCollectionResp::DeviceChange + } + } + } +} + +impl ContextOps for ClientContext { + fn init(_context_name: Option<&CStr>) -> Result<Context> { + assert_not_in_callback(); + + let params = AUDIOIPC_INIT_PARAMS.with(|p| p.replace(None).unwrap()); + let thread_create_callback = params.thread_create_callback; + let thread_destroy_callback = params.thread_destroy_callback; + + let server_connection = + unsafe { sys::Pipe::from_raw_handle(PlatformHandle::new(params.server_connection)) }; + + let rpc_thread = ipccore::EventLoopThread::new( + "AudioIPC Client RPC".to_string(), + None, + move || register_thread(thread_create_callback), + move || unregister_thread(thread_destroy_callback), + ) + .map_err(|_| Error::default())?; + let rpc = rpc_thread + .handle() + .bind_client::<CubebClient>(server_connection) + .map_err(|_| Error::default())?; + let rpc2 = rpc.clone(); + + // Don't let errors bubble from here. Later calls against this context + // will return errors the caller expects to handle. + let _ = send_recv!(rpc, ClientConnect(std::process::id()) => ClientConnected); + + let backend_id = send_recv!(rpc, ContextGetBackendId => ContextBackendId()) + .unwrap_or_else(|_| "(remote error)".to_string()); + let backend_id = CString::new(backend_id).expect("backend_id query failed"); + + // TODO: remove params.pool_size from init params. + let callback_thread = ipccore::EventLoopThread::new( + "AudioIPC Client Callback".to_string(), + Some(params.stack_size), + move || promote_and_register_thread(&rpc2, thread_create_callback), + move || unregister_thread(thread_destroy_callback), + ) + .map_err(|_| Error::default())?; + + let ctx = Box::new(ClientContext { + _ops: &CLIENT_OPS as *const _, + rpc, + rpc_thread, + callback_thread, + backend_id, + device_collection_rpc: false, + input_device_callback: Arc::new(Mutex::new(Default::default())), + output_device_callback: Arc::new(Mutex::new(Default::default())), + }); + Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) }) + } + + fn backend_id(&mut self) -> &CStr { + assert_not_in_callback(); + self.backend_id.as_c_str() + } + + fn max_channel_count(&mut self) -> Result<u32> { + assert_not_in_callback(); + send_recv!(self.rpc(), ContextGetMaxChannelCount => ContextMaxChannelCount()) + } + + fn min_latency(&mut self, params: StreamParams) -> Result<u32> { + assert_not_in_callback(); + let params = messages::StreamParams::from(params.as_ref()); + send_recv!(self.rpc(), ContextGetMinLatency(params) => ContextMinLatency()) + } + + fn preferred_sample_rate(&mut self) -> Result<u32> { + assert_not_in_callback(); + send_recv!(self.rpc(), ContextGetPreferredSampleRate => ContextPreferredSampleRate()) + } + + fn enumerate_devices( + &mut self, + devtype: DeviceType, + collection: &DeviceCollectionRef, + ) -> Result<()> { + assert_not_in_callback(); + let v: Vec<ffi::cubeb_device_info> = send_recv!( + self.rpc(), ContextGetDeviceEnumeration(devtype.bits()) => ContextEnumeratedDevices())? + .into_iter() + .map(|i| i.into()) + .collect(); + let mut vs = v.into_boxed_slice(); + let coll = unsafe { &mut *collection.as_ptr() }; + coll.device = vs.as_mut_ptr(); + coll.count = vs.len(); + // Giving away the memory owned by vs. Don't free it! + // Reclaimed in `device_collection_destroy`. + mem::forget(vs); + Ok(()) + } + + fn device_collection_destroy(&mut self, collection: &mut DeviceCollectionRef) -> Result<()> { + assert_not_in_callback(); + unsafe { + let coll = &mut *collection.as_ptr(); + let mut devices = Vec::from_raw_parts( + coll.device as *mut ffi::cubeb_device_info, + coll.count, + coll.count, + ); + for dev in &mut devices { + if !dev.device_id.is_null() { + let _ = CString::from_raw(dev.device_id as *mut _); + } + if !dev.group_id.is_null() { + let _ = CString::from_raw(dev.group_id as *mut _); + } + if !dev.vendor_name.is_null() { + let _ = CString::from_raw(dev.vendor_name as *mut _); + } + if !dev.friendly_name.is_null() { + let _ = CString::from_raw(dev.friendly_name as *mut _); + } + } + coll.device = ptr::null_mut(); + coll.count = 0; + Ok(()) + } + } + + fn stream_init( + &mut self, + stream_name: Option<&CStr>, + input_device: DeviceId, + input_stream_params: Option<&StreamParamsRef>, + output_device: DeviceId, + output_stream_params: Option<&StreamParamsRef>, + latency_frames: u32, + // These params aren't sent to the server + data_callback: ffi::cubeb_data_callback, + state_callback: ffi::cubeb_state_callback, + user_ptr: *mut c_void, + ) -> Result<Stream> { + assert_not_in_callback(); + + let stream_name = stream_name.map(|name| name.to_bytes_with_nul().to_vec()); + + let input_stream_params = input_stream_params.map(messages::StreamParams::from); + let output_stream_params = output_stream_params.map(messages::StreamParams::from); + + let init_params = messages::StreamInitParams { + stream_name, + input_device: input_device as usize, + input_stream_params, + output_device: output_device as usize, + output_stream_params, + latency_frames, + }; + stream::init(self, init_params, data_callback, state_callback, user_ptr) + } + + fn register_device_collection_changed( + &mut self, + devtype: DeviceType, + collection_changed_callback: ffi::cubeb_device_collection_changed_callback, + user_ptr: *mut c_void, + ) -> Result<()> { + assert_not_in_callback(); + + if !self.device_collection_rpc { + let mut fd = send_recv!(self.rpc(), + ContextSetupDeviceCollectionCallback => + ContextSetupDeviceCollectionCallback())?; + + let stream = unsafe { sys::Pipe::from_raw_handle(fd.platform_handle.take_handle()) }; + + let server = DeviceCollectionServer { + input_device_callback: self.input_device_callback.clone(), + output_device_callback: self.output_device_callback.clone(), + }; + + self.rpc_handle() + .bind_server(server, stream) + .map_err(|_| Error::default())?; + self.device_collection_rpc = true; + } + + if devtype.contains(cubeb_backend::DeviceType::INPUT) { + let mut cb = self.input_device_callback.lock().unwrap(); + cb.cb = collection_changed_callback; + cb.user_ptr = user_ptr as usize; + } + if devtype.contains(cubeb_backend::DeviceType::OUTPUT) { + let mut cb = self.output_device_callback.lock().unwrap(); + cb.cb = collection_changed_callback; + cb.user_ptr = user_ptr as usize; + } + + let enable = collection_changed_callback.is_some(); + send_recv!(self.rpc(), + ContextRegisterDeviceCollectionChanged(devtype.bits(), enable) => + ContextRegisteredDeviceCollectionChanged) + } +} + +impl Drop for ClientContext { + fn drop(&mut self) { + debug!("ClientContext dropped..."); + let _ = send_recv!(self.rpc(), ClientDisconnect => ClientDisconnected); + } +} + +impl fmt::Debug for ClientContext { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ClientContext") + .field("_ops", &self._ops) + .field("rpc", &self.rpc) + .field("core", &self.rpc_thread) + .field("cpu_pool", &"...") + .finish() + } +} diff --git a/third_party/rust/audioipc2-client/src/lib.rs b/third_party/rust/audioipc2-client/src/lib.rs new file mode 100644 index 0000000000..51a68af709 --- /dev/null +++ b/third_party/rust/audioipc2-client/src/lib.rs @@ -0,0 +1,81 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. +#![warn(unused_extern_crates)] + +#[macro_use] +extern crate log; + +#[macro_use] +mod send_recv; +mod context; +mod stream; + +use crate::context::ClientContext; +use crate::stream::ClientStream; +use audioipc::PlatformHandleType; +use cubeb_backend::{capi, ffi}; +use std::os::raw::{c_char, c_int}; + +thread_local!(static IN_CALLBACK: std::cell::RefCell<bool> = std::cell::RefCell::new(false)); +thread_local!(static AUDIOIPC_INIT_PARAMS: std::cell::RefCell<Option<AudioIpcInitParams>> = std::cell::RefCell::new(None)); + +#[repr(C)] +#[derive(Clone, Copy, Debug)] +pub struct AudioIpcInitParams { + // Fields only need to be public for ipctest. + pub server_connection: PlatformHandleType, + pub pool_size: usize, + pub stack_size: usize, + pub thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>, + pub thread_destroy_callback: Option<extern "C" fn()>, +} + +unsafe impl Send for AudioIpcInitParams {} + +fn set_in_callback(in_callback: bool) { + IN_CALLBACK.with(|b| { + assert_eq!(*b.borrow(), !in_callback); + *b.borrow_mut() = in_callback; + }); +} + +fn run_in_callback<F, R>(f: F) -> R +where + F: FnOnce() -> R, +{ + set_in_callback(true); + + let r = f(); + + set_in_callback(false); + + r +} + +fn assert_not_in_callback() { + IN_CALLBACK.with(|b| { + assert!(!*b.borrow()); + }); +} + +#[allow(clippy::missing_safety_doc)] +#[no_mangle] +/// Entry point from C code. +pub unsafe extern "C" fn audioipc2_client_init( + c: *mut *mut ffi::cubeb, + context_name: *const c_char, + init_params: *const AudioIpcInitParams, +) -> c_int { + if init_params.is_null() { + return cubeb_backend::ffi::CUBEB_ERROR; + } + + let init_params = &*init_params; + + AUDIOIPC_INIT_PARAMS.with(|p| { + *p.borrow_mut() = Some(*init_params); + }); + capi::capi_init::<ClientContext>(c, context_name) +} diff --git a/third_party/rust/audioipc2-client/src/send_recv.rs b/third_party/rust/audioipc2-client/src/send_recv.rs new file mode 100644 index 0000000000..1134c99a49 --- /dev/null +++ b/third_party/rust/audioipc2-client/src/send_recv.rs @@ -0,0 +1,73 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +use cubeb_backend::Error; +use std::os::raw::c_int; + +#[doc(hidden)] +pub fn _err<E>(e: E) -> Error +where + E: Into<Option<c_int>>, +{ + match e.into() { + Some(e) => Error::from_raw(e), + None => Error::error(), + } +} + +#[macro_export] +macro_rules! send_recv { + ($rpc:expr, $smsg:ident => $rmsg:ident) => {{ + let resp = send_recv!(__send $rpc, $smsg); + send_recv!(__recv resp, $rmsg) + }}; + ($rpc:expr, $smsg:ident => $rmsg:ident()) => {{ + let resp = send_recv!(__send $rpc, $smsg); + send_recv!(__recv resp, $rmsg __result) + }}; + ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident) => {{ + let resp = send_recv!(__send $rpc, $smsg, $($a),*); + send_recv!(__recv resp, $rmsg) + }}; + ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident()) => {{ + let resp = send_recv!(__send $rpc, $smsg, $($a),*); + send_recv!(__recv resp, $rmsg __result) + }}; + // + (__send $rpc:expr, $smsg:ident) => ({ + $rpc.call(ServerMessage::$smsg) + }); + (__send $rpc:expr, $smsg:ident, $($a:expr),*) => ({ + $rpc.call(ServerMessage::$smsg($($a),*)) + }); + (__recv $resp:expr, $rmsg:ident) => ({ + match $resp { + Ok(ClientMessage::$rmsg) => Ok(()), + Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)), + Ok(m) => { + debug!("received wrong message - got={:?}", m); + Err($crate::send_recv::_err(None)) + }, + Err(e) => { + debug!("received error from rpc - got={:?}", e); + Err($crate::send_recv::_err(None)) + }, + } + }); + (__recv $resp:expr, $rmsg:ident __result) => ({ + match $resp { + Ok(ClientMessage::$rmsg(v)) => Ok(v), + Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)), + Ok(m) => { + debug!("received wrong message - got={:?}", m); + Err($crate::send_recv::_err(None)) + }, + Err(e) => { + debug!("received error - got={:?}", e); + Err($crate::send_recv::_err(None)) + }, + } + }) +} diff --git a/third_party/rust/audioipc2-client/src/stream.rs b/third_party/rust/audioipc2-client/src/stream.rs new file mode 100644 index 0000000000..ab702bc98c --- /dev/null +++ b/third_party/rust/audioipc2-client/src/stream.rs @@ -0,0 +1,339 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::ClientContext; +use crate::{assert_not_in_callback, run_in_callback}; +use audioipc::messages::StreamCreateParams; +use audioipc::messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage}; +use audioipc::shm::SharedMem; +use audioipc::{rpccore, sys}; +use cubeb_backend::{ffi, DeviceRef, Error, Result, Stream, StreamOps}; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::ptr; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; + +pub struct Device(ffi::cubeb_device); + +impl Drop for Device { + fn drop(&mut self) { + unsafe { + if !self.0.input_name.is_null() { + let _ = CString::from_raw(self.0.input_name as *mut _); + } + if !self.0.output_name.is_null() { + let _ = CString::from_raw(self.0.output_name as *mut _); + } + } + } +} + +// ClientStream's layout *must* match cubeb.c's `struct cubeb_stream` for the +// common fields. +#[repr(C)] +#[derive(Debug)] +pub struct ClientStream<'ctx> { + // This must be a reference to Context for cubeb, cubeb accesses + // stream methods via stream->context->ops + context: &'ctx ClientContext, + user_ptr: *mut c_void, + token: usize, + device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>, + // Signals ClientStream that CallbackServer has dropped. + shutdown_rx: mpsc::Receiver<()>, +} + +struct CallbackServer { + shm: SharedMem, + duplex_input: Option<Vec<u8>>, + data_cb: ffi::cubeb_data_callback, + state_cb: ffi::cubeb_state_callback, + user_ptr: usize, + device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>, + // Signals ClientStream that CallbackServer has dropped. + _shutdown_tx: mpsc::Sender<()>, +} + +impl rpccore::Server for CallbackServer { + type ServerMessage = CallbackReq; + type ClientMessage = CallbackResp; + + fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage { + match req { + CallbackReq::Data { + nframes, + input_frame_size, + output_frame_size, + } => { + trace!( + "stream_thread: Data Callback: nframes={} input_fs={} output_fs={}", + nframes, + input_frame_size, + output_frame_size, + ); + + let input_nbytes = nframes as usize * input_frame_size; + let output_nbytes = nframes as usize * output_frame_size; + + // Input and output reuse the same shmem backing. Unfortunately, cubeb's data_callback isn't + // specified in such a way that would require the callee to consume all of the input before + // writing to the output (i.e., it is passed as two pointers that aren't expected to alias). + // That means we need to copy the input here. + if let Some(buf) = &mut self.duplex_input { + assert!(input_nbytes > 0); + assert!(buf.capacity() >= input_nbytes); + unsafe { + let input = self.shm.get_slice(input_nbytes).unwrap(); + ptr::copy_nonoverlapping(input.as_ptr(), buf.as_mut_ptr(), input.len()); + } + } + + run_in_callback(|| { + let nframes = unsafe { + let input_ptr = if input_frame_size > 0 { + if let Some(buf) = &mut self.duplex_input { + buf.as_ptr() + } else { + self.shm.get_slice(input_nbytes).unwrap().as_ptr() + } + } else { + ptr::null() + }; + let output_ptr = if output_frame_size > 0 { + self.shm.get_mut_slice(output_nbytes).unwrap().as_mut_ptr() + } else { + ptr::null_mut() + }; + + self.data_cb.unwrap()( + ptr::null_mut(), // https://github.com/kinetiknz/cubeb/issues/518 + self.user_ptr as *mut c_void, + input_ptr as *const _, + output_ptr as *mut _, + nframes as _, + ) + }; + + CallbackResp::Data(nframes as isize) + }) + } + CallbackReq::State(state) => { + trace!("stream_thread: State Callback: {:?}", state); + run_in_callback(|| unsafe { + self.state_cb.unwrap()(ptr::null_mut(), self.user_ptr as *mut _, state); + }); + + CallbackResp::State + } + CallbackReq::DeviceChange => { + run_in_callback(|| { + let cb = *self.device_change_cb.lock().unwrap(); + if let Some(cb) = cb { + unsafe { + cb(self.user_ptr as *mut _); + } + } else { + warn!("DeviceChange received with null callback"); + } + }); + + CallbackResp::DeviceChange + } + } + } +} + +impl<'ctx> ClientStream<'ctx> { + fn init( + ctx: &'ctx ClientContext, + init_params: messages::StreamInitParams, + data_callback: ffi::cubeb_data_callback, + state_callback: ffi::cubeb_state_callback, + user_ptr: *mut c_void, + ) -> Result<Stream> { + assert_not_in_callback(); + + let rpc = ctx.rpc(); + let create_params = StreamCreateParams { + input_stream_params: init_params.input_stream_params, + output_stream_params: init_params.output_stream_params, + }; + let mut data = send_recv!(rpc, StreamCreate(create_params) => StreamCreated())?; + + debug!( + "token = {}, handle = {:?} area_size = {:?}", + data.token, data.shm_handle, data.shm_area_size + ); + + let shm = + match unsafe { SharedMem::from(data.shm_handle.take_handle(), data.shm_area_size) } { + Ok(shm) => shm, + Err(e) => { + warn!( + "SharedMem client mapping failed (size={}, err={:?})", + data.shm_area_size, e + ); + return Err(Error::default()); + } + }; + + let duplex_input = if let (Some(_), Some(_)) = ( + init_params.input_stream_params, + init_params.output_stream_params, + ) { + let mut duplex_input = Vec::new(); + match duplex_input.try_reserve_exact(data.shm_area_size) { + Ok(()) => Some(duplex_input), + Err(e) => { + warn!( + "duplex_input allocation failed (size={}, err={:?})", + data.shm_area_size, e + ); + return Err(Error::default()); + } + } + } else { + None + }; + + let mut stream = + send_recv!(rpc, StreamInit(data.token, init_params) => StreamInitialized())?; + let stream = unsafe { sys::Pipe::from_raw_handle(stream.take_handle()) }; + + let user_data = user_ptr as usize; + + let null_cb: ffi::cubeb_device_changed_callback = None; + let device_change_cb = Arc::new(Mutex::new(null_cb)); + + let (_shutdown_tx, shutdown_rx) = mpsc::channel(); + + let server = CallbackServer { + shm, + duplex_input, + data_cb: data_callback, + state_cb: state_callback, + user_ptr: user_data, + device_change_cb: device_change_cb.clone(), + _shutdown_tx, + }; + + ctx.callback_handle() + .bind_server(server, stream) + .map_err(|_| Error::default())?; + + let stream = Box::into_raw(Box::new(ClientStream { + context: ctx, + user_ptr, + token: data.token, + device_change_cb, + shutdown_rx, + })); + Ok(unsafe { Stream::from_ptr(stream as *mut _) }) + } +} + +impl Drop for ClientStream<'_> { + fn drop(&mut self) { + debug!("ClientStream drop"); + let _ = send_recv!(self.context.rpc(), StreamDestroy(self.token) => StreamDestroyed); + debug!("ClientStream drop - stream destroyed"); + // Wait for CallbackServer to shutdown. The remote server drops the RPC + // connection during StreamDestroy, which will cause CallbackServer to drop + // once the connection close is detected. Dropping CallbackServer will + // cause the shutdown channel to error on recv, which we rely on to + // synchronize with CallbackServer dropping. + let _ = self.shutdown_rx.recv(); + debug!("ClientStream dropped"); + } +} + +impl StreamOps for ClientStream<'_> { + fn start(&mut self) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamStart(self.token) => StreamStarted) + } + + fn stop(&mut self) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamStop(self.token) => StreamStopped) + } + + fn position(&mut self) -> Result<u64> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamGetPosition(self.token) => StreamPosition()) + } + + fn latency(&mut self) -> Result<u32> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamGetLatency(self.token) => StreamLatency()) + } + + fn input_latency(&mut self) -> Result<u32> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamGetInputLatency(self.token) => StreamInputLatency()) + } + + fn set_volume(&mut self, volume: f32) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamSetVolume(self.token, volume) => StreamVolumeSet) + } + + fn set_name(&mut self, name: &CStr) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamSetName(self.token, name.to_owned()) => StreamNameSet) + } + + fn current_device(&mut self) -> Result<&DeviceRef> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + match send_recv!(rpc, StreamGetCurrentDevice(self.token) => StreamCurrentDevice()) { + Ok(d) => Ok(unsafe { DeviceRef::from_ptr(Box::into_raw(Box::new(d.into()))) }), + Err(e) => Err(e), + } + } + + fn device_destroy(&mut self, device: &DeviceRef) -> Result<()> { + assert_not_in_callback(); + if device.as_ptr().is_null() { + Err(Error::error()) + } else { + unsafe { + let _: Box<Device> = Box::from_raw(device.as_ptr() as *mut _); + } + Ok(()) + } + } + + fn register_device_changed_callback( + &mut self, + device_changed_callback: ffi::cubeb_device_changed_callback, + ) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + let enable = device_changed_callback.is_some(); + *self.device_change_cb.lock().unwrap() = device_changed_callback; + send_recv!(rpc, StreamRegisterDeviceChangeCallback(self.token, enable) => StreamRegisterDeviceChangeCallback) + } +} + +pub fn init( + ctx: &ClientContext, + init_params: messages::StreamInitParams, + data_callback: ffi::cubeb_data_callback, + state_callback: ffi::cubeb_state_callback, + user_ptr: *mut c_void, +) -> Result<Stream> { + let stm = ClientStream::init(ctx, init_params, data_callback, state_callback, user_ptr)?; + debug_assert_eq!(stm.user_ptr(), user_ptr); + Ok(stm) +} diff --git a/third_party/rust/audioipc2-server/.cargo-checksum.json b/third_party/rust/audioipc2-server/.cargo-checksum.json new file mode 100644 index 0000000000..733e618bb0 --- /dev/null +++ b/third_party/rust/audioipc2-server/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"Cargo.toml":"4b00cd5b0ef67640c29a5b4c27ff0eb2bda60ea9bfaa0321625e2eb48714efa8","cbindgen.toml":"fb6abe1671497f432a06e40b1db7ed7cd2cceecbd9a2382193ad7534e8855e34","src/lib.rs":"06aff4fd1326aeabb16b01f81a6f3c59c1717ebe96285a063724830cdf30303a","src/server.rs":"ea839fe4607ba6b70a1ef5dfb2305eb668b148820c5590b87192609fbe3c9edd"},"package":null}
\ No newline at end of file diff --git a/third_party/rust/audioipc2-server/Cargo.toml b/third_party/rust/audioipc2-server/Cargo.toml new file mode 100644 index 0000000000..dbe87893a6 --- /dev/null +++ b/third_party/rust/audioipc2-server/Cargo.toml @@ -0,0 +1,40 @@ +# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO +# +# When uploading crates to the registry Cargo will automatically +# "normalize" Cargo.toml files for maximal compatibility +# with all versions of Cargo and also rewrite `path` dependencies +# to registry (e.g., crates.io) dependencies. +# +# If you are reading this file be aware that the original Cargo.toml +# will likely look very different (and much more reasonable). +# See Cargo.toml.orig for the original contents. + +[package] +edition = "2018" +name = "audioipc2-server" +version = "0.5.0" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>", +] +description = "Remote cubeb server" +license = "ISC" + +[dependencies] +cubeb-core = "0.10.0" +log = "0.4" +once_cell = "1.2.0" +slab = "0.4" + +[dependencies.audio_thread_priority] +version = "0.26.1" +features = ["winapi"] +default-features = false + +[dependencies.audioipc] +path = "../audioipc" +package = "audioipc2" + +[dependencies.error-chain] +version = "0.12.0" +default-features = false diff --git a/third_party/rust/audioipc2-server/cbindgen.toml b/third_party/rust/audioipc2-server/cbindgen.toml new file mode 100644 index 0000000000..b3267889c6 --- /dev/null +++ b/third_party/rust/audioipc2-server/cbindgen.toml @@ -0,0 +1,28 @@ +header = """/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */""" +autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */""" +include_version = true +braces = "SameLine" +line_length = 100 +tab_width = 2 +language = "C++" +namespaces = ["mozilla", "audioipc2"] + +[export] +item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"] + +[parse] +parse_deps = true +include = ["audioipc2"] + +[fn] +args = "Vertical" +rename_args = "GeckoCase" + +[struct] +rename_fields = "GeckoCase" + +[defines] +"windows" = "XP_WIN" +"unix" = "XP_UNIX" diff --git a/third_party/rust/audioipc2-server/src/lib.rs b/third_party/rust/audioipc2-server/src/lib.rs new file mode 100644 index 0000000000..c90de38c66 --- /dev/null +++ b/third_party/rust/audioipc2-server/src/lib.rs @@ -0,0 +1,221 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details +#![warn(unused_extern_crates)] + +#[macro_use] +extern crate error_chain; +#[macro_use] +extern crate log; + +use audio_thread_priority::promote_current_thread_to_real_time; +use audioipc::ipccore; +use audioipc::sys; +use audioipc::PlatformHandleType; +use once_cell::sync::Lazy; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::ptr; +use std::sync::Mutex; +use std::thread; + +mod server; + +struct CubebContextParams { + context_name: CString, + backend_name: Option<CString>, +} + +static G_CUBEB_CONTEXT_PARAMS: Lazy<Mutex<CubebContextParams>> = Lazy::new(|| { + Mutex::new(CubebContextParams { + context_name: CString::new("AudioIPC Server").unwrap(), + backend_name: None, + }) +}); + +#[allow(deprecated)] +pub mod errors { + #![allow(clippy::upper_case_acronyms)] + error_chain! { + links { + AudioIPC(::audioipc::errors::Error, ::audioipc::errors::ErrorKind); + } + foreign_links { + Cubeb(cubeb_core::Error); + Io(::std::io::Error); + } + } +} + +use crate::errors::*; + +struct ServerWrapper { + rpc_thread: ipccore::EventLoopThread, + callback_thread: ipccore::EventLoopThread, + device_collection_thread: ipccore::EventLoopThread, +} + +fn register_thread(callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>) { + if let Some(func) = callback { + let name = CString::new(thread::current().name().unwrap()).unwrap(); + func(name.as_ptr()); + } +} + +fn unregister_thread(callback: Option<extern "C" fn()>) { + if let Some(func) = callback { + func(); + } +} + +fn init_threads( + thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>, + thread_destroy_callback: Option<extern "C" fn()>, +) -> Result<ServerWrapper> { + let rpc_name = "AudioIPC Server RPC"; + let rpc_thread = ipccore::EventLoopThread::new( + rpc_name.to_string(), + None, + move || { + trace!("Starting {} thread", rpc_name); + register_thread(thread_create_callback); + audioipc::server_platform_init(); + }, + move || { + unregister_thread(thread_destroy_callback); + trace!("Stopping {} thread", rpc_name); + }, + ) + .map_err(|e| { + debug!("Failed to start {} thread: {:?}", rpc_name, e); + e + })?; + + let callback_name = "AudioIPC Server Callback"; + let callback_thread = ipccore::EventLoopThread::new( + callback_name.to_string(), + None, + move || { + trace!("Starting {} thread", callback_name); + if let Err(e) = promote_current_thread_to_real_time(0, 48000) { + debug!( + "Failed to promote {} thread to real-time: {:?}", + callback_name, e + ); + } + register_thread(thread_create_callback); + }, + move || { + unregister_thread(thread_destroy_callback); + trace!("Stopping {} thread", callback_name); + }, + ) + .map_err(|e| { + debug!("Failed to start {} thread: {:?}", callback_name, e); + e + })?; + + let device_collection_name = "AudioIPC DeviceCollection RPC"; + let device_collection_thread = ipccore::EventLoopThread::new( + device_collection_name.to_string(), + None, + move || { + trace!("Starting {} thread", device_collection_name); + register_thread(thread_create_callback); + }, + move || { + unregister_thread(thread_destroy_callback); + trace!("Stopping {} thread", device_collection_name); + }, + ) + .map_err(|e| { + debug!("Failed to start {} thread: {:?}", device_collection_name, e); + e + })?; + + Ok(ServerWrapper { + rpc_thread, + callback_thread, + device_collection_thread, + }) +} + +#[repr(C)] +#[derive(Clone, Copy, Debug)] +pub struct AudioIpcServerInitParams { + // Fields only need to be public for ipctest. + pub thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>, + pub thread_destroy_callback: Option<extern "C" fn()>, +} + +#[allow(clippy::missing_safety_doc)] +#[no_mangle] +pub unsafe extern "C" fn audioipc2_server_start( + context_name: *const std::os::raw::c_char, + backend_name: *const std::os::raw::c_char, + init_params: *const AudioIpcServerInitParams, +) -> *mut c_void { + assert!(!init_params.is_null()); + let mut params = G_CUBEB_CONTEXT_PARAMS.lock().unwrap(); + if !context_name.is_null() { + params.context_name = CStr::from_ptr(context_name).to_owned(); + } + if !backend_name.is_null() { + let backend_string = CStr::from_ptr(backend_name).to_owned(); + params.backend_name = Some(backend_string); + } + match init_threads( + (*init_params).thread_create_callback, + (*init_params).thread_destroy_callback, + ) { + Ok(server) => Box::into_raw(Box::new(server)) as *mut _, + Err(_) => ptr::null_mut() as *mut _, + } +} + +// A `shm_area_size` of 0 allows the server to calculate an appropriate shm size for each stream. +// A non-zero `shm_area_size` forces all allocations to the specified size. +#[no_mangle] +pub extern "C" fn audioipc2_server_new_client( + p: *mut c_void, + shm_area_size: usize, +) -> PlatformHandleType { + let wrapper: &ServerWrapper = unsafe { &*(p as *mut _) }; + + // We create a connected pair of anonymous IPC endpoints. One side + // is registered with the event loop core, the other side is returned + // to the caller to be remoted to the client to complete setup. + let (server_pipe, client_pipe) = match sys::make_pipe_pair() { + Ok((server_pipe, client_pipe)) => (server_pipe, client_pipe), + Err(e) => { + error!( + "audioipc_server_new_client - make_pipe_pair failed: {:?}", + e + ); + return audioipc::INVALID_HANDLE_VALUE; + } + }; + + let rpc_thread = wrapper.rpc_thread.handle(); + let callback_thread = wrapper.callback_thread.handle(); + let device_collection_thread = wrapper.device_collection_thread.handle(); + + let server = server::CubebServer::new( + callback_thread.clone(), + device_collection_thread.clone(), + shm_area_size, + ); + if let Err(e) = rpc_thread.bind_server(server, server_pipe) { + error!("audioipc_server_new_client - bind_server failed: {:?}", e); + return audioipc::INVALID_HANDLE_VALUE; + } + + unsafe { client_pipe.into_raw() } +} + +#[no_mangle] +pub extern "C" fn audioipc2_server_stop(p: *mut c_void) { + let wrapper = unsafe { Box::<ServerWrapper>::from_raw(p as *mut _) }; + drop(wrapper); +} diff --git a/third_party/rust/audioipc2-server/src/server.rs b/third_party/rust/audioipc2-server/src/server.rs new file mode 100644 index 0000000000..0094a17681 --- /dev/null +++ b/third_party/rust/audioipc2-server/src/server.rs @@ -0,0 +1,921 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +#[cfg(target_os = "linux")] +use audio_thread_priority::{promote_thread_to_real_time, RtPriorityThreadInfo}; +use audioipc::messages::SerializableHandle; +use audioipc::messages::{ + CallbackReq, CallbackResp, ClientMessage, Device, DeviceCollectionReq, DeviceCollectionResp, + DeviceInfo, RegisterDeviceCollectionChanged, ServerMessage, StreamCreate, StreamCreateParams, + StreamInitParams, StreamParams, +}; +use audioipc::shm::SharedMem; +use audioipc::{ipccore, rpccore, sys, PlatformHandle}; +use cubeb_core as cubeb; +use cubeb_core::ffi; +use std::convert::{From, TryInto}; +use std::ffi::CStr; +use std::mem::size_of; +use std::os::raw::{c_long, c_void}; +use std::rc::Rc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::{cell::RefCell, sync::Mutex}; +use std::{panic, slice}; + +use crate::errors::*; + +fn error(error: cubeb::Error) -> ClientMessage { + ClientMessage::Error(error.raw_code()) +} + +struct CubebDeviceCollectionManager { + servers: Mutex<Vec<(Rc<DeviceCollectionChangeCallback>, cubeb::DeviceType)>>, +} + +impl CubebDeviceCollectionManager { + fn new() -> CubebDeviceCollectionManager { + CubebDeviceCollectionManager { + servers: Mutex::new(Vec::new()), + } + } + + fn register( + &self, + context: &cubeb::Context, + server: &Rc<DeviceCollectionChangeCallback>, + devtype: cubeb::DeviceType, + ) -> cubeb::Result<()> { + let mut servers = self.servers.lock().unwrap(); + if servers.is_empty() { + self.internal_register(context, true)?; + } + servers.push((server.clone(), devtype)); + Ok(()) + } + + fn unregister( + &self, + context: &cubeb::Context, + server: &Rc<DeviceCollectionChangeCallback>, + devtype: cubeb::DeviceType, + ) -> cubeb::Result<()> { + let mut servers = self.servers.lock().unwrap(); + servers.retain(|(s, d)| !Rc::ptr_eq(s, server) || d != &devtype); + if servers.is_empty() { + self.internal_register(context, false)?; + } + Ok(()) + } + + fn internal_register(&self, context: &cubeb::Context, enable: bool) -> cubeb::Result<()> { + for &(dir, cb) in &[ + ( + cubeb::DeviceType::INPUT, + device_collection_changed_input_cb_c as _, + ), + ( + cubeb::DeviceType::OUTPUT, + device_collection_changed_output_cb_c as _, + ), + ] { + unsafe { + context.register_device_collection_changed( + dir, + if enable { Some(cb) } else { None }, + if enable { + self as *const CubebDeviceCollectionManager as *mut c_void + } else { + std::ptr::null_mut() + }, + )?; + } + } + Ok(()) + } + + unsafe fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) { + let servers = self.servers.lock().unwrap(); + servers.iter().for_each(|(s, d)| { + if d.contains(cubeb::DeviceType::from_bits_truncate(device_type)) { + s.device_collection_changed_callback(device_type) + } + }); + } +} + +impl Drop for CubebDeviceCollectionManager { + fn drop(&mut self) { + assert!(self.servers.lock().unwrap().is_empty()); + } +} + +struct DevIdMap { + devices: Vec<usize>, +} + +// A cubeb_devid is an opaque type which may be implemented with a stable +// pointer in a cubeb backend. cubeb_devids received remotely must be +// validated before use, so DevIdMap provides a simple 1:1 mapping between a +// cubeb_devid and an IPC-transportable value suitable for use as a unique +// handle. +impl DevIdMap { + fn new() -> DevIdMap { + let mut d = DevIdMap { + devices: Vec::with_capacity(32), + }; + // A null cubeb_devid is used for selecting the default device. + // Pre-populate the mapping with 0 -> 0 to handle nulls. + d.devices.push(0); + d + } + + // Given a cubeb_devid, return a unique stable value suitable for use + // over IPC. + fn make_handle(&mut self, devid: usize) -> usize { + if let Some(i) = self.devices.iter().position(|&d| d == devid) { + return i; + } + self.devices.push(devid); + self.devices.len() - 1 + } + + // Given a handle produced by `make_handle`, return the associated + // cubeb_devid. Invalid handles result in a panic. + fn handle_to_id(&self, handle: usize) -> usize { + self.devices[handle] + } +} + +struct CubebContextState { + // `manager` must be dropped before the `context` is destroyed. + manager: CubebDeviceCollectionManager, + context: cubeb::Result<cubeb::Context>, +} + +thread_local!(static CONTEXT_KEY: RefCell<Option<CubebContextState>> = RefCell::new(None)); + +fn cubeb_init_from_context_params() -> cubeb::Result<cubeb::Context> { + let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap(); + let context_name = Some(params.context_name.as_c_str()); + let backend_name = params.backend_name.as_deref(); + let r = cubeb::Context::init(context_name, backend_name); + r.map_err(|e| { + info!("cubeb::Context::init failed r={:?}", e); + e + }) +} + +fn with_local_context<T, F>(f: F) -> T +where + F: FnOnce(&cubeb::Result<cubeb::Context>, &mut CubebDeviceCollectionManager) -> T, +{ + CONTEXT_KEY.with(|k| { + let mut state = k.borrow_mut(); + if state.is_none() { + *state = Some(CubebContextState { + manager: CubebDeviceCollectionManager::new(), + context: cubeb_init_from_context_params(), + }); + } + let CubebContextState { manager, context } = state.as_mut().unwrap(); + // Always reattempt to initialize cubeb, OS config may have changed. + if context.is_err() { + *context = cubeb_init_from_context_params(); + } + f(context, manager) + }) +} + +struct DeviceCollectionClient; + +impl rpccore::Client for DeviceCollectionClient { + type ServerMessage = DeviceCollectionReq; + type ClientMessage = DeviceCollectionResp; +} + +struct CallbackClient; + +impl rpccore::Client for CallbackClient { + type ServerMessage = CallbackReq; + type ClientMessage = CallbackResp; +} + +struct ServerStreamCallbacks { + /// Size of input frame in bytes + input_frame_size: u16, + /// Size of output frame in bytes + output_frame_size: u16, + /// Shared memory buffer for transporting audio data to/from client + shm: SharedMem, + /// RPC interface for data_callback (on OS audio thread) to server callback thread + data_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>, + /// RPC interface for state_callback (on any thread) to server callback thread + state_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>, + /// RPC interface for device_change_callback (on any thread) to server callback thread + device_change_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>, + /// Indicates stream is connected to client side. Callbacks received before + /// the stream is in the connected state cannot be sent to the client side, so + /// are logged and otherwise ignored. + connected: AtomicBool, +} + +impl ServerStreamCallbacks { + fn data_callback(&mut self, input: &[u8], output: &mut [u8], nframes: isize) -> isize { + trace!( + "Stream data callback: {} {} {}", + nframes, + input.len(), + output.len() + ); + if !self.connected.load(Ordering::Acquire) { + warn!("Stream data callback triggered before stream connected"); + return cubeb::ffi::CUBEB_ERROR.try_into().unwrap(); + } + + if self.input_frame_size != 0 { + if input.len() > self.shm.get_size() { + debug!( + "bad input size: input={} shm={}", + input.len(), + self.shm.get_size() + ); + return cubeb::ffi::CUBEB_ERROR.try_into().unwrap(); + } + unsafe { + self.shm + .get_mut_slice(input.len()) + .unwrap() + .copy_from_slice(input); + } + } + + if self.output_frame_size != 0 && output.len() > self.shm.get_size() { + debug!( + "bad output size: output={} shm={}", + output.len(), + self.shm.get_size() + ); + return cubeb::ffi::CUBEB_ERROR.try_into().unwrap(); + } + + let r = self.data_callback_rpc.call(CallbackReq::Data { + nframes, + input_frame_size: self.input_frame_size as usize, + output_frame_size: self.output_frame_size as usize, + }); + + match r { + Ok(CallbackResp::Data(frames)) => { + if frames >= 0 && self.output_frame_size != 0 { + let nbytes = frames as usize * self.output_frame_size as usize; + unsafe { + output[..nbytes].copy_from_slice(self.shm.get_slice(nbytes).unwrap()); + } + } + frames + } + _ => { + debug!("Unexpected message {:?} during data_callback", r); + cubeb::ffi::CUBEB_ERROR.try_into().unwrap() + } + } + } + + fn state_callback(&self, state: cubeb::State) { + trace!("Stream state callback: {:?}", state); + if !self.connected.load(Ordering::Acquire) { + warn!("Stream state callback triggered before stream connected"); + return; + } + + let r = self + .state_callback_rpc + .call(CallbackReq::State(state.into())); + match r { + Ok(CallbackResp::State) => {} + _ => { + debug!("Unexpected message {:?} during state callback", r); + } + } + } + + fn device_change_callback(&self) { + trace!("Stream device change callback"); + if !self.connected.load(Ordering::Acquire) { + warn!("Stream device_change callback triggered before stream connected"); + return; + } + let r = self + .device_change_callback_rpc + .call(CallbackReq::DeviceChange); + match r { + Ok(CallbackResp::DeviceChange) => {} + _ => { + debug!("Unexpected message {:?} during device change callback", r); + } + } + } +} + +static SHM_ID: AtomicUsize = AtomicUsize::new(0); + +// Generate a temporary shm_id fragment that is unique to the process. This +// path is used temporarily to create a shm segment, which is then +// immediately deleted from the filesystem while retaining handles to the +// shm to be shared between the server and client. +fn get_shm_id() -> String { + format!( + "cubeb-shm-{}-{}", + std::process::id(), + SHM_ID.fetch_add(1, Ordering::SeqCst) + ) +} + +struct ServerStream { + stream: Option<cubeb::Stream>, + cbs: Box<ServerStreamCallbacks>, + client_pipe: Option<PlatformHandle>, +} + +impl Drop for ServerStream { + fn drop(&mut self) { + // `stream` *must* be dropped before `cbs`. + drop(self.stream.take()); + } +} + +struct DeviceCollectionChangeCallback { + rpc: rpccore::Proxy<DeviceCollectionReq, DeviceCollectionResp>, +} + +impl DeviceCollectionChangeCallback { + fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) { + // TODO: Assert device_type is in devtype. + debug!( + "Sending device collection ({:?}) changed event", + device_type + ); + let _ = self + .rpc + .call(DeviceCollectionReq::DeviceChange(device_type)); + } +} + +pub struct CubebServer { + callback_thread: ipccore::EventLoopHandle, + device_collection_thread: ipccore::EventLoopHandle, + streams: slab::Slab<ServerStream>, + remote_pid: Option<u32>, + device_collection_change_callbacks: Option<Rc<DeviceCollectionChangeCallback>>, + devidmap: DevIdMap, + shm_area_size: usize, +} + +impl Drop for CubebServer { + fn drop(&mut self) { + if let Some(device_collection_change_callbacks) = &self.device_collection_change_callbacks { + debug!("CubebServer: dropped with device_collection_change_callbacks registered"); + CONTEXT_KEY.with(|k| { + let mut state = k.borrow_mut(); + if let Some(CubebContextState { + manager, + context: Ok(context), + }) = state.as_mut() + { + for devtype in [cubeb::DeviceType::INPUT, cubeb::DeviceType::OUTPUT] { + let r = manager.unregister( + context, + device_collection_change_callbacks, + devtype, + ); + if r.is_err() { + debug!("CubebServer: unregister failed: {:?}", r); + } + } + } + }) + } + } +} + +#[allow(unknown_lints)] // non_send_fields_in_send_ty is Nightly-only as of 2021-11-29. +#[allow(clippy::non_send_fields_in_send_ty)] +// XXX: required for server setup, verify this is safe. +unsafe impl Send for CubebServer {} + +impl rpccore::Server for CubebServer { + type ServerMessage = ServerMessage; + type ClientMessage = ClientMessage; + + fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage { + if let ServerMessage::ClientConnect(pid) = req { + self.remote_pid = Some(pid); + } + with_local_context(|context, manager| match *context { + Err(_) => error(cubeb::Error::error()), + Ok(ref context) => self.process_msg(context, manager, &req), + }) + } +} + +// Debugging for BMO 1594216/1612044. +macro_rules! try_stream { + ($self:expr, $stm_tok:expr) => { + if $self.streams.contains($stm_tok) { + $self.streams[$stm_tok] + .stream + .as_mut() + .expect("uninitialized stream") + } else { + error!( + "{}:{}:{} - Stream({}): invalid token", + file!(), + line!(), + column!(), + $stm_tok + ); + return error(cubeb::Error::invalid_parameter()); + } + }; +} + +impl CubebServer { + pub fn new( + callback_thread: ipccore::EventLoopHandle, + device_collection_thread: ipccore::EventLoopHandle, + shm_area_size: usize, + ) -> Self { + CubebServer { + callback_thread, + device_collection_thread, + streams: slab::Slab::<ServerStream>::new(), + remote_pid: None, + device_collection_change_callbacks: None, + devidmap: DevIdMap::new(), + shm_area_size, + } + } + + // Process a request coming from the client. + fn process_msg( + &mut self, + context: &cubeb::Context, + manager: &mut CubebDeviceCollectionManager, + msg: &ServerMessage, + ) -> ClientMessage { + let resp: ClientMessage = match *msg { + ServerMessage::ClientConnect(_) => { + // remote_pid is set before cubeb initialization, just verify here. + assert!(self.remote_pid.is_some()); + ClientMessage::ClientConnected + } + + ServerMessage::ClientDisconnect => { + // TODO: + //self.connection.client_disconnect(); + ClientMessage::ClientDisconnected + } + + ServerMessage::ContextGetBackendId => { + ClientMessage::ContextBackendId(context.backend_id().to_string()) + } + + ServerMessage::ContextGetMaxChannelCount => context + .max_channel_count() + .map(ClientMessage::ContextMaxChannelCount) + .unwrap_or_else(error), + + ServerMessage::ContextGetMinLatency(ref params) => { + let format = cubeb::SampleFormat::from(params.format); + let layout = cubeb::ChannelLayout::from(params.layout); + + let params = cubeb::StreamParamsBuilder::new() + .format(format) + .rate(params.rate) + .channels(params.channels) + .layout(layout) + .take(); + + context + .min_latency(¶ms) + .map(ClientMessage::ContextMinLatency) + .unwrap_or_else(error) + } + + ServerMessage::ContextGetPreferredSampleRate => context + .preferred_sample_rate() + .map(ClientMessage::ContextPreferredSampleRate) + .unwrap_or_else(error), + + ServerMessage::ContextGetDeviceEnumeration(device_type) => context + .enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type)) + .map(|devices| { + let v: Vec<DeviceInfo> = devices + .iter() + .map(|i| { + let mut tmp: DeviceInfo = i.as_ref().into(); + // Replace each cubeb_devid with a unique handle suitable for IPC. + tmp.devid = self.devidmap.make_handle(tmp.devid); + tmp + }) + .collect(); + ClientMessage::ContextEnumeratedDevices(v) + }) + .unwrap_or_else(error), + + ServerMessage::StreamCreate(ref params) => self + .process_stream_create(params) + .unwrap_or_else(|_| error(cubeb::Error::error())), + + ServerMessage::StreamInit(stm_tok, ref params) => self + .process_stream_init(context, stm_tok, params) + .unwrap_or_else(|_| error(cubeb::Error::error())), + + ServerMessage::StreamDestroy(stm_tok) => { + if self.streams.contains(stm_tok) { + debug!("Unregistering stream {:?}", stm_tok); + self.streams.remove(stm_tok); + } else { + // Debugging for BMO 1594216/1612044. + error!("StreamDestroy({}): invalid token", stm_tok); + return error(cubeb::Error::invalid_parameter()); + } + ClientMessage::StreamDestroyed + } + + ServerMessage::StreamStart(stm_tok) => try_stream!(self, stm_tok) + .start() + .map(|_| ClientMessage::StreamStarted) + .unwrap_or_else(error), + + ServerMessage::StreamStop(stm_tok) => try_stream!(self, stm_tok) + .stop() + .map(|_| ClientMessage::StreamStopped) + .unwrap_or_else(error), + + ServerMessage::StreamGetPosition(stm_tok) => try_stream!(self, stm_tok) + .position() + .map(ClientMessage::StreamPosition) + .unwrap_or_else(error), + + ServerMessage::StreamGetLatency(stm_tok) => try_stream!(self, stm_tok) + .latency() + .map(ClientMessage::StreamLatency) + .unwrap_or_else(error), + + ServerMessage::StreamGetInputLatency(stm_tok) => try_stream!(self, stm_tok) + .input_latency() + .map(ClientMessage::StreamInputLatency) + .unwrap_or_else(error), + + ServerMessage::StreamSetVolume(stm_tok, volume) => try_stream!(self, stm_tok) + .set_volume(volume) + .map(|_| ClientMessage::StreamVolumeSet) + .unwrap_or_else(error), + + ServerMessage::StreamSetName(stm_tok, ref name) => try_stream!(self, stm_tok) + .set_name(name) + .map(|_| ClientMessage::StreamNameSet) + .unwrap_or_else(error), + + ServerMessage::StreamGetCurrentDevice(stm_tok) => try_stream!(self, stm_tok) + .current_device() + .map(|device| ClientMessage::StreamCurrentDevice(Device::from(device))) + .unwrap_or_else(error), + + ServerMessage::StreamRegisterDeviceChangeCallback(stm_tok, enable) => { + try_stream!(self, stm_tok) + .register_device_changed_callback(if enable { + Some(device_change_cb_c) + } else { + None + }) + .map(|_| ClientMessage::StreamRegisterDeviceChangeCallback) + .unwrap_or_else(error) + } + + ServerMessage::ContextSetupDeviceCollectionCallback => { + let (server_pipe, client_pipe) = match sys::make_pipe_pair() { + Ok((server_pipe, client_pipe)) => (server_pipe, client_pipe), + Err(e) => { + debug!( + "ContextSetupDeviceCollectionCallback - make_pipe_pair failed: {:?}", + e + ); + return error(cubeb::Error::error()); + } + }; + + // TODO: this should bind the client_pipe and send the server_pipe to the remote, but + // additional work is required as it's not possible to convert a Windows sys::Pipe into a raw handle. + // TODO: Use the rpc_thread instead of an extra device_collection_thread, but a reentrant bind_client + // is required to support that. + let rpc = match self + .device_collection_thread + .bind_client::<DeviceCollectionClient>(server_pipe) + { + Ok(rpc) => rpc, + Err(e) => { + debug!( + "ContextSetupDeviceCollectionCallback - bind_client: {:?}", + e + ); + return error(cubeb::Error::error()); + } + }; + + self.device_collection_change_callbacks = + Some(Rc::new(DeviceCollectionChangeCallback { rpc })); + let fd = RegisterDeviceCollectionChanged { + platform_handle: SerializableHandle::new(client_pipe, self.remote_pid.unwrap()), + }; + + ClientMessage::ContextSetupDeviceCollectionCallback(fd) + } + + ServerMessage::ContextRegisterDeviceCollectionChanged(device_type, enable) => self + .process_register_device_collection_changed( + context, + manager, + cubeb::DeviceType::from_bits_truncate(device_type), + enable, + ) + .unwrap_or_else(error), + + #[cfg(target_os = "linux")] + ServerMessage::PromoteThreadToRealTime(thread_info) => { + let info = RtPriorityThreadInfo::deserialize(thread_info); + match promote_thread_to_real_time(info, 0, 48000) { + Ok(_) => { + info!("Promotion of content process thread to real-time OK"); + } + Err(_) => { + warn!("Promotion of content process thread to real-time error"); + } + } + ClientMessage::ThreadPromoted + } + }; + + trace!("process_msg: req={:?}, resp={:?}", msg, resp); + + resp + } + + fn process_register_device_collection_changed( + &mut self, + context: &cubeb::Context, + manager: &mut CubebDeviceCollectionManager, + devtype: cubeb::DeviceType, + enable: bool, + ) -> cubeb::Result<ClientMessage> { + if devtype == cubeb::DeviceType::UNKNOWN { + return Err(cubeb::Error::invalid_parameter()); + } + + assert!(self.device_collection_change_callbacks.is_some()); + let cbs = self.device_collection_change_callbacks.as_ref().unwrap(); + + if enable { + manager.register(context, cbs, devtype) + } else { + manager.unregister(context, cbs, devtype) + } + .map(|_| ClientMessage::ContextRegisteredDeviceCollectionChanged) + } + + // Stream create is special, so it's been separated from process_msg. + fn process_stream_create(&mut self, params: &StreamCreateParams) -> Result<ClientMessage> { + fn frame_size_in_bytes(params: Option<&StreamParams>) -> u16 { + params + .map(|p| { + let format = p.format.into(); + let sample_size = match format { + cubeb::SampleFormat::S16LE + | cubeb::SampleFormat::S16BE + | cubeb::SampleFormat::S16NE => 2, + cubeb::SampleFormat::Float32LE + | cubeb::SampleFormat::Float32BE + | cubeb::SampleFormat::Float32NE => 4, + }; + let channel_count = p.channels as u16; + sample_size * channel_count + }) + .unwrap_or(0u16) + } + + // Create the callback handling struct which is attached the cubeb stream. + let input_frame_size = frame_size_in_bytes(params.input_stream_params.as_ref()); + let output_frame_size = frame_size_in_bytes(params.output_stream_params.as_ref()); + + // Estimate a safe shmem size for this stream configuration. If the server was configured with a fixed + // shm_area_size override, use that instead. + // TODO: Add a new cubeb API to query the precise buffer size required for a given stream config. + // https://github.com/mozilla/audioipc-2/issues/124 + let shm_area_size = if self.shm_area_size == 0 { + let frame_size = output_frame_size.max(input_frame_size) as u32; + let in_rate = params.input_stream_params.map(|p| p.rate).unwrap_or(0); + let out_rate = params.output_stream_params.map(|p| p.rate).unwrap_or(0); + let rate = out_rate.max(in_rate); + // 1s of audio, rounded up to the nearest 64kB. + // Stream latency is capped at 1s in process_stream_init. + (((rate * frame_size) + 0xffff) & !0xffff) as usize + } else { + self.shm_area_size + }; + debug!("shm_area_size = {}", shm_area_size); + + let shm = SharedMem::new(&get_shm_id(), shm_area_size)?; + let shm_handle = unsafe { shm.make_handle()? }; + + let (server_pipe, client_pipe) = sys::make_pipe_pair()?; + // TODO: this should bind the client_pipe and send the server_pipe to the remote, but + // additional work is required as it's not possible to convert a Windows sys::Pipe into a raw handle. + let rpc = self + .callback_thread + .bind_client::<CallbackClient>(server_pipe)?; + + let cbs = Box::new(ServerStreamCallbacks { + input_frame_size, + output_frame_size, + shm, + state_callback_rpc: rpc.clone(), + device_change_callback_rpc: rpc.clone(), + data_callback_rpc: rpc, + connected: AtomicBool::new(false), + }); + + let entry = self.streams.vacant_entry(); + let key = entry.key(); + debug!("Registering stream {:?}", key); + + entry.insert(ServerStream { + stream: None, + cbs, + client_pipe: Some(client_pipe), + }); + + Ok(ClientMessage::StreamCreated(StreamCreate { + token: key, + shm_handle: SerializableHandle::new(shm_handle, self.remote_pid.unwrap()), + shm_area_size, + })) + } + + // Stream init is special, so it's been separated from process_msg. + fn process_stream_init( + &mut self, + context: &cubeb::Context, + stm_tok: usize, + params: &StreamInitParams, + ) -> Result<ClientMessage> { + // Create cubeb stream from params + let stream_name = params + .stream_name + .as_ref() + .and_then(|name| CStr::from_bytes_with_nul(name).ok()); + + // Map IPC handle back to cubeb_devid. + let input_device = self.devidmap.handle_to_id(params.input_device) as *const _; + let input_stream_params = params.input_stream_params.as_ref().map(|isp| unsafe { + cubeb::StreamParamsRef::from_ptr(isp as *const StreamParams as *mut _) + }); + + // Map IPC handle back to cubeb_devid. + let output_device = self.devidmap.handle_to_id(params.output_device) as *const _; + let output_stream_params = params.output_stream_params.as_ref().map(|osp| unsafe { + cubeb::StreamParamsRef::from_ptr(osp as *const StreamParams as *mut _) + }); + + // TODO: Manage stream latency requests with respect to the RT deadlines the callback_thread was configured for. + fn round_up_pow2(v: u32) -> u32 { + debug_assert!(v >= 1); + 1 << (32 - (v - 1).leading_zeros()) + } + let rate = params + .output_stream_params + .map(|p| p.rate) + .unwrap_or_else(|| params.input_stream_params.map(|p| p.rate).unwrap()); + // Note: minimum latency supported by AudioIPC is currently ~5ms. This restriction may be reduced by later IPC improvements. + let min_latency = round_up_pow2(5 * rate / 1000); + // Note: maximum latency is restricted by the SharedMem size. + let max_latency = rate; + let latency = params.latency_frames.clamp(min_latency, max_latency); + trace!( + "stream rate={} latency requested={} calculated={}", + rate, + params.latency_frames, + latency + ); + + let server_stream = &mut self.streams[stm_tok]; + assert!(size_of::<Box<ServerStreamCallbacks>>() == size_of::<usize>()); + let user_ptr = server_stream.cbs.as_ref() as *const ServerStreamCallbacks as *mut c_void; + + let stream = unsafe { + let stream = context.stream_init( + stream_name, + input_device, + input_stream_params, + output_device, + output_stream_params, + latency, + Some(data_cb_c), + Some(state_cb_c), + user_ptr, + ); + match stream { + Ok(stream) => stream, + Err(e) => { + debug!("Unregistering stream {:?} (stream error {:?})", stm_tok, e); + self.streams.remove(stm_tok); + return Err(e.into()); + } + } + }; + + server_stream.stream = Some(stream); + + let client_pipe = server_stream + .client_pipe + .take() + .expect("invalid state after StreamCreated"); + server_stream.cbs.connected.store(true, Ordering::Release); + Ok(ClientMessage::StreamInitialized(SerializableHandle::new( + client_pipe, + self.remote_pid.unwrap(), + ))) + } +} + +// C callable callbacks +unsafe extern "C" fn data_cb_c( + _: *mut ffi::cubeb_stream, + user_ptr: *mut c_void, + input_buffer: *const c_void, + output_buffer: *mut c_void, + nframes: c_long, +) -> c_long { + let ok = panic::catch_unwind(|| { + let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks); + let input = if input_buffer.is_null() { + &[] + } else { + let nbytes = nframes * c_long::from(cbs.input_frame_size); + slice::from_raw_parts(input_buffer as *const u8, nbytes as usize) + }; + let output: &mut [u8] = if output_buffer.is_null() { + &mut [] + } else { + let nbytes = nframes * c_long::from(cbs.output_frame_size); + slice::from_raw_parts_mut(output_buffer as *mut u8, nbytes as usize) + }; + cbs.data_callback(input, output, nframes as isize) as c_long + }); + ok.unwrap_or(cubeb::ffi::CUBEB_ERROR as c_long) +} + +unsafe extern "C" fn state_cb_c( + _: *mut ffi::cubeb_stream, + user_ptr: *mut c_void, + state: ffi::cubeb_state, +) { + let ok = panic::catch_unwind(|| { + let state = cubeb::State::from(state); + let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks); + cbs.state_callback(state); + }); + ok.expect("State callback panicked"); +} + +unsafe extern "C" fn device_change_cb_c(user_ptr: *mut c_void) { + let ok = panic::catch_unwind(|| { + let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks); + cbs.device_change_callback(); + }); + ok.expect("Device change callback panicked"); +} + +unsafe extern "C" fn device_collection_changed_input_cb_c( + _: *mut ffi::cubeb, + user_ptr: *mut c_void, +) { + let ok = panic::catch_unwind(|| { + let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager); + manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_INPUT); + }); + ok.expect("Collection changed (input) callback panicked"); +} + +unsafe extern "C" fn device_collection_changed_output_cb_c( + _: *mut ffi::cubeb, + user_ptr: *mut c_void, +) { + let ok = panic::catch_unwind(|| { + let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager); + manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_OUTPUT); + }); + ok.expect("Collection changed (output) callback panicked"); +} diff --git a/third_party/rust/audioipc2/.cargo-checksum.json b/third_party/rust/audioipc2/.cargo-checksum.json new file mode 100644 index 0000000000..96d8f77888 --- /dev/null +++ b/third_party/rust/audioipc2/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"Cargo.toml":"6d3b3004351e3313ef74472ab6b3f96d59a785b3d873698ca4caf67eb3d47aab","benches/serialization.rs":"d56855d868dab6aa22c8b03a61084535351b76c94b68d8b1d20764e352fe473f","build.rs":"65df9a97c6cdaa3faf72581f04ac289197b0b1797d69d22c1796e957ff1089e2","src/codec.rs":"38408b512d935cd7889a03b25dd14b36083ec4e6d2fcabd636182cf45e3d50bc","src/errors.rs":"67a4a994d0724397657581cde153bdfc05ce86e7efc467f23fafc8f64df80fa4","src/ipccore.rs":"ba339eebdc2d8a6d0cb9b294344809e63e404e220fda643b63a8a3ff63a755e8","src/lib.rs":"9b107cb52081eeea3fa742d30361db70f7138baa423dfe21d37dcf5087afc338","src/messages.rs":"452362da2cace9a0f2e3134c190ecb6a9997f8be4036cde06643e17c6c238240","src/rpccore.rs":"025b6614f1c42b96b0a8e74fd7881032d338c66e0d67ec0af70f910a9e30ebe1","src/shm.rs":"1d88f19606899e3e477865d6b84bbe3e272f51618a1c2d57b6dab03a4787cde3","src/sys/mod.rs":"e6fa1d260abf093e1f7b50185195e2d3aee0eb8c9774c6f253953b5896d838f3","src/sys/unix/cmsg.rs":"22349b3df39b51b9c414da363313c92d41b02a623753ffcca6f59613e8f79eb2","src/sys/unix/cmsghdr.c":"d7344b3dc15cdce410c68669b848bb81f7fe36362cd3699668cb613fa05180f8","src/sys/unix/mod.rs":"59835f0d5509940078b1820a54f49fc5514adeb3e45e7d21e3ab917431da2e74","src/sys/unix/msg.rs":"0e297d73bae9414184f85c2209cca0a3fde6d999a3f1d3f42faa3f56b6d57233","src/sys/windows/mod.rs":"3441a3212c6d44443a5975621d9594b0c841e5a7f113aa1b108a080330df2b77"},"package":null}
\ No newline at end of file diff --git a/third_party/rust/audioipc2/Cargo.toml b/third_party/rust/audioipc2/Cargo.toml new file mode 100644 index 0000000000..98de8c77a6 --- /dev/null +++ b/third_party/rust/audioipc2/Cargo.toml @@ -0,0 +1,82 @@ +# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO +# +# When uploading crates to the registry Cargo will automatically +# "normalize" Cargo.toml files for maximal compatibility +# with all versions of Cargo and also rewrite `path` dependencies +# to registry (e.g., crates.io) dependencies. +# +# If you are reading this file be aware that the original Cargo.toml +# will likely look very different (and much more reasonable). +# See Cargo.toml.orig for the original contents. + +[package] +edition = "2018" +name = "audioipc2" +version = "0.5.0" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>", +] +description = "Remote Cubeb IPC" +license = "ISC" + +[[bench]] +name = "serialization" +harness = false + +[dependencies] +bincode = "1.3" +byteorder = "1" +bytes = "1" +crossbeam-queue = "0.3" +cubeb = "0.10" +log = "0.4" +scopeguard = "1.1.0" +serde = "1" +serde_bytes = "0.11" +serde_derive = "1" +slab = "0.4" + +[dependencies.error-chain] +version = "0.12.0" +default-features = false + +[dependencies.mio] +version = "0.8" +features = [ + "os-poll", + "net", + "os-ext", +] + +[dev-dependencies] +env_logger = "0.9" + +[dev-dependencies.criterion] +version = "0.3" +features = ["html_reports"] + +[build-dependencies] +cc = "1.0" + +[target."cfg(target_os = \"android\")".dependencies] +ashmem = "0.1.2" + +[target."cfg(target_os = \"linux\")".dependencies.audio_thread_priority] +version = "0.26.1" +default-features = false + +[target."cfg(unix)".dependencies] +arrayvec = "0.7" +iovec = "0.1" +libc = "0.2" +memmap2 = "0.5" + +[target."cfg(windows)".dependencies.winapi] +version = "0.3" +features = [ + "combaseapi", + "handleapi", + "memoryapi", + "objbase", +] diff --git a/third_party/rust/audioipc2/benches/serialization.rs b/third_party/rust/audioipc2/benches/serialization.rs new file mode 100644 index 0000000000..39f770a939 --- /dev/null +++ b/third_party/rust/audioipc2/benches/serialization.rs @@ -0,0 +1,93 @@ +use audioipc::codec::{Codec, LengthDelimitedCodec}; +use audioipc::messages::DeviceInfo; +use audioipc::ClientMessage; +use audioipc2 as audioipc; +use bytes::BytesMut; +use criterion::{criterion_group, criterion_main, BatchSize, Criterion}; + +fn bench(c: &mut Criterion, name: &str, msg: impl Fn() -> ClientMessage) { + let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> = + LengthDelimitedCodec::default(); + let mut buf = BytesMut::with_capacity(8192); + c.bench_function(&format!("encode/{}", name), |b| { + b.iter_batched( + || msg(), + |msg| { + codec.encode(msg, &mut buf).unwrap(); + buf.clear(); + }, + BatchSize::SmallInput, + ) + }); + + let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> = + LengthDelimitedCodec::default(); + let mut buf = BytesMut::with_capacity(8192); + codec.encode(msg(), &mut buf).unwrap(); + c.bench_function(&format!("decode/{}", name), |b| { + b.iter_batched_ref( + || buf.clone(), + |buf| { + codec.decode(buf).unwrap().unwrap(); + }, + BatchSize::SmallInput, + ) + }); + + let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> = + LengthDelimitedCodec::default(); + let mut buf = BytesMut::with_capacity(8192); + c.bench_function(&format!("roundtrip/{}", name), |b| { + b.iter_batched( + || msg(), + |msg| { + codec.encode(msg, &mut buf).unwrap(); + codec.decode(&mut buf).unwrap().unwrap(); + }, + BatchSize::SmallInput, + ) + }); +} + +pub fn criterion_benchmark(c: &mut Criterion) { + bench(c, "tiny", || ClientMessage::ClientConnected); + bench(c, "small", || ClientMessage::StreamPosition(0)); + bench(c, "medium", || { + ClientMessage::ContextEnumeratedDevices(make_device_vec(2)) + }); + bench(c, "large", || { + ClientMessage::ContextEnumeratedDevices(make_device_vec(20)) + }); + bench(c, "huge", || { + ClientMessage::ContextEnumeratedDevices(make_device_vec(128)) + }); +} + +criterion_group!(benches, criterion_benchmark); +criterion_main!(benches); + +fn make_device_vec(n: usize) -> Vec<DeviceInfo> { + let mut devices = Vec::with_capacity(n); + for i in 0..n { + let device = DeviceInfo { + devid: i, + device_id: Some(vec![0u8; 64]), + friendly_name: Some(vec![0u8; 64]), + group_id: Some(vec![0u8; 64]), + vendor_name: Some(vec![0u8; 64]), + device_type: 0, + state: 0, + preferred: 0, + format: 0, + default_format: 0, + max_channels: 0, + default_rate: 0, + max_rate: 0, + min_rate: 0, + latency_lo: 0, + latency_hi: 0, + }; + devices.push(device); + } + devices +} diff --git a/third_party/rust/audioipc2/build.rs b/third_party/rust/audioipc2/build.rs new file mode 100644 index 0000000000..453dafad96 --- /dev/null +++ b/third_party/rust/audioipc2/build.rs @@ -0,0 +1,7 @@ +fn main() { + if std::env::var_os("CARGO_CFG_UNIX").is_some() { + cc::Build::new() + .file("src/sys/unix/cmsghdr.c") + .compile("cmsghdr"); + } +} diff --git a/third_party/rust/audioipc2/src/codec.rs b/third_party/rust/audioipc2/src/codec.rs new file mode 100644 index 0000000000..2c61faa6ea --- /dev/null +++ b/third_party/rust/audioipc2/src/codec.rs @@ -0,0 +1,203 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +//! `Encoder`s and `Decoder`s from items to/from `BytesMut` buffers. + +// The assert in LengthDelimitedCodec::decode triggers this clippy warning but +// requires upgrading the workspace to Rust 2021 to resolve. +// This should be fixed in Rust 1.68, after which the following `allow` can be deleted. +#![allow(clippy::uninlined_format_args)] + +use bincode::{self, Options}; +use byteorder::{ByteOrder, LittleEndian}; +use bytes::{Buf, BufMut, BytesMut}; +use serde::de::DeserializeOwned; +use serde::ser::Serialize; +use std::convert::TryInto; +use std::fmt::Debug; +use std::io; +use std::marker::PhantomData; +use std::mem::size_of; + +//////////////////////////////////////////////////////////////////////////////// +// Split buffer into size delimited frames - This appears more complicated than +// might be necessary due to handling the possibility of messages being split +// across reads. + +pub trait Codec { + /// The type of items to be encoded into byte buffer + type In; + + /// The type of items to be returned by decoding from byte buffer + type Out; + + /// Attempts to decode a frame from the provided buffer of bytes. + fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>>; + + /// A default method available to be called when there are no more bytes + /// available to be read from the I/O. + fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result<Self::Out> { + match self.decode(buf)? { + Some(frame) => Ok(frame), + None => Err(io::Error::new( + io::ErrorKind::Other, + "bytes remaining on stream", + )), + } + } + + /// Encodes a frame into the buffer provided. + fn encode(&mut self, msg: Self::In, buf: &mut BytesMut) -> io::Result<()>; +} + +/// Codec based upon bincode serialization +/// +/// Messages that have been serialized using bincode are prefixed with +/// the length of the message to aid in deserialization, so that it's +/// known if enough data has been received to decode a complete +/// message. +pub struct LengthDelimitedCodec<In, Out> { + state: State, + encode_buf: Vec<u8>, + __in: PhantomData<In>, + __out: PhantomData<Out>, +} + +enum State { + Length, + Data(u32), +} + +const MAX_MESSAGE_LEN: u32 = 1024 * 1024; +const MAGIC: u64 = 0xa4d1_019c_c910_1d4a; +const HEADER_LEN: usize = size_of::<u32>() + size_of::<u64>(); + +impl<In, Out> Default for LengthDelimitedCodec<In, Out> { + fn default() -> Self { + Self { + state: State::Length, + encode_buf: Vec::with_capacity(crate::ipccore::IPC_CLIENT_BUFFER_SIZE), + __in: PhantomData, + __out: PhantomData, + } + } +} + +impl<In, Out> LengthDelimitedCodec<In, Out> { + // Lengths are encoded as little endian u32 + fn decode_length(buf: &mut BytesMut) -> Option<u32> { + if buf.len() < HEADER_LEN { + // Not enough data + return None; + } + + let magic = LittleEndian::read_u64(&buf[0..8]); + assert_eq!(magic, MAGIC); + + // Consume the length field + let n = LittleEndian::read_u32(&buf[8..12]); + buf.advance(HEADER_LEN); + Some(n) + } + + fn decode_data(buf: &mut BytesMut, n: u32) -> io::Result<Option<Out>> + where + Out: DeserializeOwned + Debug, + { + let n = n.try_into().unwrap(); + + // At this point, the buffer has already had the required capacity + // reserved. All there is to do is read. + if buf.len() < n { + return Ok(None); + } + + trace!("Attempting to decode"); + let msg = bincode::options() + .with_limit(MAX_MESSAGE_LEN as u64) + .deserialize::<Out>(&buf[..n]) + .map_err(|e| match *e { + bincode::ErrorKind::Io(e) => e, + _ => io::Error::new(io::ErrorKind::Other, *e), + })?; + buf.advance(n); + + trace!("... Decoded {:?}", msg); + Ok(Some(msg)) + } +} + +impl<In, Out> Codec for LengthDelimitedCodec<In, Out> +where + In: Serialize + Debug, + Out: DeserializeOwned + Debug, +{ + type In = In; + type Out = Out; + + fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>> { + let n = match self.state { + State::Length => { + match Self::decode_length(buf) { + Some(n) => { + assert!( + n <= MAX_MESSAGE_LEN, + "assertion failed: {} <= {}", + n, + MAX_MESSAGE_LEN + ); + self.state = State::Data(n); + + // Ensure that the buffer has enough space to read the + // incoming payload + buf.reserve(n.try_into().unwrap()); + + n + } + None => return Ok(None), + } + } + State::Data(n) => n, + }; + + match Self::decode_data(buf, n)? { + Some(data) => { + // Update the decode state + self.state = State::Length; + + // Make sure the buffer has enough space to read the next length header. + buf.reserve(HEADER_LEN); + + Ok(Some(data)) + } + None => Ok(None), + } + } + + fn encode(&mut self, item: Self::In, buf: &mut BytesMut) -> io::Result<()> { + trace!("Attempting to encode"); + + self.encode_buf.clear(); + if let Err(e) = bincode::options() + .with_limit(MAX_MESSAGE_LEN as u64) + .serialize_into::<_, Self::In>(&mut self.encode_buf, &item) + { + trace!("message encode failed: {:?}", *e); + match *e { + bincode::ErrorKind::Io(e) => return Err(e), + _ => return Err(io::Error::new(io::ErrorKind::Other, *e)), + } + } + + let encoded_len = self.encode_buf.len(); + assert!(encoded_len <= MAX_MESSAGE_LEN as usize); + buf.reserve(encoded_len + HEADER_LEN); + buf.put_u64_le(MAGIC); + buf.put_u32_le(encoded_len.try_into().unwrap()); + buf.extend_from_slice(&self.encode_buf); + + Ok(()) + } +} diff --git a/third_party/rust/audioipc2/src/errors.rs b/third_party/rust/audioipc2/src/errors.rs new file mode 100644 index 0000000000..f4f3e32f5f --- /dev/null +++ b/third_party/rust/audioipc2/src/errors.rs @@ -0,0 +1,18 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +error_chain! { + // Maybe replace with chain_err to improve the error info. + foreign_links { + Bincode(bincode::Error); + Io(std::io::Error); + Cubeb(cubeb::Error); + } + + // Replace bail!(str) with explicit errors. + errors { + Disconnected + } +} diff --git a/third_party/rust/audioipc2/src/ipccore.rs b/third_party/rust/audioipc2/src/ipccore.rs new file mode 100644 index 0000000000..16468b9015 --- /dev/null +++ b/third_party/rust/audioipc2/src/ipccore.rs @@ -0,0 +1,918 @@ +// Copyright © 2021 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use std::io::{self, Result}; +use std::sync::{mpsc, Arc}; +use std::thread; + +use mio::{event::Event, Events, Interest, Poll, Registry, Token, Waker}; +use slab::Slab; + +use crate::messages::AssociateHandleForMessage; +use crate::rpccore::{ + make_client, make_server, Client, Handler, Proxy, RequestQueue, RequestQueueSender, Server, +}; +use crate::{ + codec::Codec, + codec::LengthDelimitedCodec, + sys::{self, RecvMsg, SendMsg}, +}; + +use serde::{de::DeserializeOwned, Serialize}; +use std::fmt::Debug; + +const WAKE_TOKEN: Token = Token(!0); + +thread_local!(static IN_EVENTLOOP: std::cell::RefCell<Option<thread::ThreadId>> = std::cell::RefCell::new(None)); + +fn assert_not_in_event_loop_thread() { + IN_EVENTLOOP.with(|b| { + assert_ne!(*b.borrow(), Some(thread::current().id())); + }); +} + +// Requests sent by an EventLoopHandle to be serviced by +// the handle's associated EventLoop. +enum Request { + // See EventLoop::add_connection + AddConnection( + sys::Pipe, + Box<dyn Driver + Send>, + mpsc::Sender<Result<Token>>, + ), + // See EventLoop::shutdown + Shutdown, + // See EventLoop::wake_connection + WakeConnection(Token), +} + +// EventLoopHandle is a cloneable external reference +// to a running EventLoop, allowing registration of +// new client and server connections, in addition to +// requesting the EventLoop shut down cleanly. +#[derive(Clone, Debug)] +pub struct EventLoopHandle { + waker: Arc<Waker>, + requests: RequestQueueSender<Request>, +} + +impl EventLoopHandle { + pub fn bind_client<C: Client + 'static>( + &self, + connection: sys::Pipe, + ) -> Result<Proxy<<C as Client>::ServerMessage, <C as Client>::ClientMessage>> + where + <C as Client>::ServerMessage: Serialize + Debug + AssociateHandleForMessage + Send, + <C as Client>::ClientMessage: DeserializeOwned + Debug + AssociateHandleForMessage + Send, + { + let (handler, mut proxy) = make_client::<C>()?; + let driver = Box::new(FramedDriver::new(handler)); + let r = self.add_connection(connection, driver); + trace!("EventLoop::bind_client {:?}", r); + r.map(|token| { + proxy.connect_event_loop(self.clone(), token); + proxy + }) + } + + pub fn bind_server<S: Server + Send + 'static>( + &self, + server: S, + connection: sys::Pipe, + ) -> Result<()> + where + <S as Server>::ServerMessage: DeserializeOwned + Debug + AssociateHandleForMessage + Send, + <S as Server>::ClientMessage: Serialize + Debug + AssociateHandleForMessage + Send, + { + let handler = make_server::<S>(server); + let driver = Box::new(FramedDriver::new(handler)); + let r = self.add_connection(connection, driver); + trace!("EventLoop::bind_server {:?}", r); + r.map(|_| ()) + } + + // Register a new connection with associated driver on the EventLoop. + // TODO: Since this is called from a Gecko main thread, make this non-blocking wrt. the EventLoop. + fn add_connection( + &self, + connection: sys::Pipe, + driver: Box<dyn Driver + Send>, + ) -> Result<Token> { + assert_not_in_event_loop_thread(); + let (tx, rx) = mpsc::channel(); + self.requests + .push(Request::AddConnection(connection, driver, tx)) + .map_err(|_| { + debug!("EventLoopHandle::add_connection send failed"); + io::ErrorKind::ConnectionAborted + })?; + self.waker.wake()?; + rx.recv().map_err(|_| { + debug!("EventLoopHandle::add_connection recv failed"); + io::ErrorKind::ConnectionAborted + })? + } + + // Signal EventLoop to shutdown. Causes EventLoop::poll to return Ok(false). + fn shutdown(&self) -> Result<()> { + self.requests.push(Request::Shutdown).map_err(|_| { + debug!("EventLoopHandle::shutdown send failed"); + io::ErrorKind::ConnectionAborted + })?; + self.waker.wake() + } + + // Signal EventLoop to wake connection specified by `token` for processing. + pub(crate) fn wake_connection(&self, token: Token) { + if self.requests.push(Request::WakeConnection(token)).is_ok() { + self.waker.wake().expect("wake failed"); + } + } +} + +// EventLoop owns all registered connections, and is responsible for calling each connection's +// `handle_event` or `handle_wake` any time a readiness or wake event associated with that connection is +// produced. +struct EventLoop { + poll: Poll, + events: Events, + waker: Arc<Waker>, + name: String, + connections: Slab<Connection>, + requests: Arc<RequestQueue<Request>>, +} + +const EVENT_LOOP_INITIAL_CLIENTS: usize = 64; // Initial client allocation, exceeding this will cause the connection slab to grow. +const EVENT_LOOP_EVENTS_PER_ITERATION: usize = 256; // Number of events per poll() step, arbitrary limit. + +impl EventLoop { + fn new(name: String) -> Result<EventLoop> { + let poll = Poll::new()?; + let waker = Arc::new(Waker::new(poll.registry(), WAKE_TOKEN)?); + let eventloop = EventLoop { + poll, + events: Events::with_capacity(EVENT_LOOP_EVENTS_PER_ITERATION), + waker, + name, + connections: Slab::with_capacity(EVENT_LOOP_INITIAL_CLIENTS), + requests: Arc::new(RequestQueue::new(EVENT_LOOP_INITIAL_CLIENTS)), + }; + + Ok(eventloop) + } + + // Return a cloneable handle for controlling the EventLoop externally. + fn handle(&mut self) -> EventLoopHandle { + EventLoopHandle { + waker: self.waker.clone(), + requests: self.requests.new_sender(), + } + } + + // Register a connection and driver. + fn add_connection( + &mut self, + connection: sys::Pipe, + driver: Box<dyn Driver + Send>, + ) -> Result<Token> { + if self.connections.len() == self.connections.capacity() { + trace!("{}: connection slab full, insert will allocate", self.name); + } + let entry = self.connections.vacant_entry(); + let token = Token(entry.key()); + assert_ne!(token, WAKE_TOKEN); + let connection = Connection::new(connection, token, driver, self.poll.registry())?; + debug!("{}: {:?}: new connection", self.name, token); + entry.insert(connection); + Ok(token) + } + + // Step EventLoop once. Call this in a loop from a dedicated thread. + // Returns false if EventLoop is shutting down. + // Each step may call `handle_event` on any registered connection that + // has received readiness events from the poll wakeup. + fn poll(&mut self) -> Result<bool> { + loop { + let r = self.poll.poll(&mut self.events, None); + match r { + Ok(()) => break, + Err(ref e) if interrupted(e) => continue, + Err(e) => return Err(e), + } + } + + for event in self.events.iter() { + match event.token() { + WAKE_TOKEN => { + debug!("{}: WAKE: wake event, will process requests", self.name); + } + token => { + debug!("{}: {:?}: connection event: {:?}", self.name, token, event); + let done = if let Some(connection) = self.connections.get_mut(token.0) { + match connection.handle_event(event, self.poll.registry()) { + Ok(done) => { + debug!("{}: connection {:?} done={}", self.name, token, done); + done + } + Err(e) => { + debug!("{}: {:?}: connection error: {:?}", self.name, token, e); + true + } + } + } else { + // Spurious event, log and ignore. + debug!( + "{}: {:?}: token not found in slab: {:?}", + self.name, token, event + ); + false + }; + if done { + debug!("{}: {:?}: done, removing", self.name, token); + let mut connection = self.connections.remove(token.0); + if let Err(e) = connection.shutdown(self.poll.registry()) { + debug!( + "{}: EventLoop drop - closing connection for {:?} failed: {:?}", + self.name, token, e + ); + } + } + } + } + } + + // If the waker was signalled there may be pending requests to process. + while let Some(req) = self.requests.pop() { + match req { + Request::AddConnection(pipe, driver, tx) => { + debug!("{}: EventLoop: handling add_connection", self.name); + let r = self.add_connection(pipe, driver); + tx.send(r).expect("EventLoop::add_connection"); + } + Request::Shutdown => { + debug!("{}: EventLoop: handling shutdown", self.name); + return Ok(false); + } + Request::WakeConnection(token) => { + debug!( + "{}: EventLoop: handling wake_connection {:?}", + self.name, token + ); + let done = if let Some(connection) = self.connections.get_mut(token.0) { + match connection.handle_wake(self.poll.registry()) { + Ok(done) => done, + Err(e) => { + debug!("{}: {:?}: connection error: {:?}", self.name, token, e); + true + } + } + } else { + // Spurious wake, log and ignore. + debug!( + "{}: {:?}: token not found in slab: wake_connection", + self.name, token + ); + false + }; + if done { + debug!("{}: {:?}: done (wake), removing", self.name, token); + let mut connection = self.connections.remove(token.0); + if let Err(e) = connection.shutdown(self.poll.registry()) { + debug!( + "{}: EventLoop drop - closing connection for {:?} failed: {:?}", + self.name, token, e + ); + } + } + } + } + } + + Ok(true) + } +} + +impl Drop for EventLoop { + fn drop(&mut self) { + debug!("{}: EventLoop drop", self.name); + for (token, connection) in &mut self.connections { + debug!( + "{}: EventLoop drop - closing connection for {:?}", + self.name, token + ); + if let Err(e) = connection.shutdown(self.poll.registry()) { + debug!( + "{}: EventLoop drop - closing connection for {:?} failed: {:?}", + self.name, token, e + ); + } + } + debug!("{}: EventLoop drop done", self.name); + } +} + +// Connection wraps an interprocess connection (Pipe) and manages +// receiving inbound and sending outbound buffers (and associated handles, if any). +// The associated driver is responsible for message framing and serialization. +struct Connection { + io: sys::Pipe, + token: Token, + interest: Option<Interest>, + inbound: sys::ConnectionBuffer, + outbound: sys::ConnectionBuffer, + driver: Box<dyn Driver + Send>, +} + +pub(crate) const IPC_CLIENT_BUFFER_SIZE: usize = 16384; + +impl Connection { + fn new( + mut io: sys::Pipe, + token: Token, + driver: Box<dyn Driver + Send>, + registry: &Registry, + ) -> Result<Connection> { + let interest = Interest::READABLE; + registry.register(&mut io, token, interest)?; + Ok(Connection { + io, + token, + interest: Some(interest), + inbound: sys::ConnectionBuffer::with_capacity(IPC_CLIENT_BUFFER_SIZE), + outbound: sys::ConnectionBuffer::with_capacity(IPC_CLIENT_BUFFER_SIZE), + driver, + }) + } + + fn shutdown(&mut self, registry: &Registry) -> Result<()> { + trace!( + "{:?}: connection shutdown interest={:?}", + self.token, + self.interest + ); + let r = self.io.shutdown(); + trace!("{:?}: connection shutdown r={:?}", self.token, r); + self.interest = None; + registry.deregister(&mut self.io) + } + + // Connections are always interested in READABLE. clear_readable is only + // called when the connection is in the process of shutting down. + fn clear_readable(&mut self, registry: &Registry) -> Result<()> { + self.update_registration( + registry, + self.interest.and_then(|i| i.remove(Interest::READABLE)), + ) + } + + // Connections toggle WRITABLE based on the state of the `outbound` buffer. + fn set_writable(&mut self, registry: &Registry) -> Result<()> { + self.update_registration( + registry, + Some( + self.interest + .map_or_else(|| Interest::WRITABLE, |i| i.add(Interest::WRITABLE)), + ), + ) + } + + fn clear_writable(&mut self, registry: &Registry) -> Result<()> { + self.update_registration( + registry, + self.interest.and_then(|i| i.remove(Interest::WRITABLE)), + ) + } + + // Update connection registration with the current readiness event interests. + fn update_registration( + &mut self, + registry: &Registry, + new_interest: Option<Interest>, + ) -> Result<()> { + // Note: Updating registration always triggers a writable event with NamedPipes, so + // it's important to skip updating registration when the set of interests hasn't changed. + if new_interest != self.interest { + trace!( + "{:?}: updating readiness registration old={:?} new={:?}", + self.token, + self.interest, + new_interest + ); + self.interest = new_interest; + if let Some(interest) = self.interest { + registry.reregister(&mut self.io, self.token, interest)?; + } else { + registry.deregister(&mut self.io)?; + } + } + Ok(()) + } + + // Handle readiness event. Errors returned are fatal for this connection, resulting in removal from the EventLoop connection list. + // The EventLoop will call this for any connection that has received an event. + fn handle_event(&mut self, event: &Event, registry: &Registry) -> Result<bool> { + debug!("{:?}: handling event {:?}", self.token, event); + assert_eq!(self.token, event.token()); + let done = if event.is_readable() { + self.recv_inbound()? + } else { + trace!("{:?}: not readable", self.token); + false + }; + self.flush_outbound()?; + if self.send_outbound(registry)? { + // Hit EOF during send + return Ok(true); + } + debug!( + "{:?}: handling event done (recv done={}, outbound={})", + self.token, + done, + self.outbound.is_empty() + ); + let done = done && self.outbound.is_empty(); + // If driver is done and outbound is clear, unregister connection. + if done { + trace!("{:?}: driver done, clearing read interest", self.token); + self.clear_readable(registry)?; + } + Ok(done) + } + + // Handle wake event. Errors returned are fatal for this connection, resulting in removal from the EventLoop connection list. + // The EventLoop will call this to clear the outbound buffer for any connection that has received a wake event. + fn handle_wake(&mut self, registry: &Registry) -> Result<bool> { + debug!("{:?}: handling wake", self.token); + self.flush_outbound()?; + if self.send_outbound(registry)? { + // Hit EOF during send + return Ok(true); + } + debug!("{:?}: handling wake done", self.token); + Ok(false) + } + + fn recv_inbound(&mut self) -> Result<bool> { + // If the connection is readable, read into inbound and pass to driver for processing until all ready data + // has been consumed. + loop { + trace!("{:?}: pre-recv inbound: {:?}", self.token, self.inbound); + let r = self.io.recv_msg(&mut self.inbound); + match r { + Ok(0) => { + trace!( + "{:?}: recv EOF unprocessed inbound={}", + self.token, + self.inbound.is_empty() + ); + return Ok(true); + } + Ok(n) => { + trace!("{:?}: recv bytes: {}, process_inbound", self.token, n); + let r = self.driver.process_inbound(&mut self.inbound); + trace!("{:?}: process_inbound done: {:?}", self.token, r); + match r { + Ok(done) => { + if done { + return Ok(true); + } + } + Err(e) => { + debug!( + "{:?}: process_inbound error: {:?} unprocessed inbound={}", + self.token, + e, + self.inbound.is_empty() + ); + return Err(e); + } + } + } + Err(ref e) if would_block(e) => { + trace!("{:?}: recv would_block: {:?}", self.token, e); + return Ok(false); + } + Err(ref e) if interrupted(e) => { + trace!("{:?}: recv interrupted: {:?}", self.token, e); + continue; + } + Err(e) => { + debug!("{:?}: recv error: {:?}", self.token, e); + return Err(e); + } + } + } + } + + fn flush_outbound(&mut self) -> Result<()> { + // Enqueue outbound messages to the outbound buffer, then try to write out to connection. + // There may be outbound messages even if there was no inbound processing, so always attempt + // to enqueue and flush. + trace!("{:?}: flush_outbound", self.token); + let r = self.driver.flush_outbound(&mut self.outbound); + trace!("{:?}: flush_outbound done: {:?}", self.token, r); + if let Err(e) = r { + debug!("{:?}: flush_outbound error: {:?}", self.token, e); + return Err(e); + } + Ok(()) + } + + fn send_outbound(&mut self, registry: &Registry) -> Result<bool> { + // Attempt to flush outbound buffer. If the connection's write buffer is full, register for WRITABLE + // and complete flushing when associated notitication arrives later. + while !self.outbound.is_empty() { + let r = self.io.send_msg(&mut self.outbound); + match r { + Ok(0) => { + trace!("{:?}: send EOF", self.token); + return Ok(true); + } + Ok(n) => { + trace!("{:?}: send bytes: {}", self.token, n); + } + Err(ref e) if would_block(e) => { + trace!( + "{:?}: send would_block: {:?}, setting write interest", + self.token, + e + ); + // Register for write events. + self.set_writable(registry)?; + break; + } + Err(ref e) if interrupted(e) => { + trace!("{:?}: send interrupted: {:?}", self.token, e); + continue; + } + Err(e) => { + debug!("{:?}: send error: {:?}", self.token, e); + return Err(e); + } + } + trace!("{:?}: post-send: outbound {:?}", self.token, self.outbound); + } + // Outbound buffer flushed, clear registration for WRITABLE. + if self.outbound.is_empty() { + trace!("{:?}: outbound empty, clearing write interest", self.token); + self.clear_writable(registry)?; + } + Ok(false) + } +} + +impl Drop for Connection { + fn drop(&mut self) { + debug!("{:?}: Connection drop", self.token); + } +} + +fn would_block(err: &std::io::Error) -> bool { + err.kind() == std::io::ErrorKind::WouldBlock +} + +fn interrupted(err: &std::io::Error) -> bool { + err.kind() == std::io::ErrorKind::Interrupted +} + +// Driver only has a single implementation, but must be hidden behind a Trait object to +// hide the varying FramedDriver sizes (due to different `T` values). +trait Driver { + // Handle inbound messages. Returns true if Driver is done; this will trigger Connection removal and cleanup. + fn process_inbound(&mut self, inbound: &mut sys::ConnectionBuffer) -> Result<bool>; + + // Write outbound messages to `outbound`. + fn flush_outbound(&mut self, outbound: &mut sys::ConnectionBuffer) -> Result<()>; +} + +// Length-delimited connection framing and (de)serialization is handled by the inbound and outbound processing. +// Handlers can then process message Requests and Responses without knowledge of serialization or +// handle remoting. +impl<T> Driver for FramedDriver<T> +where + T: Handler, + T::In: DeserializeOwned + Debug + AssociateHandleForMessage, + T::Out: Serialize + Debug + AssociateHandleForMessage, +{ + // Caller passes `inbound` data, this function will trim any complete messages from `inbound` and pass them to the handler for processing. + fn process_inbound(&mut self, inbound: &mut sys::ConnectionBuffer) -> Result<bool> { + debug!("process_inbound: {:?}", inbound); + + // Repeatedly call `decode` as long as it produces items, passing each produced item to the handler to action. + #[allow(unused_mut)] + while let Some(mut item) = self.codec.decode(&mut inbound.buf)? { + if item.has_associated_handle() { + // On Unix, dequeue a handle from the connection and update the item's handle. + #[cfg(unix)] + { + let new = inbound + .pop_handle() + .expect("inbound handle expected for item"); + unsafe { item.set_local_handle(new.take()) }; + } + // On Windows, the deserialized item contains the correct handle value, so + // convert it to an owned handle on the item. + #[cfg(windows)] + { + assert!(inbound.pop_handle().is_none()); + unsafe { item.set_local_handle() }; + } + } + + self.handler.consume(item)?; + } + + Ok(false) + } + + // Caller will try to write `outbound` to associated connection, queuing any data that can't be transmitted immediately. + fn flush_outbound(&mut self, outbound: &mut sys::ConnectionBuffer) -> Result<()> { + debug!("flush_outbound: {:?}", outbound.buf); + + // Repeatedly grab outgoing items from the handler, passing each to `encode` for serialization into `outbound`. + while let Some(mut item) = self.handler.produce()? { + let handle = if item.has_associated_handle() { + #[allow(unused_mut)] + let mut handle = item.take_handle(); + // On Windows, the handle is transferred by duplicating it into the target remote process. + #[cfg(windows)] + unsafe { + item.set_remote_handle(handle.send_to_target()?); + } + Some(handle) + } else { + None + }; + + self.codec.encode(item, &mut outbound.buf)?; + if let Some(handle) = handle { + // `outbound` retains ownership of the handle until the associated + // encoded item in `outbound.buf` is sent to the remote process. + outbound.push_handle(handle); + } + } + Ok(()) + } +} + +struct FramedDriver<T: Handler> { + codec: LengthDelimitedCodec<T::Out, T::In>, + handler: T, +} + +impl<T: Handler> FramedDriver<T> { + fn new(handler: T) -> FramedDriver<T> { + FramedDriver { + codec: Default::default(), + handler, + } + } +} + +#[derive(Debug)] +pub struct EventLoopThread { + thread: Option<thread::JoinHandle<Result<()>>>, + name: String, + handle: EventLoopHandle, +} + +impl EventLoopThread { + pub fn new<F1, F2>( + name: String, + stack_size: Option<usize>, + after_start: F1, + before_stop: F2, + ) -> Result<Self> + where + F1: Fn() + Send + 'static, + F2: Fn() + Send + 'static, + { + let mut event_loop = EventLoop::new(name.clone())?; + let handle = event_loop.handle(); + + let builder = thread::Builder::new() + .name(name.clone()) + .stack_size(stack_size.unwrap_or(64 * 4096)); + + let thread = builder.spawn(move || { + trace!("{}: event loop thread enter", event_loop.name); + after_start(); + let _thread_exit_guard = scopeguard::guard((), |_| before_stop()); + + let r = loop { + let start = std::time::Instant::now(); + let r = event_loop.poll(); + trace!( + "{}: event loop poll r={:?}, took={}μs", + event_loop.name, + r, + start.elapsed().as_micros() + ); + match r { + Ok(true) => continue, + _ => break r, + } + }; + + trace!("{}: event loop thread exit", event_loop.name); + r.map(|_| ()) + })?; + + Ok(EventLoopThread { + thread: Some(thread), + name, + handle, + }) + } + + pub fn handle(&self) -> &EventLoopHandle { + &self.handle + } +} + +impl Drop for EventLoopThread { + // Shut down event loop and executor thread. Blocks until complete. + fn drop(&mut self) { + trace!("{}: EventLoopThread shutdown", self.name); + if let Err(e) = self.handle.shutdown() { + debug!("{}: initiating shutdown failed: {:?}", self.name, e); + } + let thread = self.thread.take().expect("event loop thread"); + if let Err(e) = thread.join() { + error!("{}: EventLoopThread failed: {:?}", self.name, e); + } + trace!("{}: EventLoopThread shutdown done", self.name); + } +} + +#[cfg(test)] +mod test { + use serde_derive::{Deserialize, Serialize}; + + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum TestServerMessage { + TestRequest, + } + impl AssociateHandleForMessage for TestServerMessage {} + + struct TestServerImpl {} + + impl Server for TestServerImpl { + type ServerMessage = TestServerMessage; + type ClientMessage = TestClientMessage; + + fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage { + assert_eq!(req, TestServerMessage::TestRequest); + TestClientMessage::TestResponse + } + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum TestClientMessage { + TestResponse, + } + + impl AssociateHandleForMessage for TestClientMessage {} + + struct TestClientImpl {} + + impl Client for TestClientImpl { + type ServerMessage = TestServerMessage; + type ClientMessage = TestClientMessage; + } + + fn init() { + let _ = env_logger::builder().is_test(true).try_init(); + } + + fn setup() -> ( + EventLoopThread, + EventLoopThread, + Proxy<TestServerMessage, TestClientMessage>, + ) { + // Server setup and registration. + let server = EventLoopThread::new("test-server".to_string(), None, || {}, || {}) + .expect("server EventLoopThread"); + let server_handle = server.handle(); + + let (server_pipe, client_pipe) = sys::make_pipe_pair().expect("server make_pipe_pair"); + server_handle + .bind_server(TestServerImpl {}, server_pipe) + .expect("server bind_server"); + + // Client setup and registration. + let client = EventLoopThread::new("test-client".to_string(), None, || {}, || {}) + .expect("client EventLoopThread"); + let client_handle = client.handle(); + + let client_pipe = unsafe { sys::Pipe::from_raw_handle(client_pipe) }; + let client_proxy = client_handle + .bind_client::<TestClientImpl>(client_pipe) + .expect("client bind_client"); + + (server, client, client_proxy) + } + + // Verify basic EventLoopThread functionality works. Create a server and client EventLoopThread, then send + // a single message from the client to the server and wait for the expected response. + #[test] + fn basic() { + init(); + let (server, client, client_proxy) = setup(); + + // RPC message from client to server. + let response = client_proxy.call(TestServerMessage::TestRequest); + let response = response.expect("client response"); + assert_eq!(response, TestClientMessage::TestResponse); + + // Explicit shutdown. + drop(client); + drop(server); + } + + // Same as `basic`, but shut down server before client. + #[test] + fn basic_reverse_drop_order() { + init(); + let (server, client, client_proxy) = setup(); + + // RPC message from client to server. + let response = client_proxy.call(TestServerMessage::TestRequest); + let response = response.expect("client response"); + assert_eq!(response, TestClientMessage::TestResponse); + + // Explicit shutdown. + drop(server); + drop(client); + } + + #[test] + fn dead_server() { + init(); + let (server, _client, client_proxy) = setup(); + drop(server); + + let response = client_proxy.call(TestServerMessage::TestRequest); + response.expect_err("sending on closed channel"); + } + + #[test] + fn dead_client() { + init(); + let (_server, client, client_proxy) = setup(); + drop(client); + + let response = client_proxy.call(TestServerMessage::TestRequest); + response.expect_err("sending on a closed channel"); + } + + #[test] + fn disconnected_handle() { + init(); + let server = EventLoopThread::new("test-server".to_string(), None, || {}, || {}) + .expect("server EventLoopThread"); + let server_handle = server.handle().clone(); + drop(server); + + server_handle + .shutdown() + .expect_err("sending on closed channel"); + } + + #[test] + fn clone_after_drop() { + init(); + let (server, client, client_proxy) = setup(); + drop(server); + drop(client); + + let clone = client_proxy.clone(); + let response = clone.call(TestServerMessage::TestRequest); + response.expect_err("sending to a dropped ClientHandler"); + } + + #[test] + fn basic_event_loop_thread_callbacks() { + init(); + let (start_tx, start_rx) = mpsc::channel(); + let (stop_tx, stop_rx) = mpsc::channel(); + + let elt = EventLoopThread::new( + "test-thread-callbacks".to_string(), + None, + move || start_tx.send(()).unwrap(), + move || stop_tx.send(()).unwrap(), + ) + .expect("server EventLoopThread"); + + start_rx.recv().expect("after_start callback done"); + + drop(elt); + + stop_rx.recv().expect("before_stop callback done"); + } +} diff --git a/third_party/rust/audioipc2/src/lib.rs b/third_party/rust/audioipc2/src/lib.rs new file mode 100644 index 0000000000..eb6d843ba4 --- /dev/null +++ b/third_party/rust/audioipc2/src/lib.rs @@ -0,0 +1,214 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +#![warn(unused_extern_crates)] +#![recursion_limit = "1024"] +#[macro_use] +extern crate error_chain; +#[macro_use] +extern crate log; + +pub mod codec; +#[allow(deprecated)] +pub mod errors; +pub mod messages; +pub mod shm; + +pub mod ipccore; +pub mod rpccore; +pub mod sys; + +pub use crate::messages::{ClientMessage, ServerMessage}; + +#[cfg(unix)] +use std::os::unix::io::IntoRawFd; +#[cfg(windows)] +use std::os::windows::io::IntoRawHandle; + +use std::io::Result; + +// This must match the definition of +// ipc::FileDescriptor::PlatformHandleType in Gecko. +#[cfg(windows)] +pub type PlatformHandleType = std::os::windows::raw::HANDLE; +#[cfg(unix)] +pub type PlatformHandleType = libc::c_int; + +// This stands in for RawFd/RawHandle. +#[derive(Debug)] +pub struct PlatformHandle(PlatformHandleType); + +#[cfg(unix)] +pub const INVALID_HANDLE_VALUE: PlatformHandleType = -1isize as PlatformHandleType; + +#[cfg(windows)] +pub const INVALID_HANDLE_VALUE: PlatformHandleType = winapi::um::handleapi::INVALID_HANDLE_VALUE; + +#[cfg(unix)] +fn valid_handle(handle: PlatformHandleType) -> bool { + handle >= 0 +} + +#[cfg(windows)] +fn valid_handle(handle: PlatformHandleType) -> bool { + handle != INVALID_HANDLE_VALUE && !handle.is_null() +} + +impl PlatformHandle { + pub fn new(raw: PlatformHandleType) -> PlatformHandle { + assert!(valid_handle(raw)); + PlatformHandle(raw) + } + + #[cfg(windows)] + pub fn from<T: IntoRawHandle>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_handle()) + } + + #[cfg(unix)] + pub fn from<T: IntoRawFd>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_fd()) + } + + #[allow(clippy::missing_safety_doc)] + pub unsafe fn into_raw(self) -> PlatformHandleType { + let handle = self.0; + std::mem::forget(self); + handle + } + + #[cfg(unix)] + pub fn duplicate(h: PlatformHandleType) -> Result<PlatformHandle> { + unsafe { + let newfd = libc::dup(h); + if !valid_handle(newfd) { + return Err(std::io::Error::last_os_error()); + } + Ok(PlatformHandle::from(newfd)) + } + } + + #[allow(clippy::missing_safety_doc)] + #[cfg(windows)] + pub unsafe fn duplicate(h: PlatformHandleType) -> Result<PlatformHandle> { + let dup = duplicate_platform_handle(h, None)?; + Ok(PlatformHandle::new(dup)) + } +} + +impl Drop for PlatformHandle { + fn drop(&mut self) { + unsafe { close_platform_handle(self.0) } + } +} + +#[cfg(unix)] +unsafe fn close_platform_handle(handle: PlatformHandleType) { + libc::close(handle); +} + +#[cfg(windows)] +unsafe fn close_platform_handle(handle: PlatformHandleType) { + winapi::um::handleapi::CloseHandle(handle); +} + +#[cfg(windows)] +use winapi::shared::minwindef::{DWORD, FALSE}; +#[cfg(windows)] +use winapi::um::{handleapi, processthreadsapi, winnt}; + +// Duplicate `source_handle` to `target_pid`. Returns the value of the new handle inside the target process. +// If `target_pid` is `None`, `source_handle` is duplicated in the current process. +#[cfg(windows)] +pub(crate) unsafe fn duplicate_platform_handle( + source_handle: PlatformHandleType, + target_pid: Option<DWORD>, +) -> Result<PlatformHandleType> { + let source_process = processthreadsapi::GetCurrentProcess(); + let target_process = if let Some(pid) = target_pid { + let target = processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, pid); + if !valid_handle(target) { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "invalid target process", + )); + } + Some(target) + } else { + None + }; + + let mut target_handle = std::ptr::null_mut(); + let ok = handleapi::DuplicateHandle( + source_process, + source_handle, + target_process.unwrap_or(source_process), + &mut target_handle, + 0, + FALSE, + winnt::DUPLICATE_SAME_ACCESS, + ); + if let Some(target) = target_process { + handleapi::CloseHandle(target); + }; + if ok == FALSE { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "DuplicateHandle failed", + )); + } + Ok(target_handle) +} + +// Close `target_handle_to_close` inside target process `target_pid` using a +// special invocation of `DuplicateHandle`. See +// https://docs.microsoft.com/en-us/windows/win32/api/handleapi/nf-handleapi-duplicatehandle#:~:text=Normally%20the%20target,dwOptions%20to%20DUPLICATE_CLOSE_SOURCE. +#[cfg(windows)] +pub(crate) unsafe fn close_target_handle( + target_handle_to_close: PlatformHandleType, + target_pid: DWORD, +) -> Result<()> { + let target_process = + processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, target_pid); + if !valid_handle(target_process) { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "invalid target process", + )); + } + + let ok = handleapi::DuplicateHandle( + target_process, + target_handle_to_close, + std::ptr::null_mut(), + std::ptr::null_mut(), + 0, + FALSE, + winnt::DUPLICATE_CLOSE_SOURCE, + ); + handleapi::CloseHandle(target_process); + if ok == FALSE { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "DuplicateHandle failed", + )); + } + Ok(()) +} + +#[cfg(windows)] +pub fn server_platform_init() { + use winapi::shared::winerror; + use winapi::um::combaseapi; + use winapi::um::objbase; + + unsafe { + let r = combaseapi::CoInitializeEx(std::ptr::null_mut(), objbase::COINIT_MULTITHREADED); + assert!(winerror::SUCCEEDED(r)); + } +} + +#[cfg(unix)] +pub fn server_platform_init() {} diff --git a/third_party/rust/audioipc2/src/messages.rs b/third_party/rust/audioipc2/src/messages.rs new file mode 100644 index 0000000000..853f056bae --- /dev/null +++ b/third_party/rust/audioipc2/src/messages.rs @@ -0,0 +1,632 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::PlatformHandle; +use crate::PlatformHandleType; +use crate::INVALID_HANDLE_VALUE; +#[cfg(target_os = "linux")] +use audio_thread_priority::RtPriorityThreadInfo; +use cubeb::{self, ffi}; +use serde_derive::Deserialize; +use serde_derive::Serialize; +use std::ffi::{CStr, CString}; +use std::os::raw::{c_char, c_int, c_uint}; +use std::ptr; + +#[derive(Debug, Serialize, Deserialize)] +pub struct Device { + #[serde(with = "serde_bytes")] + pub output_name: Option<Vec<u8>>, + #[serde(with = "serde_bytes")] + pub input_name: Option<Vec<u8>>, +} + +impl<'a> From<&'a cubeb::DeviceRef> for Device { + fn from(info: &'a cubeb::DeviceRef) -> Self { + Self { + output_name: info.output_name_bytes().map(|s| s.to_vec()), + input_name: info.input_name_bytes().map(|s| s.to_vec()), + } + } +} + +impl From<ffi::cubeb_device> for Device { + fn from(info: ffi::cubeb_device) -> Self { + Self { + output_name: dup_str(info.output_name), + input_name: dup_str(info.input_name), + } + } +} + +impl From<Device> for ffi::cubeb_device { + fn from(info: Device) -> Self { + Self { + output_name: opt_str(info.output_name), + input_name: opt_str(info.input_name), + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct DeviceInfo { + pub devid: usize, + #[serde(with = "serde_bytes")] + pub device_id: Option<Vec<u8>>, + #[serde(with = "serde_bytes")] + pub friendly_name: Option<Vec<u8>>, + #[serde(with = "serde_bytes")] + pub group_id: Option<Vec<u8>>, + #[serde(with = "serde_bytes")] + pub vendor_name: Option<Vec<u8>>, + + pub device_type: ffi::cubeb_device_type, + pub state: ffi::cubeb_device_state, + pub preferred: ffi::cubeb_device_pref, + + pub format: ffi::cubeb_device_fmt, + pub default_format: ffi::cubeb_device_fmt, + pub max_channels: u32, + pub default_rate: u32, + pub max_rate: u32, + pub min_rate: u32, + + pub latency_lo: u32, + pub latency_hi: u32, +} + +impl<'a> From<&'a cubeb::DeviceInfoRef> for DeviceInfo { + fn from(info: &'a cubeb::DeviceInfoRef) -> Self { + let info = unsafe { &*info.as_ptr() }; + DeviceInfo { + devid: info.devid as _, + device_id: dup_str(info.device_id), + friendly_name: dup_str(info.friendly_name), + group_id: dup_str(info.group_id), + vendor_name: dup_str(info.vendor_name), + + device_type: info.device_type, + state: info.state, + preferred: info.preferred, + + format: info.format, + default_format: info.default_format, + max_channels: info.max_channels, + default_rate: info.default_rate, + max_rate: info.max_rate, + min_rate: info.min_rate, + + latency_lo: info.latency_lo, + latency_hi: info.latency_hi, + } + } +} + +impl From<DeviceInfo> for ffi::cubeb_device_info { + fn from(info: DeviceInfo) -> Self { + ffi::cubeb_device_info { + devid: info.devid as _, + device_id: opt_str(info.device_id), + friendly_name: opt_str(info.friendly_name), + group_id: opt_str(info.group_id), + vendor_name: opt_str(info.vendor_name), + + device_type: info.device_type, + state: info.state, + preferred: info.preferred, + + format: info.format, + default_format: info.default_format, + max_channels: info.max_channels, + default_rate: info.default_rate, + max_rate: info.max_rate, + min_rate: info.min_rate, + + latency_lo: info.latency_lo, + latency_hi: info.latency_hi, + } + } +} + +#[repr(C)] +#[derive(Clone, Copy, Debug, Deserialize, Serialize)] +pub struct StreamParams { + pub format: ffi::cubeb_sample_format, + pub rate: c_uint, + pub channels: c_uint, + pub layout: ffi::cubeb_channel_layout, + pub prefs: ffi::cubeb_stream_prefs, +} + +impl From<&cubeb::StreamParamsRef> for StreamParams { + fn from(x: &cubeb::StreamParamsRef) -> StreamParams { + unsafe { *(x.as_ptr() as *mut StreamParams) } + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct StreamCreateParams { + pub input_stream_params: Option<StreamParams>, + pub output_stream_params: Option<StreamParams>, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct StreamInitParams { + #[serde(with = "serde_bytes")] + pub stream_name: Option<Vec<u8>>, + pub input_device: usize, + pub input_stream_params: Option<StreamParams>, + pub output_device: usize, + pub output_stream_params: Option<StreamParams>, + pub latency_frames: u32, +} + +fn dup_str(s: *const c_char) -> Option<Vec<u8>> { + if s.is_null() { + None + } else { + let vec: Vec<u8> = unsafe { CStr::from_ptr(s) }.to_bytes().to_vec(); + Some(vec) + } +} + +fn opt_str(v: Option<Vec<u8>>) -> *mut c_char { + match v { + Some(v) => match CString::new(v) { + Ok(s) => s.into_raw(), + Err(_) => { + debug!("Failed to convert bytes to CString"); + ptr::null_mut() + } + }, + None => ptr::null_mut(), + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct StreamCreate { + pub token: usize, + pub shm_handle: SerializableHandle, + pub shm_area_size: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RegisterDeviceCollectionChanged { + pub platform_handle: SerializableHandle, +} + +// Client -> Server messages. +// TODO: Callbacks should be different messages types so +// ServerConn::process_msg doesn't have a catch-all case. +#[derive(Debug, Serialize, Deserialize)] +pub enum ServerMessage { + ClientConnect(u32), + ClientDisconnect, + + ContextGetBackendId, + ContextGetMaxChannelCount, + ContextGetMinLatency(StreamParams), + ContextGetPreferredSampleRate, + ContextGetDeviceEnumeration(ffi::cubeb_device_type), + ContextSetupDeviceCollectionCallback, + ContextRegisterDeviceCollectionChanged(ffi::cubeb_device_type, bool), + + StreamCreate(StreamCreateParams), + StreamInit(usize, StreamInitParams), + StreamDestroy(usize), + + StreamStart(usize), + StreamStop(usize), + StreamGetPosition(usize), + StreamGetLatency(usize), + StreamGetInputLatency(usize), + StreamSetVolume(usize, f32), + StreamSetName(usize, CString), + StreamGetCurrentDevice(usize), + StreamRegisterDeviceChangeCallback(usize, bool), + + #[cfg(target_os = "linux")] + PromoteThreadToRealTime([u8; std::mem::size_of::<RtPriorityThreadInfo>()]), +} + +// Server -> Client messages. +// TODO: Streams need id. +#[derive(Debug, Serialize, Deserialize)] +pub enum ClientMessage { + ClientConnected, + ClientDisconnected, + + ContextBackendId(String), + ContextMaxChannelCount(u32), + ContextMinLatency(u32), + ContextPreferredSampleRate(u32), + ContextEnumeratedDevices(Vec<DeviceInfo>), + ContextSetupDeviceCollectionCallback(RegisterDeviceCollectionChanged), + ContextRegisteredDeviceCollectionChanged, + + StreamCreated(StreamCreate), + StreamInitialized(SerializableHandle), + StreamDestroyed, + + StreamStarted, + StreamStopped, + StreamPosition(u64), + StreamLatency(u32), + StreamInputLatency(u32), + StreamVolumeSet, + StreamNameSet, + StreamCurrentDevice(Device), + StreamRegisterDeviceChangeCallback, + + #[cfg(target_os = "linux")] + ThreadPromoted, + + Error(c_int), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum CallbackReq { + Data { + nframes: isize, + input_frame_size: usize, + output_frame_size: usize, + }, + State(ffi::cubeb_state), + DeviceChange, +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum CallbackResp { + Data(isize), + State, + DeviceChange, + Error(c_int), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionReq { + DeviceChange(ffi::cubeb_device_type), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionResp { + DeviceChange, +} + +// Represents a platform handle in various transitional states during serialization and remoting. +// The process of serializing and remoting handles and the ownership during various states differs +// between Windows and Unix. SerializableHandle changes during IPC as follows: +// +// 1. Created in the initial state `Owned`, with a valid `target_pid`. +// 2. Ownership is transferred out for processing during IPC send, becoming `Empty` temporarily. +// See `AssociateHandleForMessage::take_handle`. +// - Windows: DuplicateHandle transfers the handle to the remote process. +// This produces a new handle value in the local process representing the remote handle. +// This value must be sent to the remote, so `AssociateHandleForMessage::set_remote_handle` +// is used to transform the handle into a `SerializableValue`. +// - Unix: sendmsg transfers the handle to the remote process. The handle is left `Empty`. +// (Note: this occurs later, when the serialized message buffer is sent) +// 3. Message containing `SerializableValue` or `Empty` (depending on handle processing in step 2) +// is serialized and sent via IPC. +// 4. Message received and deserialized in target process. +// - Windows: `AssociateHandleForMessage::set_local_handle converts the received `SerializableValue` into `Owned`, ready for use. +// - Unix: Handle (with a new value in the target process) is received out-of-band via `recvmsg` +// and converted to `Owned` via `AssociateHandleForMessage::set_local_handle`. +#[derive(Debug)] +pub enum SerializableHandle { + // Owned handle, with optional target_pid on sending side. + Owned(PlatformHandle, Option<u32>), + // Transitional IPC states: + SerializableValue(PlatformHandleType), // Windows + Empty, // Unix +} + +// PlatformHandle is non-Send and contains a pointer (HANDLE) on Windows. +#[allow(clippy::non_send_fields_in_send_ty)] +unsafe impl Send for SerializableHandle {} + +impl SerializableHandle { + pub fn new(handle: PlatformHandle, target_pid: u32) -> SerializableHandle { + SerializableHandle::Owned(handle, Some(target_pid)) + } + + // Called on the receiving side to take ownership of the handle. + pub fn take_handle(&mut self) -> PlatformHandle { + match std::mem::replace(self, SerializableHandle::Empty) { + SerializableHandle::Owned(handle, target_pid) => { + assert!(target_pid.is_none()); + handle + } + _ => panic!("take_handle called in invalid state"), + } + } + + // Called on the sending side to take ownership of the handle for + // handling platform-specific remoting. + fn take_handle_for_send(&mut self) -> RemoteHandle { + match std::mem::replace(self, SerializableHandle::Empty) { + SerializableHandle::Owned(handle, target_pid) => unsafe { + RemoteHandle::new( + handle.into_raw(), + target_pid.expect("target process required"), + ) + }, + _ => panic!("take_handle_for_send called in invalid state"), + } + } + + fn new_owned(handle: PlatformHandleType) -> SerializableHandle { + SerializableHandle::Owned(PlatformHandle::new(handle), None) + } + + #[cfg(windows)] + fn make_owned(&mut self) { + if let SerializableHandle::SerializableValue(handle) = self { + *self = SerializableHandle::new_owned(*handle); + } else { + panic!("make_owned called in invalid state") + } + } + + fn new_serializable_value(handle: PlatformHandleType) -> SerializableHandle { + SerializableHandle::SerializableValue(handle) + } + + fn get_serializable_value(&self) -> PlatformHandleType { + match *self { + SerializableHandle::SerializableValue(handle) => handle, + SerializableHandle::Empty => INVALID_HANDLE_VALUE, + _ => panic!("get_remote_handle called in invalid state"), + } + } +} + +// Raw handle values are serialized as i64. Additional handling external to (de)serialization is required during IPC +// send/receive to convert these raw values into valid handles. +impl serde::Serialize for SerializableHandle { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: serde::Serializer, + { + let handle = self.get_serializable_value(); + serializer.serialize_i64(handle as i64) + } +} + +impl<'de> serde::Deserialize<'de> for SerializableHandle { + fn deserialize<D>(deserializer: D) -> Result<SerializableHandle, D::Error> + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_i64(SerializableHandleVisitor) + } +} + +struct SerializableHandleVisitor; +impl serde::de::Visitor<'_> for SerializableHandleVisitor { + type Value = SerializableHandle; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("an integer between -2^63 and 2^63") + } + + fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> + where + E: serde::de::Error, + { + Ok(SerializableHandle::new_serializable_value( + value as PlatformHandleType, + )) + } +} + +// Represents a PlatformHandle in-flight between processes. +// On Unix platforms, this is just a plain owned Handle, closed on drop. +// On Windows, `RemoteHandle` also retains ownership of the `target_handle` +// in the `target` process. Once the handle has been successfully sent +// to the remote, the sender should call `mark_sent()` to relinquish +// ownership of `target_handle` in the remote. +#[derive(Debug)] +pub struct RemoteHandle { + pub(crate) handle: PlatformHandleType, + #[cfg(windows)] + pub(crate) target: u32, + #[cfg(windows)] + pub(crate) target_handle: Option<PlatformHandleType>, +} + +impl RemoteHandle { + #[allow(clippy::missing_safety_doc)] + pub unsafe fn new(handle: PlatformHandleType, _target: u32) -> Self { + RemoteHandle { + handle, + #[cfg(windows)] + target: _target, + #[cfg(windows)] + target_handle: None, + } + } + + #[cfg(windows)] + pub fn mark_sent(&mut self) { + self.target_handle.take(); + } + + #[cfg(windows)] + #[allow(clippy::missing_safety_doc)] + pub unsafe fn send_to_target(&mut self) -> std::io::Result<PlatformHandleType> { + let target_handle = crate::duplicate_platform_handle(self.handle, Some(self.target))?; + self.target_handle = Some(target_handle); + Ok(target_handle) + } + + #[cfg(unix)] + #[allow(clippy::missing_safety_doc)] + pub unsafe fn take(self) -> PlatformHandleType { + let h = self.handle; + std::mem::forget(self); + h + } +} + +impl Drop for RemoteHandle { + fn drop(&mut self) { + unsafe { + crate::close_platform_handle(self.handle); + } + #[cfg(windows)] + unsafe { + if let Some(target_handle) = self.target_handle { + if let Err(e) = crate::close_target_handle(target_handle, self.target) { + trace!("RemoteHandle failed to close target handle: {:?}", e); + } + } + } + } +} + +unsafe impl Send for RemoteHandle {} + +pub trait AssociateHandleForMessage { + // True if this item has an associated handle attached for remoting. + fn has_associated_handle(&self) -> bool { + false + } + + // Take ownership of the associated handle, leaving the item's + // associated handle empty. + fn take_handle(&mut self) -> RemoteHandle { + panic!("take_handle called on item without associated handle"); + } + + #[allow(clippy::missing_safety_doc)] + // Replace an empty associated handle with a non-owning serializable value + // indicating the value of the handle in the remote process. + #[cfg(windows)] + unsafe fn set_remote_handle(&mut self, _: PlatformHandleType) { + panic!("set_remote_handle called on item without associated handle"); + } + + #[allow(clippy::missing_safety_doc)] + // Replace a serialized associated handle value with an owned local handle. + #[cfg(windows)] + unsafe fn set_local_handle(&mut self) { + panic!("set_local_handle called on item without associated handle"); + } + + #[allow(clippy::missing_safety_doc)] + // Replace an empty associated handle with an owned local handle. + #[cfg(unix)] + unsafe fn set_local_handle(&mut self, _: PlatformHandleType) { + panic!("set_local_handle called on item without associated handle"); + } +} + +impl AssociateHandleForMessage for ClientMessage { + fn has_associated_handle(&self) -> bool { + matches!( + *self, + ClientMessage::StreamCreated(_) + | ClientMessage::StreamInitialized(_) + | ClientMessage::ContextSetupDeviceCollectionCallback(_) + ) + } + + fn take_handle(&mut self) -> RemoteHandle { + match *self { + ClientMessage::StreamCreated(ref mut data) => data.shm_handle.take_handle_for_send(), + ClientMessage::StreamInitialized(ref mut data) => data.take_handle_for_send(), + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + data.platform_handle.take_handle_for_send() + } + _ => panic!("take_handle called on item without associated handle"), + } + } + + #[cfg(windows)] + unsafe fn set_remote_handle(&mut self, handle: PlatformHandleType) { + match *self { + ClientMessage::StreamCreated(ref mut data) => { + data.shm_handle = SerializableHandle::new_serializable_value(handle); + } + ClientMessage::StreamInitialized(ref mut data) => { + *data = SerializableHandle::new_serializable_value(handle); + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + data.platform_handle = SerializableHandle::new_serializable_value(handle); + } + _ => panic!("set_remote_handle called on item without associated handle"), + } + } + + #[cfg(windows)] + unsafe fn set_local_handle(&mut self) { + match *self { + ClientMessage::StreamCreated(ref mut data) => data.shm_handle.make_owned(), + ClientMessage::StreamInitialized(ref mut data) => data.make_owned(), + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + data.platform_handle.make_owned() + } + _ => panic!("set_local_handle called on item without associated handle"), + } + } + + #[cfg(unix)] + unsafe fn set_local_handle(&mut self, handle: PlatformHandleType) { + match *self { + ClientMessage::StreamCreated(ref mut data) => { + data.shm_handle = SerializableHandle::new_owned(handle); + } + ClientMessage::StreamInitialized(ref mut data) => { + *data = SerializableHandle::new_owned(handle); + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + data.platform_handle = SerializableHandle::new_owned(handle); + } + _ => panic!("set_local_handle called on item without associated handle"), + } + } +} + +impl AssociateHandleForMessage for ServerMessage {} + +impl AssociateHandleForMessage for DeviceCollectionReq {} +impl AssociateHandleForMessage for DeviceCollectionResp {} + +impl AssociateHandleForMessage for CallbackReq {} +impl AssociateHandleForMessage for CallbackResp {} + +#[cfg(test)] +mod test { + use super::StreamParams; + use cubeb::ffi; + use std::mem; + + #[test] + fn stream_params_size_check() { + assert_eq!( + mem::size_of::<StreamParams>(), + mem::size_of::<ffi::cubeb_stream_params>() + ) + } + + #[test] + fn stream_params_from() { + let raw = ffi::cubeb_stream_params { + format: ffi::CUBEB_SAMPLE_FLOAT32BE, + rate: 96_000, + channels: 32, + layout: ffi::CUBEB_LAYOUT_3F1_LFE, + prefs: ffi::CUBEB_STREAM_PREF_LOOPBACK, + }; + let wrapped = ::cubeb::StreamParams::from(raw); + let params = StreamParams::from(wrapped.as_ref()); + assert_eq!(params.format, raw.format); + assert_eq!(params.rate, raw.rate); + assert_eq!(params.channels, raw.channels); + assert_eq!(params.layout, raw.layout); + assert_eq!(params.prefs, raw.prefs); + } +} diff --git a/third_party/rust/audioipc2/src/rpccore.rs b/third_party/rust/audioipc2/src/rpccore.rs new file mode 100644 index 0000000000..69ed2e8cf0 --- /dev/null +++ b/third_party/rust/audioipc2/src/rpccore.rs @@ -0,0 +1,470 @@ +// Copyright © 2021 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crossbeam_queue::ArrayQueue; +use mio::Token; +use std::cell::UnsafeCell; +use std::collections::VecDeque; +use std::io::{self, Error, ErrorKind, Result}; +use std::marker::PhantomPinned; +use std::mem::ManuallyDrop; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Weak}; + +use crate::ipccore::EventLoopHandle; + +// This provides a safe-ish method for a thread to allocate +// stack storage space for a result, then pass a (wrapped) +// pointer to that location to another thread via +// a CompletionWriter to eventually store a result into. +struct Completion<T> { + item: UnsafeCell<Option<T>>, + writer: AtomicBool, + _pin: PhantomPinned, // disable rustc's no-alias +} + +impl<T> Completion<T> { + fn new() -> Self { + Completion { + item: UnsafeCell::new(None), + writer: AtomicBool::new(false), + _pin: PhantomPinned, + } + } + + // Wait until the writer completes, then return the result. + // This is intended to be a single-use function, once the writer + // has completed any further attempts to wait will return None. + fn wait(&self) -> Option<T> { + // Wait for the writer to complete or be dropped. + while self.writer.load(Ordering::Acquire) { + std::thread::park(); + } + unsafe { (*self.item.get()).take() } + } + + // Create a writer for the other thread to store the + // expected result into. + fn writer(&self) -> CompletionWriter<T> { + assert!(!self.writer.load(Ordering::Relaxed)); + self.writer.store(true, Ordering::Release); + CompletionWriter { + ptr: self as *const _ as *mut _, + waiter: std::thread::current(), + } + } +} + +impl<T> Drop for Completion<T> { + fn drop(&mut self) { + // Wait for the outstanding writer to complete before + // dropping, since the CompletionWriter references + // memory owned by this object. + while self.writer.load(Ordering::Acquire) { + std::thread::park(); + } + } +} + +struct CompletionWriter<T> { + ptr: *mut Completion<T>, // Points to a Completion on another thread's stack + waiter: std::thread::Thread, // Identifies thread waiting for completion +} + +impl<T> CompletionWriter<T> { + fn set(self, value: T) { + // Store the result into the Completion's memory. + // Since `set` consumes `self`, rely on `Drop` to + // mark the writer as done and wake the Completion's + // thread. + unsafe { + assert!((*self.ptr).writer.load(Ordering::Relaxed)); + *(*self.ptr).item.get() = Some(value); + } + } +} + +impl<T> Drop for CompletionWriter<T> { + fn drop(&mut self) { + // Mark writer as complete - if `set` was not called, + // the waiter will receive `None`. + unsafe { + (*self.ptr).writer.store(false, Ordering::Release); + } + // Wake the Completion's thread. + self.waiter.unpark(); + } +} + +// Safety: CompletionWriter holds a pointer to a Completion +// residing on another thread's stack. The Completion always +// waits for an outstanding writer if present, and CompletionWriter +// releases the waiter and wakes the Completion's thread on drop, +// so this pointer will always be live for the duration of a +// CompletionWriter. +unsafe impl<T> Send for CompletionWriter<T> {} + +// RPC message handler. Implemented by ClientHandler (for Client) +// and ServerHandler (for Server). +pub(crate) trait Handler { + type In; + type Out; + + // Consume a request + fn consume(&mut self, request: Self::In) -> Result<()>; + + // Produce a response + fn produce(&mut self) -> Result<Option<Self::Out>>; +} + +// Client RPC definition. This supplies the expected message +// request and response types. +pub trait Client { + type ServerMessage; + type ClientMessage; +} + +// Server RPC definition. This supplies the expected message +// request and response types. `process` is passed inbound RPC +// requests by the ServerHandler to be responded to by the server. +pub trait Server { + type ServerMessage; + type ClientMessage; + + fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage; +} + +// RPC Client Proxy implementation. +type ProxyRequest<Request, Response> = (Request, CompletionWriter<Response>); + +// RPC Proxy that may be `clone`d for use by multiple owners/threads. +// A Proxy `call` arranges for the supplied request to be transmitted +// to the associated Server via RPC and blocks awaiting the response +// via the associated `Completion`. +// A ClientHandler normally lives until the last Proxy is dropped, but if the ClientHandler +// encounters an internal error, `requests` will fail to upgrade, allowing +// the proxy to report an error. +#[derive(Debug)] +pub struct Proxy<Request, Response> { + handle: Option<(EventLoopHandle, Token)>, + requests: ManuallyDrop<RequestQueueSender<ProxyRequest<Request, Response>>>, +} + +impl<Request, Response> Proxy<Request, Response> { + fn new(requests: RequestQueueSender<ProxyRequest<Request, Response>>) -> Self { + Self { + handle: None, + requests: ManuallyDrop::new(requests), + } + } + + pub fn call(&self, request: Request) -> Result<Response> { + let response = Completion::new(); + self.requests.push((request, response.writer()))?; + self.wake_connection(); + match response.wait() { + Some(resp) => Ok(resp), + None => Err(Error::new(ErrorKind::Other, "proxy recv error")), + } + } + + pub(crate) fn connect_event_loop(&mut self, handle: EventLoopHandle, token: Token) { + self.handle = Some((handle, token)); + } + + fn wake_connection(&self) { + let (handle, token) = self + .handle + .as_ref() + .expect("proxy not connected to event loop"); + handle.wake_connection(*token); + } +} + +impl<Request, Response> Clone for Proxy<Request, Response> { + fn clone(&self) -> Self { + let mut clone = Self::new((*self.requests).clone()); + let (handle, token) = self + .handle + .as_ref() + .expect("proxy not connected to event loop"); + clone.connect_event_loop(handle.clone(), *token); + clone + } +} + +impl<Request, Response> Drop for Proxy<Request, Response> { + fn drop(&mut self) { + trace!("Proxy drop, waking EventLoop"); + // Must drop `requests` before waking the connection, otherwise + // the wake may be processed before the (last) weak reference is + // dropped. + let last_proxy = self.requests.live_proxies(); + unsafe { + ManuallyDrop::drop(&mut self.requests); + } + if last_proxy == 1 && self.handle.is_some() { + self.wake_connection() + } + } +} + +const RPC_CLIENT_INITIAL_PROXIES: usize = 32; // Initial proxy pre-allocation per client. + +// Client-specific Handler implementation. +// The IPC EventLoop Driver calls this to execute client-specific +// RPC handling. Serialized messages sent via a Proxy are queued +// for transmission when `produce` is called. +// Deserialized messages are passed via `consume` to +// trigger response completion by sending the response via a channel +// connected to a ProxyResponse. +pub(crate) struct ClientHandler<C: Client> { + in_flight: VecDeque<CompletionWriter<C::ClientMessage>>, + requests: Arc<RequestQueue<ProxyRequest<C::ServerMessage, C::ClientMessage>>>, +} + +impl<C: Client> ClientHandler<C> { + fn new( + requests: Arc<RequestQueue<ProxyRequest<C::ServerMessage, C::ClientMessage>>>, + ) -> ClientHandler<C> { + ClientHandler::<C> { + in_flight: VecDeque::with_capacity(RPC_CLIENT_INITIAL_PROXIES), + requests, + } + } +} + +impl<C: Client> Handler for ClientHandler<C> { + type In = C::ClientMessage; + type Out = C::ServerMessage; + + fn consume(&mut self, response: Self::In) -> Result<()> { + trace!("ClientHandler::consume"); + if let Some(response_writer) = self.in_flight.pop_front() { + response_writer.set(response); + } else { + return Err(Error::new(ErrorKind::Other, "request/response mismatch")); + } + + Ok(()) + } + + fn produce(&mut self) -> Result<Option<Self::Out>> { + trace!("ClientHandler::produce"); + + // If the weak count is zero, no proxies are attached and + // no further proxies can be attached since every proxy + // after the initial one is cloned from an existing instance. + self.requests.check_live_proxies()?; + // Try to get a new message + match self.requests.pop() { + Some((request, response_writer)) => { + trace!(" --> received request"); + self.in_flight.push_back(response_writer); + Ok(Some(request)) + } + None => { + trace!(" --> no request"); + Ok(None) + } + } + } +} + +#[derive(Debug)] +pub(crate) struct RequestQueue<T> { + queue: ArrayQueue<T>, +} + +impl<T> RequestQueue<T> { + pub(crate) fn new(size: usize) -> Self { + RequestQueue { + queue: ArrayQueue::new(size), + } + } + + pub(crate) fn pop(&self) -> Option<T> { + self.queue.pop() + } + + pub(crate) fn new_sender(self: &Arc<Self>) -> RequestQueueSender<T> { + RequestQueueSender { + inner: Arc::downgrade(self), + } + } + + pub(crate) fn check_live_proxies(self: &Arc<Self>) -> Result<()> { + if Arc::weak_count(self) == 0 { + return Err(io::ErrorKind::ConnectionAborted.into()); + } + Ok(()) + } +} + +pub(crate) struct RequestQueueSender<T> { + inner: Weak<RequestQueue<T>>, +} + +impl<T> RequestQueueSender<T> { + pub(crate) fn push(&self, request: T) -> Result<()> { + if let Some(consumer) = self.inner.upgrade() { + if consumer.queue.push(request).is_err() { + debug!("Proxy[{:p}]: call failed - CH::requests full", self); + return Err(io::ErrorKind::ConnectionAborted.into()); + } + return Ok(()); + } + debug!("Proxy[{:p}]: call failed - CH::requests dropped", self); + Err(Error::new(ErrorKind::Other, "proxy send error")) + } + + pub(crate) fn live_proxies(&self) -> usize { + Weak::weak_count(&self.inner) + } +} + +impl<T> Clone for RequestQueueSender<T> { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + } + } +} + +impl<T> std::fmt::Debug for RequestQueueSender<T> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RequestQueueProducer") + .field("inner", &self.inner.as_ptr()) + .finish() + } +} + +#[allow(clippy::type_complexity)] +pub(crate) fn make_client<C: Client>( +) -> Result<(ClientHandler<C>, Proxy<C::ServerMessage, C::ClientMessage>)> { + let requests = Arc::new(RequestQueue::new(RPC_CLIENT_INITIAL_PROXIES)); + let proxy_req = requests.new_sender(); + let handler = ClientHandler::new(requests); + + Ok((handler, Proxy::new(proxy_req))) +} + +// Server-specific Handler implementation. +// The IPC EventLoop Driver calls this to execute server-specific +// RPC handling. Deserialized messages are passed via `consume` to the +// associated `server` for processing. Server responses are then queued +// for RPC to the associated client when `produce` is called. +pub(crate) struct ServerHandler<S: Server> { + server: S, + in_flight: VecDeque<S::ClientMessage>, +} + +impl<S: Server> Handler for ServerHandler<S> { + type In = S::ServerMessage; + type Out = S::ClientMessage; + + fn consume(&mut self, message: Self::In) -> Result<()> { + trace!("ServerHandler::consume"); + let response = self.server.process(message); + self.in_flight.push_back(response); + Ok(()) + } + + fn produce(&mut self) -> Result<Option<Self::Out>> { + trace!("ServerHandler::produce"); + + // Return the ready response + match self.in_flight.pop_front() { + Some(res) => { + trace!(" --> received response"); + Ok(Some(res)) + } + None => { + trace!(" --> no response ready"); + Ok(None) + } + } + } +} + +const RPC_SERVER_INITIAL_CLIENTS: usize = 32; // Initial client allocation per server. + +pub(crate) fn make_server<S: Server>(server: S) -> ServerHandler<S> { + ServerHandler::<S> { + server, + in_flight: VecDeque::with_capacity(RPC_SERVER_INITIAL_CLIENTS), + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn basic() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + assert!(queue.pop().is_none()); + producer.push(1).unwrap(); + assert!(queue.pop().is_some()); + assert!(queue.pop().is_none()); + } + + #[test] + fn queue_dropped() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + drop(queue); + assert!(producer.push(1).is_err()); + } + + #[test] + fn queue_full() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + producer.push(1).unwrap(); + assert!(producer.push(2).is_err()); + } + + #[test] + fn queue_producer_clone() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + let producer2 = producer.clone(); + producer.push(1).unwrap(); + assert!(producer2.push(2).is_err()); + } + + #[test] + fn queue_producer_drop() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + let producer2 = producer.clone(); + drop(producer); + assert!(producer2.push(2).is_ok()); + } + + #[test] + fn queue_producer_weak() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + let producer2 = producer.clone(); + drop(queue); + assert!(producer2.push(2).is_err()); + } + + #[test] + fn queue_producer_shutdown() { + let queue = Arc::new(RequestQueue::new(1)); + let producer = queue.new_sender(); + let producer2 = producer.clone(); + producer.push(1).unwrap(); + assert!(Arc::weak_count(&queue) == 2); + drop(producer); + assert!(Arc::weak_count(&queue) == 1); + drop(producer2); + assert!(Arc::weak_count(&queue) == 0); + } +} diff --git a/third_party/rust/audioipc2/src/shm.rs b/third_party/rust/audioipc2/src/shm.rs new file mode 100644 index 0000000000..2aaa92ef36 --- /dev/null +++ b/third_party/rust/audioipc2/src/shm.rs @@ -0,0 +1,334 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +#![allow(clippy::missing_safety_doc)] + +use crate::errors::*; +use crate::PlatformHandle; +use std::{convert::TryInto, ffi::c_void, slice}; + +#[cfg(unix)] +pub use unix::SharedMem; +#[cfg(windows)] +pub use windows::SharedMem; + +#[derive(Copy, Clone)] +struct SharedMemView { + ptr: *mut c_void, + size: usize, +} + +unsafe impl Send for SharedMemView {} + +impl SharedMemView { + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + let map = slice::from_raw_parts(self.ptr as _, self.size); + if size <= self.size { + Ok(&map[..size]) + } else { + bail!("mmap size"); + } + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + let map = slice::from_raw_parts_mut(self.ptr as _, self.size); + if size <= self.size { + Ok(&mut map[..size]) + } else { + bail!("mmap size") + } + } +} + +#[cfg(unix)] +mod unix { + use super::*; + use memmap2::{MmapMut, MmapOptions}; + use std::fs::File; + use std::os::unix::io::{AsRawFd, FromRawFd}; + + #[cfg(target_os = "android")] + fn open_shm_file(_id: &str, size: usize) -> Result<File> { + unsafe { + let fd = ashmem::ASharedMemory_create(std::ptr::null(), size); + if fd >= 0 { + // Drop PROT_EXEC + let r = ashmem::ASharedMemory_setProt(fd, libc::PROT_READ | libc::PROT_WRITE); + assert_eq!(r, 0); + return Ok(File::from_raw_fd(fd.try_into().unwrap())); + } + Err(std::io::Error::last_os_error().into()) + } + } + + #[cfg(not(target_os = "android"))] + fn open_shm_file(id: &str, size: usize) -> Result<File> { + let file = open_shm_file_impl(id)?; + allocate_file(&file, size)?; + Ok(file) + } + + #[cfg(not(target_os = "android"))] + fn open_shm_file_impl(id: &str) -> Result<File> { + use std::env::temp_dir; + use std::fs::{remove_file, OpenOptions}; + + let id_cstring = std::ffi::CString::new(id).unwrap(); + + #[cfg(target_os = "linux")] + { + unsafe { + let r = libc::syscall(libc::SYS_memfd_create, id_cstring.as_ptr(), 0); + if r >= 0 { + return Ok(File::from_raw_fd(r.try_into().unwrap())); + } + } + + let mut path = std::path::PathBuf::from("/dev/shm"); + path.push(id); + + if let Ok(file) = OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .open(&path) + { + let _ = remove_file(&path); + return Ok(file); + } + } + + unsafe { + let fd = libc::shm_open( + id_cstring.as_ptr(), + libc::O_RDWR | libc::O_CREAT | libc::O_EXCL, + 0o600, + ); + if fd >= 0 { + libc::shm_unlink(id_cstring.as_ptr()); + return Ok(File::from_raw_fd(fd)); + } + } + + let mut path = temp_dir(); + path.push(id); + + let file = OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .open(&path)?; + + let _ = remove_file(&path); + Ok(file) + } + + #[cfg(not(target_os = "android"))] + fn handle_enospc(s: &str) -> Result<()> { + let err = std::io::Error::last_os_error(); + let errno = err.raw_os_error().unwrap_or(0); + assert_ne!(errno, 0); + debug!("allocate_file: {} failed errno={}", s, errno); + if errno == libc::ENOSPC { + return Err(err.into()); + } + Ok(()) + } + + #[cfg(not(target_os = "android"))] + fn allocate_file(file: &File, size: usize) -> Result<()> { + // First, set the file size. This may create a sparse file on + // many systems, which can fail with SIGBUS when accessed via a + // mapping and the lazy backing allocation fails due to low disk + // space. To avoid this, try to force the entire file to be + // preallocated before mapping using OS-specific approaches below. + + file.set_len(size.try_into().unwrap())?; + + let fd = file.as_raw_fd(); + let size: libc::off_t = size.try_into().unwrap(); + + // Try Linux-specific fallocate. + #[cfg(target_os = "linux")] + { + if unsafe { libc::fallocate(fd, 0, 0, size) } == 0 { + return Ok(()); + } + handle_enospc("fallocate()")?; + } + + // Try macOS-specific fcntl. + #[cfg(target_os = "macos")] + { + let params = libc::fstore_t { + fst_flags: libc::F_ALLOCATEALL, + fst_posmode: libc::F_PEOFPOSMODE, + fst_offset: 0, + fst_length: size, + fst_bytesalloc: 0, + }; + if unsafe { libc::fcntl(fd, libc::F_PREALLOCATE, ¶ms) } == 0 { + return Ok(()); + } + handle_enospc("fcntl(F_PREALLOCATE)")?; + } + + // Fall back to portable version, where available. + #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "dragonfly"))] + { + if unsafe { libc::posix_fallocate(fd, 0, size) } == 0 { + return Ok(()); + } + handle_enospc("posix_fallocate()")?; + } + + Ok(()) + } + + pub struct SharedMem { + file: File, + _mmap: MmapMut, + view: SharedMemView, + } + + impl SharedMem { + pub fn new(id: &str, size: usize) -> Result<SharedMem> { + let file = open_shm_file(id, size)?; + let mut mmap = unsafe { MmapOptions::new().len(size).map_mut(&file)? }; + assert_eq!(mmap.len(), size); + let view = SharedMemView { + ptr: mmap.as_mut_ptr() as _, + size, + }; + Ok(SharedMem { + file, + _mmap: mmap, + view, + }) + } + + pub unsafe fn make_handle(&self) -> Result<PlatformHandle> { + PlatformHandle::duplicate(self.file.as_raw_fd()).map_err(|e| e.into()) + } + + pub unsafe fn from(handle: PlatformHandle, size: usize) -> Result<SharedMem> { + let file = File::from_raw_fd(handle.into_raw()); + let mut mmap = MmapOptions::new().len(size).map_mut(&file)?; + assert_eq!(mmap.len(), size); + let view = SharedMemView { + ptr: mmap.as_mut_ptr() as _, + size, + }; + Ok(SharedMem { + file, + _mmap: mmap, + view, + }) + } + + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + self.view.get_slice(size) + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + self.view.get_mut_slice(size) + } + + pub fn get_size(&self) -> usize { + self.view.size + } + } +} + +#[cfg(windows)] +mod windows { + use super::*; + use std::ptr; + use winapi::{ + shared::{minwindef::DWORD, ntdef::HANDLE}, + um::{ + handleapi::CloseHandle, + memoryapi::{MapViewOfFile, UnmapViewOfFile, FILE_MAP_ALL_ACCESS}, + winbase::CreateFileMappingA, + winnt::PAGE_READWRITE, + }, + }; + + use crate::INVALID_HANDLE_VALUE; + + pub struct SharedMem { + handle: HANDLE, + view: SharedMemView, + } + + unsafe impl Send for SharedMem {} + + impl Drop for SharedMem { + fn drop(&mut self) { + unsafe { + let ok = UnmapViewOfFile(self.view.ptr); + assert_ne!(ok, 0); + let ok = CloseHandle(self.handle); + assert_ne!(ok, 0); + } + } + } + + impl SharedMem { + pub fn new(_id: &str, size: usize) -> Result<SharedMem> { + unsafe { + let handle = CreateFileMappingA( + INVALID_HANDLE_VALUE, + ptr::null_mut(), + PAGE_READWRITE, + (size as u64 >> 32).try_into().unwrap(), + (size as u64 & (DWORD::MAX as u64)).try_into().unwrap(), + ptr::null(), + ); + if handle.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + + let ptr = MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, size); + if ptr.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + + Ok(SharedMem { + handle, + view: SharedMemView { ptr, size }, + }) + } + } + + pub unsafe fn make_handle(&self) -> Result<PlatformHandle> { + PlatformHandle::duplicate(self.handle).map_err(|e| e.into()) + } + + pub unsafe fn from(handle: PlatformHandle, size: usize) -> Result<SharedMem> { + let handle = handle.into_raw(); + let ptr = MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, size); + if ptr.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + Ok(SharedMem { + handle, + view: SharedMemView { ptr, size }, + }) + } + + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + self.view.get_slice(size) + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + self.view.get_mut_slice(size) + } + + pub fn get_size(&self) -> usize { + self.view.size + } + } +} diff --git a/third_party/rust/audioipc2/src/sys/mod.rs b/third_party/rust/audioipc2/src/sys/mod.rs new file mode 100644 index 0000000000..0bcfdaa15e --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/mod.rs @@ -0,0 +1,77 @@ +// Copyright © 2021 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use std::{collections::VecDeque, io::Result}; + +use bytes::BytesMut; +use mio::{event::Source, Interest, Registry, Token}; + +#[cfg(unix)] +mod unix; +use crate::messages::RemoteHandle; + +#[cfg(unix)] +pub use self::unix::*; + +#[cfg(windows)] +mod windows; +#[cfg(windows)] +pub use self::windows::*; + +impl Source for Pipe { + fn register(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> { + self.io.register(registry, token, interests) + } + + fn reregister(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> { + self.io.reregister(registry, token, interests) + } + + fn deregister(&mut self, registry: &Registry) -> Result<()> { + self.io.deregister(registry) + } +} + +const HANDLE_QUEUE_LIMIT: usize = 16; + +#[derive(Debug)] +pub struct ConnectionBuffer { + pub buf: BytesMut, + handles: VecDeque<RemoteHandle>, +} + +impl ConnectionBuffer { + pub fn with_capacity(cap: usize) -> Self { + ConnectionBuffer { + buf: BytesMut::with_capacity(cap), + handles: VecDeque::with_capacity(HANDLE_QUEUE_LIMIT), + } + } + + pub fn is_empty(&self) -> bool { + self.buf.is_empty() + } + + pub fn push_handle(&mut self, handle: RemoteHandle) { + assert!(self.handles.len() < self.handles.capacity()); + self.handles.push_back(handle) + } + + pub fn pop_handle(&mut self) -> Option<RemoteHandle> { + self.handles.pop_front() + } +} + +pub trait RecvMsg { + // Receive data from the associated connection. `recv_msg` expects the capacity of + // the `ConnectionBuffer` members have been adjusted appropriately by the caller. + fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize>; +} + +pub trait SendMsg { + // Send data on the associated connection. `send_msg` consumes and adjusts the length of the + // `ConnectionBuffer` members based on the size of the successful send operation. + fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize>; +} diff --git a/third_party/rust/audioipc2/src/sys/unix/cmsg.rs b/third_party/rust/audioipc2/src/sys/unix/cmsg.rs new file mode 100644 index 0000000000..581df9a847 --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/unix/cmsg.rs @@ -0,0 +1,104 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::sys::HANDLE_QUEUE_LIMIT; +use bytes::{BufMut, BytesMut}; +use libc::{self, cmsghdr}; +use std::convert::TryInto; +use std::os::unix::io::RawFd; +use std::{mem, slice}; + +trait AsBytes { + fn as_bytes(&self) -> &[u8]; +} + +impl<'a, T: Sized> AsBytes for &'a [T] { + fn as_bytes(&self) -> &[u8] { + // TODO: This should account for the alignment of T + let byte_count = self.len() * mem::size_of::<T>(); + unsafe { slice::from_raw_parts(self.as_ptr() as *const _, byte_count) } + } +} + +// Encode `handles` into a cmsghdr in `buf`. +pub fn encode_handles(cmsg: &mut BytesMut, handles: &[RawFd]) { + assert!(handles.len() <= HANDLE_QUEUE_LIMIT); + let msg = handles.as_bytes(); + + let cmsg_space = space(msg.len()); + assert!(cmsg.remaining_mut() >= cmsg_space); + + // Some definitions of cmsghdr contain padding. Rather + // than try to keep an up-to-date #cfg list to handle + // that, just use a pre-zeroed struct to fill out any + // fields we don't care about. + let zeroed = unsafe { mem::zeroed() }; + #[allow(clippy::needless_update)] + // `cmsg_len` is `usize` on some platforms, `u32` on others. + #[allow(clippy::useless_conversion)] + let cmsghdr = cmsghdr { + cmsg_len: len(msg.len()).try_into().unwrap(), + cmsg_level: libc::SOL_SOCKET, + cmsg_type: libc::SCM_RIGHTS, + ..zeroed + }; + + unsafe { + let cmsghdr_ptr = cmsg.chunk_mut().as_mut_ptr(); + std::ptr::copy_nonoverlapping( + &cmsghdr as *const _ as *const _, + cmsghdr_ptr, + mem::size_of::<cmsghdr>(), + ); + let cmsg_data_ptr = libc::CMSG_DATA(cmsghdr_ptr as _); + std::ptr::copy_nonoverlapping(msg.as_ptr(), cmsg_data_ptr, msg.len()); + cmsg.advance_mut(cmsg_space); + } +} + +// Decode `buf` containing a cmsghdr with one or more handle(s). +pub fn decode_handles(buf: &mut BytesMut) -> arrayvec::ArrayVec<RawFd, HANDLE_QUEUE_LIMIT> { + let mut fds = arrayvec::ArrayVec::<RawFd, HANDLE_QUEUE_LIMIT>::new(); + + let cmsghdr_len = len(0); + + if buf.len() < cmsghdr_len { + // No more entries---not enough data in `buf` for a + // complete message. + return fds; + } + + let cmsg: &cmsghdr = unsafe { &*(buf.as_ptr() as *const _) }; + #[allow(clippy::unnecessary_cast)] // `cmsg_len` type is platform-dependent. + let cmsg_len = cmsg.cmsg_len as usize; + + match (cmsg.cmsg_level, cmsg.cmsg_type) { + (libc::SOL_SOCKET, libc::SCM_RIGHTS) => { + trace!("Found SCM_RIGHTS..."); + let slice = &buf[cmsghdr_len..cmsg_len]; + let slice = unsafe { + slice::from_raw_parts( + slice.as_ptr() as *const _, + slice.len() / mem::size_of::<i32>(), + ) + }; + fds.try_extend_from_slice(slice).unwrap(); + } + (level, kind) => { + trace!("Skipping cmsg level, {}, type={}...", level, kind); + } + } + + assert!(fds.len() <= HANDLE_QUEUE_LIMIT); + fds +} + +fn len(len: usize) -> usize { + unsafe { libc::CMSG_LEN(len.try_into().unwrap()) as usize } +} + +pub fn space(len: usize) -> usize { + unsafe { libc::CMSG_SPACE(len.try_into().unwrap()) as usize } +} diff --git a/third_party/rust/audioipc2/src/sys/unix/cmsghdr.c b/third_party/rust/audioipc2/src/sys/unix/cmsghdr.c new file mode 100644 index 0000000000..82d7852867 --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/unix/cmsghdr.c @@ -0,0 +1,23 @@ +#include <sys/socket.h> +#include <inttypes.h> +#include <string.h> + +const uint8_t* +cmsghdr_bytes(size_t* size) +{ + int myfd = 0; + + static union { + uint8_t buf[CMSG_SPACE(sizeof(myfd))]; + struct cmsghdr align; + } u; + + u.align.cmsg_len = CMSG_LEN(sizeof(myfd)); + u.align.cmsg_level = SOL_SOCKET; + u.align.cmsg_type = SCM_RIGHTS; + + memcpy(CMSG_DATA(&u.align), &myfd, sizeof(myfd)); + + *size = sizeof(u); + return (const uint8_t*)&u.buf; +} diff --git a/third_party/rust/audioipc2/src/sys/unix/mod.rs b/third_party/rust/audioipc2/src/sys/unix/mod.rs new file mode 100644 index 0000000000..84f3f1edf2 --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/unix/mod.rs @@ -0,0 +1,126 @@ +// Copyright © 2021 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use std::io::Result; +use std::os::unix::prelude::{AsRawFd, FromRawFd}; + +use bytes::{Buf, BufMut, BytesMut}; +use iovec::IoVec; +use mio::net::UnixStream; + +use crate::PlatformHandle; + +use super::{ConnectionBuffer, RecvMsg, SendMsg, HANDLE_QUEUE_LIMIT}; + +pub mod cmsg; +mod msg; + +pub struct Pipe { + pub(crate) io: UnixStream, + cmsg: BytesMut, +} + +impl Pipe { + fn new(io: UnixStream) -> Self { + Pipe { + io, + cmsg: BytesMut::with_capacity(cmsg::space( + std::mem::size_of::<i32>() * HANDLE_QUEUE_LIMIT, + )), + } + } +} + +// Create a connected "pipe" pair. The `Pipe` is the server end, +// the `PlatformHandle` is the client end to be remoted. +pub fn make_pipe_pair() -> Result<(Pipe, PlatformHandle)> { + let (server, client) = UnixStream::pair()?; + Ok((Pipe::new(server), PlatformHandle::from(client))) +} + +impl Pipe { + #[allow(clippy::missing_safety_doc)] + pub unsafe fn from_raw_handle(handle: crate::PlatformHandle) -> Pipe { + Pipe::new(UnixStream::from_raw_fd(handle.into_raw())) + } + + pub fn shutdown(&mut self) -> Result<()> { + self.io.shutdown(std::net::Shutdown::Both) + } +} + +impl RecvMsg for Pipe { + // Receive data (and fds) from the associated connection. `recv_msg` expects the capacity of + // the `ConnectionBuffer` members has been adjusted appropriate by the caller. + fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> { + assert!(buf.buf.remaining_mut() > 0); + // TODO: MSG_CMSG_CLOEXEC not portable. + // TODO: MSG_NOSIGNAL not portable; macOS can set socket option SO_NOSIGPIPE instead. + #[cfg(target_os = "linux")] + let flags = libc::MSG_CMSG_CLOEXEC | libc::MSG_NOSIGNAL; + #[cfg(not(target_os = "linux"))] + let flags = 0; + let r = unsafe { + let chunk = buf.buf.chunk_mut(); + let slice = std::slice::from_raw_parts_mut(chunk.as_mut_ptr(), chunk.len()); + let mut iovec = [<&mut IoVec>::from(slice)]; + msg::recv_msg_with_flags( + self.io.as_raw_fd(), + &mut iovec, + self.cmsg.chunk_mut(), + flags, + ) + }; + match r { + Ok((n, cmsg_n, msg_flags)) => unsafe { + trace!("recv_msg_with_flags flags={}", msg_flags); + buf.buf.advance_mut(n); + self.cmsg.advance_mut(cmsg_n); + let handles = cmsg::decode_handles(&mut self.cmsg); + self.cmsg.clear(); + let unused = 0; + for h in handles { + buf.push_handle(super::RemoteHandle::new(h, unused)); + } + Ok(n) + }, + Err(e) => Err(e), + } + } +} + +impl SendMsg for Pipe { + // Send data (and fds) on the associated connection. `send_msg` adjusts the length of the + // `ConnectionBuffer` members based on the size of the successful send operation. + fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> { + assert!(!buf.buf.is_empty()); + if !buf.handles.is_empty() { + let mut handles = [-1i32; HANDLE_QUEUE_LIMIT]; + for (i, h) in buf.handles.iter().enumerate() { + handles[i] = h.handle; + } + cmsg::encode_handles(&mut self.cmsg, &handles[..buf.handles.len()]); + } + let r = { + // TODO: MSG_NOSIGNAL not portable; macOS can set socket option SO_NOSIGPIPE instead. + #[cfg(target_os = "linux")] + let flags = libc::MSG_NOSIGNAL; + #[cfg(not(target_os = "linux"))] + let flags = 0; + let iovec = [<&IoVec>::from(&buf.buf[..buf.buf.len()])]; + msg::send_msg_with_flags(self.io.as_raw_fd(), &iovec, &self.cmsg, flags) + }; + match r { + Ok(n) => { + buf.buf.advance(n); + // Discard sent handles. + while buf.handles.pop_front().is_some() {} + self.cmsg.clear(); + Ok(n) + } + Err(e) => Err(e), + } + } +} diff --git a/third_party/rust/audioipc2/src/sys/unix/msg.rs b/third_party/rust/audioipc2/src/sys/unix/msg.rs new file mode 100644 index 0000000000..c2cd353289 --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/unix/msg.rs @@ -0,0 +1,82 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +use bytes::buf::UninitSlice; +use iovec::unix; +use iovec::IoVec; +use std::os::unix::io::RawFd; +use std::{cmp, io, mem, ptr}; + +fn cvt(r: libc::ssize_t) -> io::Result<usize> { + if r == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(r as usize) + } +} + +// Convert return of -1 into error message, handling retry on EINTR +fn cvt_r<F: FnMut() -> libc::ssize_t>(mut f: F) -> io::Result<usize> { + loop { + match cvt(f()) { + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} + other => return other, + } + } +} + +pub(crate) fn recv_msg_with_flags( + socket: RawFd, + bufs: &mut [&mut IoVec], + cmsg: &mut UninitSlice, + flags: libc::c_int, +) -> io::Result<(usize, usize, libc::c_int)> { + let slice = unix::as_os_slice_mut(bufs); + let len = cmp::min(<libc::c_int>::max_value() as usize, slice.len()); + let (control, controllen) = if cmsg.len() == 0 { + (ptr::null_mut(), 0) + } else { + (cmsg.as_mut_ptr() as _, cmsg.len()) + }; + + let mut msghdr: libc::msghdr = unsafe { mem::zeroed() }; + msghdr.msg_name = ptr::null_mut(); + msghdr.msg_namelen = 0; + msghdr.msg_iov = slice.as_mut_ptr(); + msghdr.msg_iovlen = len as _; + msghdr.msg_control = control; + msghdr.msg_controllen = controllen as _; + + let n = cvt_r(|| unsafe { libc::recvmsg(socket, &mut msghdr as *mut _, flags) })?; + + #[allow(clippy::unnecessary_cast)] // `msg_controllen` type is platform-dependent. + let controllen = msghdr.msg_controllen as usize; + Ok((n, controllen, msghdr.msg_flags)) +} + +pub(crate) fn send_msg_with_flags( + socket: RawFd, + bufs: &[&IoVec], + cmsg: &[u8], + flags: libc::c_int, +) -> io::Result<usize> { + let slice = unix::as_os_slice(bufs); + let len = cmp::min(<libc::c_int>::max_value() as usize, slice.len()); + let (control, controllen) = if cmsg.is_empty() { + (ptr::null_mut(), 0) + } else { + (cmsg.as_ptr() as *mut _, cmsg.len()) + }; + + let mut msghdr: libc::msghdr = unsafe { mem::zeroed() }; + msghdr.msg_name = ptr::null_mut(); + msghdr.msg_namelen = 0; + msghdr.msg_iov = slice.as_ptr() as *mut _; + msghdr.msg_iovlen = len as _; + msghdr.msg_control = control; + msghdr.msg_controllen = controllen as _; + + cvt_r(|| unsafe { libc::sendmsg(socket, &msghdr as *const _, flags) }) +} diff --git a/third_party/rust/audioipc2/src/sys/windows/mod.rs b/third_party/rust/audioipc2/src/sys/windows/mod.rs new file mode 100644 index 0000000000..973e9608d1 --- /dev/null +++ b/third_party/rust/audioipc2/src/sys/windows/mod.rs @@ -0,0 +1,102 @@ +// Copyright © 2021 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use std::{ + fs::OpenOptions, + io::{Read, Write}, + os::windows::prelude::{FromRawHandle, OpenOptionsExt}, + sync::atomic::{AtomicUsize, Ordering}, +}; + +use std::io::Result; + +use bytes::{Buf, BufMut}; +use mio::windows::NamedPipe; +use winapi::um::winbase::FILE_FLAG_OVERLAPPED; + +use crate::PlatformHandle; + +use super::{ConnectionBuffer, RecvMsg, SendMsg}; + +pub struct Pipe { + pub(crate) io: NamedPipe, +} + +// Create a connected "pipe" pair. The `Pipe` is the server end, +// the `PlatformHandle` is the client end to be remoted. +pub fn make_pipe_pair() -> Result<(Pipe, PlatformHandle)> { + let pipe_name = get_pipe_name(); + let server = NamedPipe::new(&pipe_name)?; + + let client = { + let mut opts = OpenOptions::new(); + opts.read(true) + .write(true) + .custom_flags(FILE_FLAG_OVERLAPPED); + let file = opts.open(&pipe_name)?; + PlatformHandle::from(file) + }; + + Ok((Pipe::new(server), client)) +} + +static PIPE_ID: AtomicUsize = AtomicUsize::new(0); + +fn get_pipe_name() -> String { + let pid = std::process::id(); + let pipe_id = PIPE_ID.fetch_add(1, Ordering::Relaxed); + format!("\\\\.\\pipe\\LOCAL\\cubeb-pipe-{pid}-{pipe_id}") +} + +impl Pipe { + pub fn new(io: NamedPipe) -> Self { + Self { io } + } + + #[allow(clippy::missing_safety_doc)] + pub unsafe fn from_raw_handle(handle: crate::PlatformHandle) -> Pipe { + Pipe::new(NamedPipe::from_raw_handle(handle.into_raw())) + } + + pub fn shutdown(&mut self) -> Result<()> { + self.io.disconnect() + } +} + +impl RecvMsg for Pipe { + // Receive data from the associated connection. `recv_msg` expects the capacity of + // the `ConnectionBuffer` members has been adjusted appropriate by the caller. + fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> { + assert!(buf.buf.remaining_mut() > 0); + let r = unsafe { + let chunk = buf.buf.chunk_mut(); + let slice = std::slice::from_raw_parts_mut(chunk.as_mut_ptr(), chunk.len()); + self.io.read(slice) + }; + match r { + Ok(n) => unsafe { + buf.buf.advance_mut(n); + Ok(n) + }, + e => e, + } + } +} + +impl SendMsg for Pipe { + // Send data on the associated connection. `send_msg` adjusts the length of the + // `ConnectionBuffer` members based on the size of the successful send operation. + fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> { + assert!(!buf.buf.is_empty()); + let r = self.io.write(&buf.buf[..buf.buf.len()]); + if let Ok(n) = r { + buf.buf.advance(n); + while let Some(mut handle) = buf.pop_handle() { + handle.mark_sent() + } + } + r + } +} |