From 43a97878ce14b72f0981164f87f2e35e14151312 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 7 Apr 2024 11:22:09 +0200 Subject: Adding upstream version 110.0.1. Signed-off-by: Daniel Baumann --- .../rust/audioipc-client/.cargo-checksum.json | 1 + third_party/rust/audioipc-client/Cargo.toml | 23 ++ third_party/rust/audioipc-client/cbindgen.toml | 28 ++ third_party/rust/audioipc-client/src/context.rs | 431 +++++++++++++++++++++ third_party/rust/audioipc-client/src/lib.rs | 83 ++++ third_party/rust/audioipc-client/src/send_recv.rs | 73 ++++ third_party/rust/audioipc-client/src/stream.rs | 422 ++++++++++++++++++++ 7 files changed, 1061 insertions(+) create mode 100644 third_party/rust/audioipc-client/.cargo-checksum.json create mode 100644 third_party/rust/audioipc-client/Cargo.toml create mode 100644 third_party/rust/audioipc-client/cbindgen.toml create mode 100644 third_party/rust/audioipc-client/src/context.rs create mode 100644 third_party/rust/audioipc-client/src/lib.rs create mode 100644 third_party/rust/audioipc-client/src/send_recv.rs create mode 100644 third_party/rust/audioipc-client/src/stream.rs (limited to 'third_party/rust/audioipc-client') diff --git a/third_party/rust/audioipc-client/.cargo-checksum.json b/third_party/rust/audioipc-client/.cargo-checksum.json new file mode 100644 index 0000000000..b6934efe0a --- /dev/null +++ b/third_party/rust/audioipc-client/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"Cargo.toml":"1a69f4283975d1dff38d6ff0652032ea80b0c90eef78f9970f2d47a15f91ee22","cbindgen.toml":"bd89c5a9f52395b1c703ff04d1c0019dc3c92b691d571ae503c4b85753a44a39","src/context.rs":"6c50d02883421c3dc7de70721a099636dc9ae2285fb5a86279868b1c840630f2","src/lib.rs":"7016a5de8770d3269e62dd7ad096fdd00cf821febf916bbc3163225cf6a53002","src/send_recv.rs":"064a657c845762be1dbcbbfc18b3f8a51582eb540def8d2ceecf200184ad4f7a","src/stream.rs":"f493ddc3ae950fc18fe42249b13194bc0a80bd0ce88bc8c7d3295f9ee10a932e"},"package":null} \ No newline at end of file diff --git a/third_party/rust/audioipc-client/Cargo.toml b/third_party/rust/audioipc-client/Cargo.toml new file mode 100644 index 0000000000..481e257c30 --- /dev/null +++ b/third_party/rust/audioipc-client/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "audioipc-client" +version = "0.4.0" +authors = [ + "Matthew Gregan ", + "Dan Glastonbury " + ] +license = "ISC" +description = "Cubeb Backend for talking to remote cubeb server." +edition = "2018" + +[dependencies] +audioipc = { path="../audioipc" } +cubeb-backend = "0.10" +futures = { version="0.1.18", default-features=false, features=["use_std"] } +futures-cpupool = { version="0.1.8", default-features=false } +log = "0.4" +tokio = { version="0.1", default-features=false, features = ["rt-full"] } + +[dependencies.audio_thread_priority] +version = "0.26.1" +default-features = false +features = ["winapi"] diff --git a/third_party/rust/audioipc-client/cbindgen.toml b/third_party/rust/audioipc-client/cbindgen.toml new file mode 100644 index 0000000000..9a7c204fe4 --- /dev/null +++ b/third_party/rust/audioipc-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", "audioipc"] + +[export] +item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"] + +[parse] +parse_deps = true +include = ['audioipc'] + +[fn] +args = "Vertical" +rename_args = "GeckoCase" + +[struct] +rename_fields = "GeckoCase" + +[defines] +"windows" = "XP_WIN" +"unix" = "XP_UNIX" diff --git a/third_party/rust/audioipc-client/src/context.rs b/third_party/rust/audioipc-client/src/context.rs new file mode 100644 index 0000000000..2737d37c08 --- /dev/null +++ b/third_party/rust/audioipc-client/src/context.rs @@ -0,0 +1,431 @@ +// 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::codec::LengthDelimitedCodec; +use audioipc::framing::{framed, Framed}; +use audioipc::{core, rpc}; +use audioipc::{ + messages, messages::DeviceCollectionReq, messages::DeviceCollectionResp, ClientMessage, + ServerMessage, +}; +use cubeb_backend::{ + ffi, Context, ContextOps, DeviceCollectionRef, DeviceId, DeviceType, Error, Ops, Result, + Stream, StreamParams, StreamParamsRef, +}; +use futures::Future; +use futures_cpupool::{CpuFuture, CpuPool}; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::{fmt, mem, ptr}; +use tokio::reactor; +use tokio::runtime::current_thread; + +struct CubebClient; + +impl rpc::Client for CubebClient { + type Request = ServerMessage; + type Response = ClientMessage; + type Transport = + Framed>; +} + +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: rpc::ClientProxy, + core: core::CoreThread, + cpu_pool: CpuPool, + backend_id: CString, + device_collection_rpc: bool, + input_device_callback: Arc>, + output_device_callback: Arc>, +} + +impl ClientContext { + #[doc(hidden)] + pub fn handle(&self) -> current_thread::Handle { + self.core.handle() + } + + #[doc(hidden)] + pub fn rpc(&self) -> rpc::ClientProxy { + self.rpc.clone() + } + + #[doc(hidden)] + pub fn cpu_pool(&self) -> CpuPool { + self.cpu_pool.clone() + } +} + +#[cfg(target_os = "linux")] +fn promote_thread(rpc: &rpc::ClientProxy) { + match get_current_thread_info() { + Ok(info) => { + let bytes = info.serialize(); + // Don't wait for the response, this is on the callback thread, which must not block. + rpc.call(ServerMessage::PromoteThreadToRealTime(bytes)); + } + Err(_) => { + warn!("Could not remotely promote thread to RT."); + } + } +} + +#[cfg(not(target_os = "linux"))] +fn promote_thread(_rpc: &rpc::ClientProxy) { + 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) { + 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) { + if let Some(func) = callback { + func(); + } +} + +fn promote_and_register_thread( + rpc: &rpc::ClientProxy, + callback: Option, +) { + 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>, + output_device_callback: Arc>, + cpu_pool: CpuPool, +} + +impl rpc::Server for DeviceCollectionServer { + type Request = DeviceCollectionReq; + type Response = DeviceCollectionResp; + type Future = CpuFuture; + type Transport = + Framed>; + + fn process(&mut self, req: Self::Request) -> Self::Future { + 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) + }; + + self.cpu_pool.spawn_fn(move || { + 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) + } + } + }); + + Ok(DeviceCollectionResp::DeviceChange) + }) + } + } + } +} + +impl ContextOps for ClientContext { + fn init(_context_name: Option<&CStr>) -> Result { + fn bind_and_send_client( + stream: audioipc::AsyncMessageStream, + tx_rpc: &mpsc::Sender>, + ) { + let transport = framed(stream, Default::default()); + let rpc = rpc::bind_client::(transport); + // If send fails then the rx end has closed + // which is unlikely here. + let _ = tx_rpc.send(rpc); + } + + assert_not_in_callback(); + + let (tx_rpc, rx_rpc) = mpsc::channel(); + + 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_stream = + unsafe { audioipc::MessageStream::from_raw_handle(params.server_connection) }; + + let core = core::spawn_thread( + "AudioIPC Client RPC", + move || { + let handle = reactor::Handle::default(); + + register_thread(thread_create_callback); + + let stream = server_stream.into_tokio_ipc(&handle).unwrap(); + bind_and_send_client(stream, &tx_rpc); + Ok(()) + }, + move || unregister_thread(thread_destroy_callback), + ) + .map_err(|_| Error::default())?; + + let rpc = rx_rpc.recv().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"); + + let cpu_pool = futures_cpupool::Builder::new() + .name_prefix("AudioIPC") + .after_start(move || promote_and_register_thread(&rpc2, thread_create_callback)) + .before_stop(move || unregister_thread(thread_destroy_callback)) + .pool_size(params.pool_size) + .stack_size(params.stack_size) + .create(); + + let ctx = Box::new(ClientContext { + _ops: &CLIENT_OPS as *const _, + rpc, + core, + cpu_pool, + 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 { + assert_not_in_callback(); + send_recv!(self.rpc(), ContextGetMaxChannelCount => ContextMaxChannelCount()) + } + + fn min_latency(&mut self, params: StreamParams) -> Result { + 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 { + 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 = 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 { + 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 { + audioipc::MessageStream::from_raw_handle( + fd.platform_handle.take_handle().into_raw(), + ) + }; + + let server = DeviceCollectionServer { + input_device_callback: self.input_device_callback.clone(), + output_device_callback: self.output_device_callback.clone(), + cpu_pool: self.cpu_pool(), + }; + + let (wait_tx, wait_rx) = mpsc::channel(); + self.handle() + .spawn(futures::future::lazy(move || { + let handle = reactor::Handle::default(); + let stream = stream.into_tokio_ipc(&handle).unwrap(); + let transport = framed(stream, Default::default()); + rpc::bind_server(transport, server); + wait_tx.send(()).unwrap(); + Ok(()) + })) + .expect("Failed to spawn DeviceCollectionServer"); + wait_rx.recv().unwrap(); + 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.core) + .field("cpu_pool", &"...") + .finish() + } +} diff --git a/third_party/rust/audioipc-client/src/lib.rs b/third_party/rust/audioipc-client/src/lib.rs new file mode 100644 index 0000000000..48d785d8e1 --- /dev/null +++ b/third_party/rust/audioipc-client/src/lib.rs @@ -0,0 +1,83 @@ +// 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 cubeb_backend; +#[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 = std::cell::RefCell::new(false)); +thread_local!(static AUDIOIPC_INIT_PARAMS: std::cell::RefCell> = 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, + pub thread_destroy_callback: Option, +} + +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: 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 audioipc_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::(c, context_name) +} diff --git a/third_party/rust/audioipc-client/src/send_recv.rs b/third_party/rust/audioipc-client/src/send_recv.rs new file mode 100644 index 0000000000..b0d41aca2c --- /dev/null +++ b/third_party/rust/audioipc-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) -> Error +where + E: Into>, +{ + 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.wait() { + 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.wait() { + 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/audioipc-client/src/stream.rs b/third_party/rust/audioipc-client/src/stream.rs new file mode 100644 index 0000000000..e4d4b02b3c --- /dev/null +++ b/third_party/rust/audioipc-client/src/stream.rs @@ -0,0 +1,422 @@ +// 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::rpc; +use audioipc::shm::SharedMem; +use audioipc::{codec::LengthDelimitedCodec, messages::StreamCreateParams}; +use audioipc::{ + framing::{framed, Framed}, + messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage}, +}; +use cubeb_backend::{ffi, DeviceRef, Error, Result, Stream, StreamOps}; +use futures::Future; +use futures_cpupool::{CpuFuture, CpuPool}; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::ptr; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; +use tokio::reactor; + +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>, + // Signals ClientStream that CallbackServer has dropped. + shutdown_rx: mpsc::Receiver<()>, +} + +#[derive(Copy, Clone, Debug, PartialEq)] +enum StreamDirection { + Input, + Output, + Duplex, +} + +struct CallbackServer { + dir: StreamDirection, + shm: Option, + duplex_input: Option>, + data_cb: ffi::cubeb_data_callback, + state_cb: ffi::cubeb_state_callback, + user_ptr: usize, + cpu_pool: CpuPool, + device_change_cb: Arc>, + // Signals ClientStream that CallbackServer has dropped. + _shutdown_tx: mpsc::Sender<()>, +} + +impl rpc::Server for CallbackServer { + type Request = CallbackReq; + type Response = CallbackResp; + type Future = CpuFuture; + type Transport = + Framed>; + + fn process(&mut self, req: Self::Request) -> Self::Future { + 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; + + // Clone values that need to be moved into the cpu pool thread. + let mut shm = unsafe { self.shm.as_ref().unwrap().unsafe_view() }; + + let duplex_copy_ptr = match &mut self.duplex_input { + Some(buf) => { + assert_eq!(self.dir, StreamDirection::Duplex); + assert!(input_frame_size > 0); + assert!(buf.capacity() >= input_nbytes); + buf.as_mut_ptr() + } + None => ptr::null_mut(), + } as usize; + let user_ptr = self.user_ptr; + let cb = self.data_cb.unwrap(); + let dir = self.dir; + + self.cpu_pool.spawn_fn(move || { + // 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. + let (input_ptr, output_ptr) = match dir { + StreamDirection::Duplex => unsafe { + assert!(input_frame_size > 0); + assert!(output_frame_size > 0); + assert_ne!(duplex_copy_ptr, 0); + let input = shm.get_slice(input_nbytes).unwrap(); + ptr::copy_nonoverlapping( + input.as_ptr(), + duplex_copy_ptr as *mut _, + input.len(), + ); + ( + duplex_copy_ptr as _, + shm.get_mut_slice(output_nbytes).unwrap().as_mut_ptr(), + ) + }, + StreamDirection::Input => unsafe { + assert!(input_frame_size > 0); + assert_eq!(output_frame_size, 0); + ( + shm.get_slice(input_nbytes).unwrap().as_ptr(), + ptr::null_mut(), + ) + }, + StreamDirection::Output => unsafe { + assert!(output_frame_size > 0); + assert_eq!(input_frame_size, 0); + ( + ptr::null(), + shm.get_mut_slice(output_nbytes).unwrap().as_mut_ptr(), + ) + }, + }; + + run_in_callback(|| { + let nframes = unsafe { + cb( + ptr::null_mut(), // https://github.com/kinetiknz/cubeb/issues/518 + user_ptr as *mut c_void, + input_ptr as *const _, + output_ptr as *mut _, + nframes as _, + ) + }; + + Ok(CallbackResp::Data(nframes as isize)) + }) + }) + } + CallbackReq::State(state) => { + trace!("stream_thread: State Callback: {:?}", state); + let user_ptr = self.user_ptr; + let cb = self.state_cb.unwrap(); + self.cpu_pool.spawn_fn(move || { + run_in_callback(|| unsafe { + cb(ptr::null_mut(), user_ptr as *mut _, state); + }); + + Ok(CallbackResp::State) + }) + } + CallbackReq::DeviceChange => { + let cb = self.device_change_cb.clone(); + let user_ptr = self.user_ptr; + self.cpu_pool.spawn_fn(move || { + run_in_callback(|| { + let cb = cb.lock().unwrap(); + if let Some(cb) = *cb { + unsafe { + cb(user_ptr as *mut _); + } + } else { + warn!("DeviceChange received with null callback"); + } + }); + + Ok(CallbackResp::DeviceChange) + }) + } + CallbackReq::SharedMem(mut handle, shm_area_size) => { + self.shm = match unsafe { SharedMem::from(handle.take_handle(), shm_area_size) } { + Ok(shm) => Some(shm), + Err(e) => { + warn!( + "sharedmem client mapping failed (size={}, err={:?})", + shm_area_size, e + ); + return self + .cpu_pool + .spawn_fn(move || Ok(CallbackResp::Error(ffi::CUBEB_ERROR))); + } + }; + + self.duplex_input = if let StreamDirection::Duplex = self.dir { + let mut duplex_input = Vec::new(); + match duplex_input.try_reserve_exact(shm_area_size) { + Ok(()) => Some(duplex_input), + Err(e) => { + warn!( + "duplex_input allocation failed (size={}, err={:?})", + shm_area_size, e + ); + return self + .cpu_pool + .spawn_fn(move || Ok(CallbackResp::Error(ffi::CUBEB_ERROR))); + } + } + } else { + None + }; + self.cpu_pool.spawn_fn(move || Ok(CallbackResp::SharedMem)) + } + } + } +} + +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 { + 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 = {:?}", + data.token, data.platform_handle + ); + + let stream = unsafe { + audioipc::MessageStream::from_raw_handle(data.platform_handle.take_handle().into_raw()) + }; + + let user_data = user_ptr as usize; + + let cpu_pool = ctx.cpu_pool(); + + 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 dir = match ( + init_params.input_stream_params, + init_params.output_stream_params, + ) { + (Some(_), Some(_)) => StreamDirection::Duplex, + (Some(_), None) => StreamDirection::Input, + (None, Some(_)) => StreamDirection::Output, + (None, None) => unreachable!(), + }; + + let server = CallbackServer { + dir, + shm: None, + duplex_input: None, + data_cb: data_callback, + state_cb: state_callback, + user_ptr: user_data, + cpu_pool, + device_change_cb: device_change_cb.clone(), + _shutdown_tx, + }; + + let (wait_tx, wait_rx) = mpsc::channel(); + ctx.handle() + .spawn(futures::future::lazy(move || { + let handle = reactor::Handle::default(); + let stream = stream.into_tokio_ipc(&handle).unwrap(); + let transport = framed(stream, Default::default()); + rpc::bind_server(transport, server); + wait_tx.send(()).unwrap(); + Ok(()) + })) + .expect("Failed to spawn CallbackServer"); + wait_rx.recv().unwrap(); + + send_recv!(rpc, StreamInit(data.token, init_params) => StreamInitialized)?; + + 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<'ctx> Drop for ClientStream<'ctx> { + fn drop(&mut self) { + debug!("ClientStream drop"); + let rpc = self.context.rpc(); + let _ = send_recv!(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<'ctx> StreamOps for ClientStream<'ctx> { + 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 { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamGetPosition(self.token) => StreamPosition()) + } + + fn latency(&mut self) -> Result { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamGetLatency(self.token) => StreamLatency()) + } + + fn input_latency(&mut self) -> Result { + 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(); + // It's all unsafe... + if device.as_ptr().is_null() { + Err(Error::error()) + } else { + unsafe { + let _: Box = 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 { + let stm = ClientStream::init(ctx, init_params, data_callback, state_callback, user_ptr)?; + debug_assert_eq!(stm.user_ptr(), user_ptr); + Ok(stm) +} -- cgit v1.2.3