From 2aa4a82499d4becd2284cdb482213d541b8804dd Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 28 Apr 2024 16:29:10 +0200 Subject: Adding upstream version 86.0.1. Signed-off-by: Daniel Baumann --- media/audioipc/server/src/lib.rs | 166 +++++++ media/audioipc/server/src/server.rs | 862 ++++++++++++++++++++++++++++++++++++ 2 files changed, 1028 insertions(+) create mode 100644 media/audioipc/server/src/lib.rs create mode 100644 media/audioipc/server/src/server.rs (limited to 'media/audioipc/server/src') diff --git a/media/audioipc/server/src/lib.rs b/media/audioipc/server/src/lib.rs new file mode 100644 index 0000000000..4d49c8fc08 --- /dev/null +++ b/media/audioipc/server/src/lib.rs @@ -0,0 +1,166 @@ +// 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::core; +use audioipc::platformhandle_passing::framed_with_platformhandles; +use audioipc::rpc; +use audioipc::{MessageStream, PlatformHandle, PlatformHandleType}; +use futures::sync::oneshot; +use futures::Future; +use once_cell::sync::Lazy; +use std::ffi::{CStr, CString}; +use std::os::raw::c_void; +use std::ptr; +use std::sync::Mutex; +use tokio::reactor; + +mod server; + +struct CubebContextParams { + context_name: CString, + backend_name: Option, +} + +static G_CUBEB_CONTEXT_PARAMS: Lazy> = Lazy::new(|| { + Mutex::new(CubebContextParams { + context_name: CString::new("AudioIPC Server").unwrap(), + backend_name: None, + }) +}); + +#[allow(deprecated)] +pub mod errors { + error_chain! { + links { + AudioIPC(::audioipc::errors::Error, ::audioipc::errors::ErrorKind); + } + foreign_links { + Cubeb(cubeb_core::Error); + Io(::std::io::Error); + Canceled(::futures::sync::oneshot::Canceled); + } + } +} + +use crate::errors::*; + +struct ServerWrapper { + core_thread: core::CoreThread, + callback_thread: core::CoreThread, +} + +fn run() -> Result { + trace!("Starting up cubeb audio server event loop thread..."); + + let callback_thread = core::spawn_thread( + "AudioIPC Callback RPC", + || { + match promote_current_thread_to_real_time(0, 48000) { + Ok(_) => {} + Err(_) => { + debug!("Failed to promote audio callback thread to real-time."); + } + } + trace!("Starting up cubeb audio callback event loop thread..."); + Ok(()) + }, + || {}, + ) + .or_else(|e| { + debug!( + "Failed to start cubeb audio callback event loop thread: {:?}", + e + ); + Err(e) + })?; + + let core_thread = core::spawn_thread( + "AudioIPC Server RPC", + move || { + audioipc::server_platform_init(); + Ok(()) + }, + || {}, + ) + .or_else(|e| { + debug!("Failed to cubeb audio core event loop thread: {:?}", e); + Err(e) + })?; + + Ok(ServerWrapper { + core_thread, + callback_thread, + }) +} + +#[no_mangle] +pub unsafe extern "C" fn audioipc_server_start( + context_name: *const std::os::raw::c_char, + backend_name: *const std::os::raw::c_char, +) -> *mut c_void { + 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 run() { + Ok(server) => Box::into_raw(Box::new(server)) as *mut _, + Err(_) => ptr::null_mut() as *mut _, + } +} + +#[no_mangle] +pub extern "C" fn audioipc_server_new_client(p: *mut c_void) -> PlatformHandleType { + let (wait_tx, wait_rx) = oneshot::channel(); + let wrapper: &ServerWrapper = unsafe { &*(p as *mut _) }; + + let core_handle = wrapper.callback_thread.handle(); + + // We create a connected pair of anonymous IPC endpoints. One side + // is registered with the reactor core, the other side is returned + // to the caller. + MessageStream::anonymous_ipc_pair() + .and_then(|(ipc_server, ipc_client)| { + // Spawn closure to run on same thread as reactor::Core + // via remote handle. + wrapper + .core_thread + .handle() + .spawn(futures::future::lazy(|| { + trace!("Incoming connection"); + let handle = reactor::Handle::default(); + ipc_server.into_tokio_ipc(&handle) + .and_then(|sock| { + let transport = framed_with_platformhandles(sock, Default::default()); + rpc::bind_server(transport, server::CubebServer::new(core_handle)); + Ok(()) + }).map_err(|_| ()) + // Notify waiting thread that server has been registered. + .and_then(|_| wait_tx.send(())) + })) + .expect("Failed to spawn CubebServer"); + // Wait for notification that server has been registered + // with reactor::Core. + let _ = wait_rx.wait(); + Ok(unsafe { PlatformHandle::from(ipc_client).into_raw() }) + }) + .unwrap_or(audioipc::INVALID_HANDLE_VALUE) +} + +#[no_mangle] +pub extern "C" fn audioipc_server_stop(p: *mut c_void) { + let wrapper = unsafe { Box::::from_raw(p as *mut _) }; + drop(wrapper); +} diff --git a/media/audioipc/server/src/server.rs b/media/audioipc/server/src/server.rs new file mode 100644 index 0000000000..c2d4f31a02 --- /dev/null +++ b/media/audioipc/server/src/server.rs @@ -0,0 +1,862 @@ +// 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; +use audioipc::codec::LengthDelimitedCodec; +use audioipc::frame::{framed, Framed}; +use audioipc::messages::{ + CallbackReq, CallbackResp, ClientMessage, Device, DeviceCollectionReq, DeviceCollectionResp, + DeviceInfo, RegisterDeviceCollectionChanged, ServerMessage, StreamCreate, StreamCreateParams, + StreamInitParams, StreamParams, +}; +use audioipc::platformhandle_passing::FramedWithPlatformHandles; +use audioipc::rpc; +use audioipc::shm::{SharedMemReader, SharedMemWriter}; +use audioipc::{MessageStream, PlatformHandle}; +use cubeb_core as cubeb; +use cubeb_core::ffi; +use futures::future::{self, FutureResult}; +use futures::sync::oneshot; +use futures::Future; +use slab; +use std::convert::From; +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::{AtomicUsize, Ordering}; +use std::{cell::RefCell, sync::Mutex}; +use std::{panic, slice}; +use tokio::reactor; +use tokio::runtime::current_thread; + +use crate::errors::*; + +fn error(error: cubeb::Error) -> ClientMessage { + ClientMessage::Error(error.raw_code()) +} + +struct CubebDeviceCollectionManager { + servers: Mutex>>>, +} + +impl CubebDeviceCollectionManager { + fn new() -> CubebDeviceCollectionManager { + CubebDeviceCollectionManager { + servers: Mutex::new(Vec::new()), + } + } + + fn register( + &mut self, + context: &cubeb::Context, + server: &Rc>, + devtype: cubeb::DeviceType, + ) -> cubeb::Result<()> { + let mut servers = self.servers.lock().unwrap(); + if servers.is_empty() { + self.internal_register(context, true)?; + } + server.borrow_mut().devtype.insert(devtype); + if servers.iter().find(|s| Rc::ptr_eq(s, server)).is_none() { + servers.push(server.clone()); + } + Ok(()) + } + + fn unregister( + &mut self, + context: &cubeb::Context, + server: &Rc>, + devtype: cubeb::DeviceType, + ) -> cubeb::Result<()> { + let mut servers = self.servers.lock().unwrap(); + server.borrow_mut().devtype.remove(devtype); + if server.borrow().devtype.is_empty() { + servers.retain(|s| !Rc::ptr_eq(&s, server)); + } + if servers.is_empty() { + self.internal_register(context, false)?; + } + Ok(()) + } + + fn internal_register(&self, context: &cubeb::Context, enable: bool) -> cubeb::Result<()> { + let user_ptr = if enable { + self as *const CubebDeviceCollectionManager as *mut c_void + } else { + std::ptr::null_mut() + }; + 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 }, + user_ptr, + )?; + } + } + Ok(()) + } + + // Warning: this is called from an internal cubeb thread, so we must not mutate unprotected shared state. + unsafe fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) { + let servers = self.servers.lock().unwrap(); + servers.iter().for_each(|server| { + if server + .borrow() + .devtype + .contains(cubeb::DeviceType::from_bits_truncate(device_type)) + { + server + .borrow_mut() + .device_collection_changed_callback(device_type) + } + }); + } +} + +struct DevIdMap { + devices: Vec, +} + +// 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 to_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 `to_handle`, return the associated + // cubeb_devid. Invalid handles result in a panic. + fn from_handle(&self, handle: usize) -> usize { + self.devices[handle] + } +} + +struct CubebContextState { + context: cubeb::Result, + manager: CubebDeviceCollectionManager, +} + +thread_local!(static CONTEXT_KEY: RefCell> = RefCell::new(None)); + +fn cubeb_init_from_context_params() -> cubeb::Result { + let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap(); + let context_name = Some(params.context_name.as_c_str()); + let backend_name = if let Some(ref name) = params.backend_name { + Some(name.as_c_str()) + } else { + None + }; + 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(f: F) -> T +where + F: FnOnce(&cubeb::Result, &mut CubebDeviceCollectionManager) -> T, +{ + CONTEXT_KEY.with(|k| { + let mut state = k.borrow_mut(); + if state.is_none() { + *state = Some(CubebContextState { + context: cubeb_init_from_context_params(), + manager: CubebDeviceCollectionManager::new(), + }); + } + let CubebContextState { context, manager } = 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 rpc::Client for DeviceCollectionClient { + type Request = DeviceCollectionReq; + type Response = DeviceCollectionResp; + type Transport = + Framed>; +} + +struct CallbackClient; + +impl rpc::Client for CallbackClient { + type Request = CallbackReq; + type Response = CallbackResp; + type Transport = + Framed>; +} + +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 sending input data to client + input_shm: SharedMemWriter, + /// Shared memory buffer for receiving output data from client + output_shm: SharedMemReader, + /// RPC interface to callback server running in client + rpc: rpc::ClientProxy, +} + +impl ServerStreamCallbacks { + fn data_callback(&mut self, input: &[u8], output: &mut [u8], nframes: isize) -> isize { + trace!( + "Stream data callback: {} {} {}", + nframes, + input.len(), + output.len() + ); + + self.input_shm.write(input).unwrap(); + + let r = self + .rpc + .call(CallbackReq::Data { + nframes, + input_frame_size: self.input_frame_size as usize, + output_frame_size: self.output_frame_size as usize, + }) + .wait(); + + match r { + Ok(CallbackResp::Data(frames)) => { + if frames >= 0 { + let nbytes = frames as usize * self.output_frame_size as usize; + trace!("Reslice output to {}", nbytes); + self.output_shm.read(&mut output[..nbytes]).unwrap(); + } + frames + } + _ => { + debug!("Unexpected message {:?} during data_callback", r); + // TODO: Return a CUBEB_ERROR result here once + // https://github.com/kinetiknz/cubeb/issues/553 is + // fixed. + 0 + } + } + } + + fn state_callback(&mut self, state: cubeb::State) { + trace!("Stream state callback: {:?}", state); + let r = self.rpc.call(CallbackReq::State(state.into())).wait(); + match r { + Ok(CallbackResp::State) => {} + _ => { + debug!("Unexpected message {:?} during state callback", r); + } + } + } + + fn device_change_callback(&mut self) { + trace!("Stream device change callback"); + let r = self.rpc.call(CallbackReq::DeviceChange).wait(); + 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, + cbs: Box, +} + +impl Drop for ServerStream { + fn drop(&mut self) { + // `stream` *must* be dropped before `cbs`. + drop(self.stream.take()); + } +} + +type StreamSlab = slab::Slab; + +struct CubebServerCallbacks { + rpc: rpc::ClientProxy, + devtype: cubeb::DeviceType, +} + +impl CubebServerCallbacks { + fn device_collection_changed_callback(&mut 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)) + .wait(); + } +} + +pub struct CubebServer { + handle: current_thread::Handle, + streams: StreamSlab, + remote_pid: Option, + cbs: Option>>, + devidmap: DevIdMap, +} + +impl rpc::Server for CubebServer { + type Request = ServerMessage; + type Response = ClientMessage; + type Future = FutureResult; + type Transport = FramedWithPlatformHandles< + audioipc::AsyncMessageStream, + LengthDelimitedCodec, + >; + + fn process(&mut self, req: Self::Request) -> Self::Future { + if let ServerMessage::ClientConnect(pid) = req { + self.remote_pid = Some(pid); + } + let resp = with_local_context(|context, manager| match *context { + Err(_) => error(cubeb::Error::error()), + Ok(ref context) => self.process_msg(context, manager, &req), + }); + future::ok(resp) + } +} + +// 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(handle: current_thread::Handle) -> Self { + CubebServer { + handle, + streams: StreamSlab::new(), + remote_pid: None, + cbs: None, + devidmap: DevIdMap::new(), + } + } + + // 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 = 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.to_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::StreamResetDefaultDevice(stm_tok) => try_stream!(self, stm_tok) + .reset_default_device() + .map(|_| ClientMessage::StreamDefaultDeviceReset) + .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 => { + if let Ok((ipc_server, ipc_client)) = MessageStream::anonymous_ipc_pair() { + debug!( + "Created device collection RPC pair: {:?}-{:?}", + ipc_server, ipc_client + ); + + // This code is currently running on the Client/Server RPC + // handling thread. We need to move the registration of the + // bind_client to the callback RPC handling thread. This is + // done by spawning a future on `handle`. + let (tx, rx) = oneshot::channel(); + self.handle + .spawn(futures::future::lazy(move || { + let handle = reactor::Handle::default(); + let stream = ipc_server.into_tokio_ipc(&handle).unwrap(); + let transport = framed(stream, Default::default()); + let rpc = rpc::bind_client::(transport); + drop(tx.send(rpc)); + Ok(()) + })) + .expect("Failed to spawn DeviceCollectionClient"); + + // TODO: The lowest comms layer expects exactly 3 PlatformHandles, but we only + // need one here. Send some dummy handles over for the other side to discard. + let (dummy1, dummy2) = + MessageStream::anonymous_ipc_pair().expect("need dummy IPC pair"); + if let Ok(rpc) = rx.wait() { + self.cbs = Some(Rc::new(RefCell::new(CubebServerCallbacks { + rpc, + devtype: cubeb::DeviceType::empty(), + }))); + let fds = RegisterDeviceCollectionChanged { + platform_handles: [ + PlatformHandle::from(ipc_client), + PlatformHandle::from(dummy1), + PlatformHandle::from(dummy2), + ], + target_pid: self.remote_pid.unwrap(), + }; + + ClientMessage::ContextSetupDeviceCollectionCallback(fds) + } else { + warn!("Failed to setup RPC client"); + error(cubeb::Error::error()) + } + } else { + warn!("Failed to create RPC pair"); + error(cubeb::Error::error()) + } + } + + 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 { + if devtype == cubeb::DeviceType::UNKNOWN { + return Err(cubeb::Error::invalid_parameter()); + } + + assert!(self.cbs.is_some()); + let cbs = self.cbs.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 { + 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()); + + let (ipc_server, ipc_client) = MessageStream::anonymous_ipc_pair()?; + debug!("Created callback pair: {:?}-{:?}", ipc_server, ipc_client); + let shm_id = get_shm_id(); + let (input_shm, input_file) = + SharedMemWriter::new(&format!("{}-input", shm_id), audioipc::SHM_AREA_SIZE)?; + let (output_shm, output_file) = + SharedMemReader::new(&format!("{}-output", shm_id), audioipc::SHM_AREA_SIZE)?; + + // This code is currently running on the Client/Server RPC + // handling thread. We need to move the registration of the + // bind_client to the callback RPC handling thread. This is + // done by spawning a future on `handle`. + let (tx, rx) = oneshot::channel(); + self.handle + .spawn(futures::future::lazy(move || { + let handle = reactor::Handle::default(); + let stream = ipc_server.into_tokio_ipc(&handle).unwrap(); + let transport = framed(stream, Default::default()); + let rpc = rpc::bind_client::(transport); + drop(tx.send(rpc)); + Ok(()) + })) + .expect("Failed to spawn CallbackClient"); + + let rpc: rpc::ClientProxy = match rx.wait() { + Ok(rpc) => rpc, + Err(_) => bail!("Failed to create callback rpc."), + }; + + let cbs = Box::new(ServerStreamCallbacks { + input_frame_size, + output_frame_size, + input_shm, + output_shm, + rpc, + }); + + let entry = self.streams.vacant_entry(); + let key = entry.key(); + debug!("Registering stream {:?}", key); + + entry.insert(ServerStream { stream: None, cbs }); + + Ok(ClientMessage::StreamCreated(StreamCreate { + token: key, + platform_handles: [ + PlatformHandle::from(ipc_client), + PlatformHandle::from(input_file), + PlatformHandle::from(output_file), + ], + target_pid: self.remote_pid.unwrap(), + })) + } + + // 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 { + // 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.from_handle(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.from_handle(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 _) + }); + + let latency = params.latency_frames; + + let server_stream = &mut self.streams[stm_tok]; + assert!(size_of::>() == size_of::()); + 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) => return Err(e.into()), // XXX full teardown of ServerStream? + } + }; + + server_stream.stream = Some(stream); + + Ok(ClientMessage::StreamInitialized) + } +} + +// 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 + }); + // TODO: Return a CUBEB_ERROR result here once + // https://github.com/kinetiknz/cubeb/issues/553 is fixed. + ok.unwrap_or(0) +} + +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"); +} -- cgit v1.2.3