diff options
Diffstat (limited to 'media/audioipc')
38 files changed, 6057 insertions, 0 deletions
diff --git a/media/audioipc/Cargo.toml b/media/audioipc/Cargo.toml new file mode 100644 index 0000000000..64d3e2bdb1 --- /dev/null +++ b/media/audioipc/Cargo.toml @@ -0,0 +1,3 @@ +[workspace] +members = ["audioipc", "client", "server"] + diff --git a/media/audioipc/README.md b/media/audioipc/README.md new file mode 100644 index 0000000000..ecbf259ec6 --- /dev/null +++ b/media/audioipc/README.md @@ -0,0 +1 @@ +# Cubeb Audio Remoting Prototype diff --git a/media/audioipc/README_MOZILLA b/media/audioipc/README_MOZILLA new file mode 100644 index 0000000000..df76f22059 --- /dev/null +++ b/media/audioipc/README_MOZILLA @@ -0,0 +1,8 @@ +The source from this directory was copied from the audioipc-2 +git repository using the update.sh script. The only changes +made were those applied by update.sh and the addition of +Makefile.in build files for the Mozilla build system. + +The audioipc-2 git repository is: https://github.com/djg/audioipc-2.git + +The git commit ID used was db32323c92b2b4f5ca2e702dcf987fedc1e71768 (2020-10-21 09:02:32 +1300) diff --git a/media/audioipc/audioipc/Cargo.toml b/media/audioipc/audioipc/Cargo.toml new file mode 100644 index 0000000000..95441e345f --- /dev/null +++ b/media/audioipc/audioipc/Cargo.toml @@ -0,0 +1,42 @@ +[package] +name = "audioipc" +version = "0.2.5" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>" + ] +description = "Remote Cubeb IPC" +edition = "2018" + +[dependencies] +bincode = "1" +bytes = "0.4" +cubeb = "0.8" +futures = "0.1.29" +log = "0.4" +memmap = "0.7" +serde = "1" +serde_derive = "1" +tokio = "0.1" +tokio-io = "0.1" +audio_thread_priority = "0.23.4" + +[target.'cfg(unix)'.dependencies] +iovec = "0.1" +libc = "0.2" +mio = "0.6.19" +mio-uds = "0.6.7" +tokio-reactor = "0.1" + +[target.'cfg(windows)'.dependencies] +mio = "0.6.19" +miow = "0.3.3" +mio-named-pipes = { git = "https://github.com/kinetiknz/mio-named-pipes", rev = "21c26326f5f45f415c49eac4ba5bc41a2f961321" } +winapi = { version = "0.3.6", features = ["combaseapi", "objbase"] } + +[dependencies.error-chain] +version = "0.11.0" +default-features = false + +[build-dependencies] +cc = "1.0" diff --git a/media/audioipc/audioipc/src/async_msg.rs b/media/audioipc/audioipc/src/async_msg.rs new file mode 100644 index 0000000000..fe682a71ed --- /dev/null +++ b/media/audioipc/audioipc/src/async_msg.rs @@ -0,0 +1,168 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +//! Various async helpers modelled after futures-rs and tokio-io. + +#[cfg(unix)] +use crate::msg::{RecvMsg, SendMsg}; +use bytes::{Buf, BufMut}; +#[cfg(unix)] +use futures::Async; +use futures::Poll; +#[cfg(unix)] +use iovec::IoVec; +#[cfg(unix)] +use mio::Ready; +use std::io; +use tokio_io::{AsyncRead, AsyncWrite}; + +pub trait AsyncRecvMsg: AsyncRead { + /// Pull some bytes from this source into the specified `Buf`, returning + /// how many bytes were read. + /// + /// The `buf` provided will have bytes read into it and the internal cursor + /// will be advanced if any bytes were read. Note that this method typically + /// will not reallocate the buffer provided. + fn recv_msg_buf<B>(&mut self, buf: &mut B, cmsg: &mut B) -> Poll<(usize, i32), io::Error> + where + Self: Sized, + B: BufMut; +} + +/// A trait for writable objects which operated in an async fashion. +/// +/// This trait inherits from `std::io::Write` and indicates that an I/O object is +/// **nonblocking**, meaning that it will return an error instead of blocking +/// when bytes cannot currently be written, but hasn't closed. Specifically +/// this means that the `write` function for types that implement this trait +/// can have a few return values: +/// +/// * `Ok(n)` means that `n` bytes of data was immediately written . +/// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was +/// written from the buffer provided. The I/O object is not currently +/// writable but may become writable in the future. +/// * `Err(e)` for other errors are standard I/O errors coming from the +/// underlying object. +pub trait AsyncSendMsg: AsyncWrite { + /// Write a `Buf` into this value, returning how many bytes were written. + /// + /// Note that this method will advance the `buf` provided automatically by + /// the number of bytes written. + fn send_msg_buf<B, C>(&mut self, buf: &mut B, cmsg: &C) -> Poll<usize, io::Error> + where + Self: Sized, + B: Buf, + C: Buf; +} + +//////////////////////////////////////////////////////////////////////////////// + +#[cfg(unix)] +impl AsyncRecvMsg for super::AsyncMessageStream { + fn recv_msg_buf<B>(&mut self, buf: &mut B, cmsg: &mut B) -> Poll<(usize, i32), io::Error> + where + B: BufMut, + { + if let Async::NotReady = + <super::AsyncMessageStream>::poll_read_ready(self, Ready::readable())? + { + return Ok(Async::NotReady); + } + let r = unsafe { + // The `IoVec` type can't have a 0-length size, so we create a bunch + // of dummy versions on the stack with 1 length which we'll quickly + // overwrite. + let b1: &mut [u8] = &mut [0]; + let b2: &mut [u8] = &mut [0]; + let b3: &mut [u8] = &mut [0]; + let b4: &mut [u8] = &mut [0]; + let b5: &mut [u8] = &mut [0]; + let b6: &mut [u8] = &mut [0]; + let b7: &mut [u8] = &mut [0]; + let b8: &mut [u8] = &mut [0]; + let b9: &mut [u8] = &mut [0]; + let b10: &mut [u8] = &mut [0]; + let b11: &mut [u8] = &mut [0]; + let b12: &mut [u8] = &mut [0]; + let b13: &mut [u8] = &mut [0]; + let b14: &mut [u8] = &mut [0]; + let b15: &mut [u8] = &mut [0]; + let b16: &mut [u8] = &mut [0]; + let mut bufs: [&mut IoVec; 16] = [ + b1.into(), + b2.into(), + b3.into(), + b4.into(), + b5.into(), + b6.into(), + b7.into(), + b8.into(), + b9.into(), + b10.into(), + b11.into(), + b12.into(), + b13.into(), + b14.into(), + b15.into(), + b16.into(), + ]; + let n = buf.bytes_vec_mut(&mut bufs); + self.recv_msg(&mut bufs[..n], cmsg.bytes_mut()) + }; + + match r { + Ok((n, cmsg_len, flags)) => { + unsafe { + buf.advance_mut(n); + } + unsafe { + cmsg.advance_mut(cmsg_len); + } + Ok((n, flags).into()) + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_read_ready(mio::Ready::readable())?; + Ok(Async::NotReady) + } + Err(e) => Err(e), + } + } +} + +#[cfg(unix)] +impl AsyncSendMsg for super::AsyncMessageStream { + fn send_msg_buf<B, C>(&mut self, buf: &mut B, cmsg: &C) -> Poll<usize, io::Error> + where + B: Buf, + C: Buf, + { + if let Async::NotReady = <super::AsyncMessageStream>::poll_write_ready(self)? { + return Ok(Async::NotReady); + } + let r = { + // The `IoVec` type can't have a zero-length size, so create a dummy + // version from a 1-length slice which we'll overwrite with the + // `bytes_vec` method. + static DUMMY: &[u8] = &[0]; + let nom = <&IoVec>::from(DUMMY); + let mut bufs = [ + nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, + ]; + let n = buf.bytes_vec(&mut bufs); + self.send_msg(&bufs[..n], cmsg.bytes()) + }; + match r { + Ok(n) => { + buf.advance(n); + Ok(Async::Ready(n)) + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + self.clear_write_ready()?; + Ok(Async::NotReady) + } + Err(e) => Err(e), + } + } +} diff --git a/media/audioipc/audioipc/src/cmsg.rs b/media/audioipc/audioipc/src/cmsg.rs new file mode 100644 index 0000000000..19adb2a4fb --- /dev/null +++ b/media/audioipc/audioipc/src/cmsg.rs @@ -0,0 +1,178 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use bytes::{BufMut, Bytes, BytesMut}; +use libc::{self, cmsghdr}; +use std::convert::TryInto; +use std::os::unix::io::RawFd; +use std::{convert, mem, ops, slice}; + +#[derive(Clone, Debug)] +pub struct Fds { + fds: Bytes, +} + +impl convert::AsRef<[RawFd]> for Fds { + fn as_ref(&self) -> &[RawFd] { + let n = self.fds.len() / mem::size_of::<RawFd>(); + unsafe { slice::from_raw_parts(self.fds.as_ptr() as *const _, n) } + } +} + +impl ops::Deref for Fds { + type Target = [RawFd]; + + #[inline] + fn deref(&self) -> &[RawFd] { + self.as_ref() + } +} + +pub struct ControlMsgIter { + control: Bytes, +} + +pub fn iterator(c: Bytes) -> ControlMsgIter { + ControlMsgIter { control: c } +} + +impl Iterator for ControlMsgIter { + type Item = Fds; + + fn next(&mut self) -> Option<Self::Item> { + loop { + let control = self.control.clone(); + let cmsghdr_len = len(0); + + if control.len() < cmsghdr_len { + // No more entries---not enough data in `control` for a + // complete message. + return None; + } + + let cmsg: &cmsghdr = unsafe { &*(control.as_ptr() as *const _) }; + // The offset to the next cmsghdr in control. This must be + // aligned to a boundary that matches the type used to + // represent the length of the message. + let cmsg_len = cmsg.cmsg_len as usize; + let cmsg_space = space(cmsg_len - cmsghdr_len); + self.control = if cmsg_space > control.len() { + // No more entries---not enough data in `control` for a + // complete message. + Bytes::new() + } else { + control.slice_from(cmsg_space) + }; + + match (cmsg.cmsg_level, cmsg.cmsg_type) { + (libc::SOL_SOCKET, libc::SCM_RIGHTS) => { + trace!("Found SCM_RIGHTS..."); + return Some(Fds { + fds: control.slice(cmsghdr_len, cmsg_len as _), + }); + } + (level, kind) => { + trace!("Skipping cmsg level, {}, type={}...", level, kind); + } + } + } + } +} + +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub enum Error { + /// Not enough space in storage to insert control mesage. + NoSpace, +} + +#[must_use] +pub struct ControlMsgBuilder { + result: Result<BytesMut, Error>, +} + +pub fn builder(buf: &mut BytesMut) -> ControlMsgBuilder { + let buf = aligned(buf); + ControlMsgBuilder { result: Ok(buf) } +} + +impl ControlMsgBuilder { + fn msg(mut self, level: libc::c_int, kind: libc::c_int, msg: &[u8]) -> Self { + self.result = self.result.and_then(|mut cmsg| { + let cmsg_space = space(msg.len()); + if cmsg.remaining_mut() < cmsg_space { + return Err(Error::NoSpace); + } + + // 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)] + let cmsghdr = cmsghdr { + cmsg_len: len(msg.len()).try_into().unwrap(), + cmsg_level: level, + cmsg_type: kind, + ..zeroed + }; + + unsafe { + let cmsghdr_ptr = cmsg.bytes_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); + } + + Ok(cmsg) + }); + + self + } + + pub fn rights(self, fds: &[RawFd]) -> Self { + self.msg(libc::SOL_SOCKET, libc::SCM_RIGHTS, fds.as_bytes()) + } + + pub fn finish(self) -> Result<Bytes, Error> { + self.result.map(|mut cmsg| cmsg.take().freeze()) + } +} + +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) } + } +} + +fn aligned(buf: &BytesMut) -> BytesMut { + let mut aligned_buf = buf.clone(); + aligned_buf.reserve(buf.remaining_mut()); + let cmsghdr_align = mem::align_of::<cmsghdr>(); + let n = unsafe { aligned_buf.bytes_mut().as_ptr() } as usize & (cmsghdr_align - 1); + if n != 0 { + unsafe { aligned_buf.advance_mut(n) }; + drop(aligned_buf.take()); + } + aligned_buf +} + +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/media/audioipc/audioipc/src/codec.rs b/media/audioipc/audioipc/src/codec.rs new file mode 100644 index 0000000000..2a6d15de0e --- /dev/null +++ b/media/audioipc/audioipc/src/codec.rs @@ -0,0 +1,185 @@ +// 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. + +use bincode::{self, deserialize, serialized_size}; +use bytes::{BufMut, ByteOrder, BytesMut, LittleEndian}; +use serde::de::DeserializeOwned; +use serde::ser::Serialize; +use std::fmt::Debug; +use std::io; +use std::marker::PhantomData; + +//////////////////////////////////////////////////////////////////////////////// +// 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 intox 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, + __in: PhantomData<In>, + __out: PhantomData<Out>, +} + +enum State { + Length, + Data(usize), +} + +const MAX_MESSAGE_LEN: u64 = 1024 * 1024; +const MESSAGE_LENGTH_SIZE: usize = std::mem::size_of::<u32>(); +// TODO: static assert that MAX_MESSAGE_LEN can be encoded into MESSAGE_LENGTH_SIZE. + +impl<In, Out> Default for LengthDelimitedCodec<In, Out> { + fn default() -> Self { + LengthDelimitedCodec { + state: State::Length, + __in: PhantomData, + __out: PhantomData, + } + } +} + +impl<In, Out> LengthDelimitedCodec<In, Out> { + // Lengths are encoded as little endian u32 + fn decode_length(&mut self, buf: &mut BytesMut) -> io::Result<Option<usize>> { + if buf.len() < MESSAGE_LENGTH_SIZE { + // Not enough data + return Ok(None); + } + + let n = LittleEndian::read_u32(buf.as_ref()); + + // Consume the length field + let _ = buf.split_to(MESSAGE_LENGTH_SIZE); + + Ok(Some(n as usize)) + } + + fn decode_data(&mut self, buf: &mut BytesMut, n: usize) -> io::Result<Option<Out>> + where + Out: DeserializeOwned + Debug, + { + // 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); + } + + let buf = buf.split_to(n).freeze(); + + trace!("Attempting to decode"); + let msg = deserialize::<Out>(buf.as_ref()).map_err(|e| match *e { + bincode::ErrorKind::Io(e) => e, + _ => io::Error::new(io::ErrorKind::Other, *e), + })?; + + 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) => { + self.state = State::Data(n); + + // Ensure that the buffer has enough space to read the + // incoming payload + buf.reserve(n); + + 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 head + buf.reserve(MESSAGE_LENGTH_SIZE); + + Ok(Some(data)) + } + None => Ok(None), + } + } + + fn encode(&mut self, item: Self::In, buf: &mut BytesMut) -> io::Result<()> { + trace!("Attempting to encode"); + let encoded_len = serialized_size(&item).unwrap(); + if encoded_len > MAX_MESSAGE_LEN { + trace!("oversized message {}", encoded_len); + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "encoded message too big", + )); + } + + buf.reserve((encoded_len as usize) + MESSAGE_LENGTH_SIZE); + + buf.put_u32_le(encoded_len as u32); + + if let Err(e) = bincode::config() + .limit(encoded_len) + .serialize_into::<_, Self::In>(&mut buf.writer(), &item) + { + match *e { + bincode::ErrorKind::Io(e) => return Err(e), + _ => return Err(io::Error::new(io::ErrorKind::Other, *e)), + } + } + + Ok(()) + } +} diff --git a/media/audioipc/audioipc/src/core.rs b/media/audioipc/audioipc/src/core.rs new file mode 100644 index 0000000000..0edb02b6e0 --- /dev/null +++ b/media/audioipc/audioipc/src/core.rs @@ -0,0 +1,85 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +// Ease accessing reactor::Core handles. + +use futures::sync::oneshot; +use std::sync::mpsc; +use std::{fmt, io, thread}; +use tokio::runtime::current_thread; + +struct Inner { + join: thread::JoinHandle<()>, + shutdown: oneshot::Sender<()>, +} + +pub struct CoreThread { + inner: Option<Inner>, + handle: current_thread::Handle, +} + +impl CoreThread { + pub fn handle(&self) -> current_thread::Handle { + self.handle.clone() + } +} + +impl Drop for CoreThread { + fn drop(&mut self) { + debug!("Shutting down {:?}", self); + if let Some(inner) = self.inner.take() { + let _ = inner.shutdown.send(()); + drop(inner.join.join()); + } + } +} + +impl fmt::Debug for CoreThread { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // f.debug_tuple("CoreThread").field(&"...").finish() + f.debug_tuple("CoreThread").field(&self.handle).finish() + } +} + +pub fn spawn_thread<S, F, D>(name: S, f: F, d: D) -> io::Result<CoreThread> +where + S: Into<String>, + F: FnOnce() -> io::Result<()> + Send + 'static, + D: FnOnce() -> () + Send + 'static, +{ + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + let (handle_tx, handle_rx) = mpsc::channel::<current_thread::Handle>(); + + let join = thread::Builder::new().name(name.into()).spawn(move || { + let mut rt = + current_thread::Runtime::new().expect("Failed to create current_thread::Runtime"); + let handle = rt.handle(); + drop(handle_tx.send(handle.clone())); + + rt.spawn(futures::future::lazy(|| { + let _ = f(); + Ok(()) + })); + + let _ = rt.block_on(shutdown_rx); + trace!("thread shutdown..."); + d(); + })?; + + let handle = handle_rx.recv().or_else(|_| { + Err(io::Error::new( + io::ErrorKind::Other, + "Failed to receive remote handle from spawned thread", + )) + })?; + + Ok(CoreThread { + inner: Some(Inner { + join, + shutdown: shutdown_tx, + }), + handle, + }) +} diff --git a/media/audioipc/audioipc/src/errors.rs b/media/audioipc/audioipc/src/errors.rs new file mode 100644 index 0000000000..5cc37503f9 --- /dev/null +++ b/media/audioipc/audioipc/src/errors.rs @@ -0,0 +1,22 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +use bincode; +use cubeb; +use std; + +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/media/audioipc/audioipc/src/fd_passing.rs b/media/audioipc/audioipc/src/fd_passing.rs new file mode 100644 index 0000000000..411dcfd803 --- /dev/null +++ b/media/audioipc/audioipc/src/fd_passing.rs @@ -0,0 +1,363 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::async_msg::{AsyncRecvMsg, AsyncSendMsg}; +use crate::cmsg; +use crate::codec::Codec; +use crate::messages::AssocRawPlatformHandle; +use bytes::{Bytes, BytesMut, IntoBuf}; +use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream}; +use std::collections::VecDeque; +use std::os::unix::io::RawFd; +use std::{fmt, io, mem}; + +const INITIAL_CAPACITY: usize = 1024; +const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY; +const FDS_CAPACITY: usize = 16; + +struct IncomingFds { + cmsg: BytesMut, + recv_fds: Option<cmsg::ControlMsgIter>, +} + +impl IncomingFds { + pub fn new(c: usize) -> Self { + let capacity = c * cmsg::space(mem::size_of::<[RawFd; 3]>()); + IncomingFds { + cmsg: BytesMut::with_capacity(capacity), + recv_fds: None, + } + } + + pub fn take_fds(&mut self) -> Option<[RawFd; 3]> { + loop { + let fds = self + .recv_fds + .as_mut() + .and_then(|recv_fds| recv_fds.next()) + .and_then(|fds| Some(clone_into_array(&fds))); + + if fds.is_some() { + return fds; + } + + if self.cmsg.is_empty() { + return None; + } + + self.recv_fds = Some(cmsg::iterator(self.cmsg.take().freeze())); + } + } + + pub fn cmsg(&mut self) -> &mut BytesMut { + self.cmsg.reserve(cmsg::space(mem::size_of::<[RawFd; 3]>())); + &mut self.cmsg + } +} + +#[derive(Debug)] +struct Frame { + msgs: Bytes, + fds: Option<Bytes>, +} + +/// A unified `Stream` and `Sink` interface over an I/O object, using +/// the `Codec` trait to encode and decode the payload. +pub struct FramedWithPlatformHandles<A, C> { + io: A, + codec: C, + // Stream + read_buf: BytesMut, + incoming_fds: IncomingFds, + is_readable: bool, + eof: bool, + // Sink + frames: VecDeque<Frame>, + write_buf: BytesMut, + outgoing_fds: BytesMut, +} + +impl<A, C> FramedWithPlatformHandles<A, C> +where + A: AsyncSendMsg, +{ + // If there is a buffered frame, try to write it to `A` + fn do_write(&mut self) -> Poll<(), io::Error> { + trace!("do_write..."); + // Create a frame from any pending message in `write_buf`. + if !self.write_buf.is_empty() { + self.set_frame(None); + } + + trace!("pending frames: {:?}", self.frames); + + let mut processed = 0; + + loop { + let n = match self.frames.front() { + Some(frame) => { + trace!("sending msg {:?}, fds {:?}", frame.msgs, frame.fds); + let mut msgs = frame.msgs.clone().into_buf(); + let fds = match frame.fds { + Some(ref fds) => fds.clone(), + None => Bytes::new(), + } + .into_buf(); + try_ready!(self.io.send_msg_buf(&mut msgs, &fds)) + } + _ => { + // No pending frames. + return Ok(().into()); + } + }; + + match self.frames.pop_front() { + Some(mut frame) => { + processed += 1; + + // Close any fds that have been sent. The fds are + // encoded in cmsg format inside frame.fds. Use + // the cmsg iterator to access msg and extract + // RawFds. + frame.fds.take().and_then(|cmsg| { + for fds in cmsg::iterator(cmsg) { + close_fds(&*fds) + } + Some(()) + }); + + if n != frame.msgs.len() { + // If only part of the message was sent then + // re-queue the remaining message at the head + // of the queue. (Don't need to resend the fds + // since they've been sent with the first + // part.) + drop(frame.msgs.split_to(n)); + self.frames.push_front(frame); + break; + } + } + _ => panic!(), + } + } + trace!("process {} frames", processed); + trace!("pending frames: {:?}", self.frames); + + Ok(().into()) + } + + fn set_frame(&mut self, fds: Option<Bytes>) { + if self.write_buf.is_empty() { + assert!(fds.is_none()); + trace!("set_frame: No pending messages..."); + return; + } + + let msgs = self.write_buf.take().freeze(); + trace!("set_frame: msgs={:?} fds={:?}", msgs, fds); + + self.frames.push_back(Frame { msgs, fds }); + } +} + +impl<A, C> Stream for FramedWithPlatformHandles<A, C> +where + A: AsyncRecvMsg, + C: Codec, + C::Out: AssocRawPlatformHandle, +{ + type Item = C::Out; + type Error = io::Error; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + loop { + // Repeatedly call `decode` or `decode_eof` as long as it is + // "readable". Readable is defined as not having returned `None`. If + // the upstream has returned EOF, and the decoder is no longer + // readable, it can be assumed that the decoder will never become + // readable again, at which point the stream is terminated. + if self.is_readable { + if self.eof { + let mut item = self.codec.decode_eof(&mut self.read_buf)?; + item.take_platform_handles(|| self.incoming_fds.take_fds()); + return Ok(Some(item).into()); + } + + trace!("attempting to decode a frame"); + + if let Some(mut item) = self.codec.decode(&mut self.read_buf)? { + trace!("frame decoded from buffer"); + item.take_platform_handles(|| self.incoming_fds.take_fds()); + return Ok(Some(item).into()); + } + + self.is_readable = false; + } + + assert!(!self.eof); + + // Otherwise, try to read more data and try again. Make sure we've + // got room for at least one byte to read to ensure that we don't + // get a spurious 0 that looks like EOF + let (n, _) = try_ready!(self + .io + .recv_msg_buf(&mut self.read_buf, self.incoming_fds.cmsg())); + + if n == 0 { + self.eof = true; + } + + self.is_readable = true; + } + } +} + +impl<A, C> Sink for FramedWithPlatformHandles<A, C> +where + A: AsyncSendMsg, + C: Codec, + C::In: AssocRawPlatformHandle + fmt::Debug, +{ + type SinkItem = C::In; + type SinkError = io::Error; + + fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> { + trace!("start_send: item={:?}", item); + + // If the buffer is already over BACKPRESSURE_THRESHOLD, + // then attempt to flush it. If after flush it's *still* + // over BACKPRESSURE_THRESHOLD, then reject the send. + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + self.poll_complete()?; + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + return Ok(AsyncSink::NotReady(item)); + } + } + + let fds = item.platform_handles(); + self.codec.encode(item, &mut self.write_buf)?; + let fds = fds.and_then(|fds| { + cmsg::builder(&mut self.outgoing_fds) + .rights(&fds.0[..]) + .finish() + .ok() + }); + + trace!("item fds: {:?}", fds); + + if fds.is_some() { + // Enforce splitting sends on messages that contain file + // descriptors. + self.set_frame(fds); + } + + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), Self::SinkError> { + trace!("flushing framed transport"); + + try_ready!(self.do_write()); + + try_nb!(self.io.flush()); + + trace!("framed transport flushed"); + Ok(().into()) + } + + fn close(&mut self) -> Poll<(), Self::SinkError> { + if task::is_in_task() { + try_ready!(self.poll_complete()); + } + self.io.shutdown() + } +} + +pub fn framed_with_platformhandles<A, C>(io: A, codec: C) -> FramedWithPlatformHandles<A, C> { + FramedWithPlatformHandles { + io, + codec, + read_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + incoming_fds: IncomingFds::new(FDS_CAPACITY), + is_readable: false, + eof: false, + frames: VecDeque::new(), + write_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + outgoing_fds: BytesMut::with_capacity( + FDS_CAPACITY * cmsg::space(mem::size_of::<[RawFd; 3]>()), + ), + } +} + +fn clone_into_array<A, T>(slice: &[T]) -> A +where + A: Sized + Default + AsMut<[T]>, + T: Clone, +{ + let mut a = Default::default(); + <A as AsMut<[T]>>::as_mut(&mut a).clone_from_slice(slice); + a +} + +fn close_fds(fds: &[RawFd]) { + for fd in fds { + unsafe { + super::close_platformhandle(*fd); + } + } +} + +#[cfg(test)] +mod tests { + use bytes::BufMut; + use libc; + use std; + + extern "C" { + fn cmsghdr_bytes(size: *mut libc::size_t) -> *const u8; + } + + fn cmsg_bytes() -> &'static [u8] { + let mut size = 0; + unsafe { + let ptr = cmsghdr_bytes(&mut size); + std::slice::from_raw_parts(ptr, size) + } + } + + #[test] + fn single_cmsg() { + let mut incoming = super::IncomingFds::new(16); + + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fds().is_some()); + assert!(incoming.take_fds().is_none()); + } + + #[test] + fn multiple_cmsg_1() { + let mut incoming = super::IncomingFds::new(16); + + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fds().is_some()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fds().is_some()); + assert!(incoming.take_fds().is_none()); + } + + #[test] + fn multiple_cmsg_2() { + let mut incoming = super::IncomingFds::new(16); + println!("cmsg_bytes() {}", cmsg_bytes().len()); + + incoming.cmsg().put_slice(cmsg_bytes()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fds().is_some()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fds().is_some()); + assert!(incoming.take_fds().is_some()); + assert!(incoming.take_fds().is_none()); + } +} diff --git a/media/audioipc/audioipc/src/frame.rs b/media/audioipc/audioipc/src/frame.rs new file mode 100644 index 0000000000..fa5b0bd4bd --- /dev/null +++ b/media/audioipc/audioipc/src/frame.rs @@ -0,0 +1,165 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::codec::Codec; +use bytes::{Buf, Bytes, BytesMut, IntoBuf}; +use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream}; +use std::io; +use tokio_io::{AsyncRead, AsyncWrite}; + +const INITIAL_CAPACITY: usize = 1024; +const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY; + +/// A unified `Stream` and `Sink` interface over an I/O object, using +/// the `Codec` trait to encode and decode the payload. +pub struct Framed<A, C> { + io: A, + codec: C, + read_buf: BytesMut, + write_buf: BytesMut, + frame: Option<<Bytes as IntoBuf>::Buf>, + is_readable: bool, + eof: bool, +} + +impl<A, C> Framed<A, C> +where + A: AsyncWrite, +{ + // If there is a buffered frame, try to write it to `A` + fn do_write(&mut self) -> Poll<(), io::Error> { + loop { + if self.frame.is_none() { + self.set_frame(); + } + + if self.frame.is_none() { + return Ok(().into()); + } + + let done = { + let frame = self.frame.as_mut().unwrap(); + try_ready!(self.io.write_buf(frame)); + !frame.has_remaining() + }; + + if done { + self.frame = None; + } + } + } + + fn set_frame(&mut self) { + if self.write_buf.is_empty() { + return; + } + + debug_assert!(self.frame.is_none()); + + self.frame = Some(self.write_buf.take().freeze().into_buf()); + } +} + +impl<A, C> Stream for Framed<A, C> +where + A: AsyncRead, + C: Codec, +{ + type Item = C::Out; + type Error = io::Error; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + loop { + // Repeatedly call `decode` or `decode_eof` as long as it is + // "readable". Readable is defined as not having returned `None`. If + // the upstream has returned EOF, and the decoder is no longer + // readable, it can be assumed that the decoder will never become + // readable again, at which point the stream is terminated. + if self.is_readable { + if self.eof { + let frame = self.codec.decode_eof(&mut self.read_buf)?; + return Ok(Some(frame).into()); + } + + trace!("attempting to decode a frame"); + + if let Some(frame) = self.codec.decode(&mut self.read_buf)? { + trace!("frame decoded from buffer"); + return Ok(Some(frame).into()); + } + + self.is_readable = false; + } + + assert!(!self.eof); + + // XXX(kinetik): work around tokio_named_pipes assuming at least 1kB available. + self.read_buf.reserve(INITIAL_CAPACITY); + + // Otherwise, try to read more data and try again. Make sure we've + // got room for at least one byte to read to ensure that we don't + // get a spurious 0 that looks like EOF + if try_ready!(self.io.read_buf(&mut self.read_buf)) == 0 { + self.eof = true; + } + + self.is_readable = true; + } + } +} + +impl<A, C> Sink for Framed<A, C> +where + A: AsyncWrite, + C: Codec, +{ + type SinkItem = C::In; + type SinkError = io::Error; + + fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> { + // If the buffer is already over BACKPRESSURE_THRESHOLD, + // then attempt to flush it. If after flush it's *still* + // over BACKPRESSURE_THRESHOLD, then reject the send. + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + self.poll_complete()?; + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + return Ok(AsyncSink::NotReady(item)); + } + } + + self.codec.encode(item, &mut self.write_buf)?; + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), Self::SinkError> { + trace!("flushing framed transport"); + + try_ready!(self.do_write()); + + try_nb!(self.io.flush()); + + trace!("framed transport flushed"); + Ok(().into()) + } + + fn close(&mut self) -> Poll<(), Self::SinkError> { + if task::is_in_task() { + try_ready!(self.poll_complete()); + } + self.io.shutdown() + } +} + +pub fn framed<A, C>(io: A, codec: C) -> Framed<A, C> { + Framed { + io, + codec, + read_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + write_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + frame: None, + is_readable: false, + eof: false, + } +} diff --git a/media/audioipc/audioipc/src/handle_passing.rs b/media/audioipc/audioipc/src/handle_passing.rs new file mode 100644 index 0000000000..347613e5f5 --- /dev/null +++ b/media/audioipc/audioipc/src/handle_passing.rs @@ -0,0 +1,278 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::codec::Codec; +use crate::messages::AssocRawPlatformHandle; +use bytes::{Bytes, BytesMut, IntoBuf}; +use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream}; +use std::collections::VecDeque; +use std::{fmt, io}; +use tokio_io::{AsyncRead, AsyncWrite}; + +const INITIAL_CAPACITY: usize = 1024; +const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY; + +#[derive(Debug)] +struct Frame { + msgs: Bytes, +} + +/// A unified `Stream` and `Sink` interface over an I/O object, using +/// the `Codec` trait to encode and decode the payload. +pub struct FramedWithPlatformHandles<A, C> { + io: A, + codec: C, + // Stream + read_buf: BytesMut, + is_readable: bool, + eof: bool, + // Sink + frames: VecDeque<Frame>, + write_buf: BytesMut, +} + +impl<A, C> FramedWithPlatformHandles<A, C> +where + A: AsyncWrite, +{ + // If there is a buffered frame, try to write it to `A` + fn do_write(&mut self) -> Poll<(), io::Error> { + trace!("do_write..."); + // Create a frame from any pending message in `write_buf`. + if !self.write_buf.is_empty() { + self.set_frame(); + } + + trace!("pending frames: {:?}", self.frames); + + let mut processed = 0; + + loop { + let n = match self.frames.front() { + Some(frame) => { + trace!("sending msg {:?}", frame.msgs); + let mut msgs = frame.msgs.clone().into_buf(); + try_ready!(self.io.write_buf(&mut msgs)) + } + _ => { + // No pending frames. + return Ok(().into()); + } + }; + + match self.frames.pop_front() { + Some(mut frame) => { + processed += 1; + + if n != frame.msgs.len() { + // If only part of the message was sent then + // re-queue the remaining message at the head + // of the queue. (Don't need to resend the + // handles since they've been sent with the + // first part.) + drop(frame.msgs.split_to(n)); + self.frames.push_front(frame); + break; + } + } + _ => panic!(), + } + } + trace!("process {} frames", processed); + trace!("pending frames: {:?}", self.frames); + + Ok(().into()) + } + + fn set_frame(&mut self) { + if self.write_buf.is_empty() { + trace!("set_frame: No pending messages..."); + return; + } + + let msgs = self.write_buf.take().freeze(); + trace!("set_frame: msgs={:?}", msgs); + + self.frames.push_back(Frame { msgs }); + } +} + +impl<A, C> Stream for FramedWithPlatformHandles<A, C> +where + A: AsyncRead, + C: Codec, + C::Out: AssocRawPlatformHandle, +{ + type Item = C::Out; + type Error = io::Error; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + loop { + // Repeatedly call `decode` or `decode_eof` as long as it is + // "readable". Readable is defined as not having returned `None`. If + // the upstream has returned EOF, and the decoder is no longer + // readable, it can be assumed that the decoder will never become + // readable again, at which point the stream is terminated. + if self.is_readable { + if self.eof { + let item = self.codec.decode_eof(&mut self.read_buf)?; + return Ok(Some(item).into()); + } + + trace!("attempting to decode a frame"); + + if let Some(item) = self.codec.decode(&mut self.read_buf)? { + trace!("frame decoded from buffer"); + return Ok(Some(item).into()); + } + + self.is_readable = false; + } + + assert!(!self.eof); + + // XXX(kinetik): work around tokio_named_pipes assuming at least 1kB available. + self.read_buf.reserve(INITIAL_CAPACITY); + + // Otherwise, try to read more data and try again. Make sure we've + // got room for at least one byte to read to ensure that we don't + // get a spurious 0 that looks like EOF + let n = try_ready!(self.io.read_buf(&mut self.read_buf)); + + if n == 0 { + self.eof = true; + } + + self.is_readable = true; + } + } +} + +impl<A, C> Sink for FramedWithPlatformHandles<A, C> +where + A: AsyncWrite, + C: Codec, + C::In: AssocRawPlatformHandle + fmt::Debug, +{ + type SinkItem = C::In; + type SinkError = io::Error; + + fn start_send( + &mut self, + mut item: Self::SinkItem, + ) -> StartSend<Self::SinkItem, Self::SinkError> { + trace!("start_send: item={:?}", item); + + // If the buffer is already over BACKPRESSURE_THRESHOLD, + // then attempt to flush it. If after flush it's *still* + // over BACKPRESSURE_THRESHOLD, then reject the send. + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + self.poll_complete()?; + if self.write_buf.len() > BACKPRESSURE_THRESHOLD { + return Ok(AsyncSink::NotReady(item)); + } + } + + let mut got_handles = false; + if let Some((handles, target_pid)) = item.platform_handles() { + got_handles = true; + let remote_handles = unsafe { + // Attempt to duplicate all 3 handles before checking + // result, since we rely on duplicate_platformhandle closing + // our source handles. + let r1 = duplicate_platformhandle(handles[0], target_pid); + let r2 = duplicate_platformhandle(handles[1], target_pid); + let r3 = duplicate_platformhandle(handles[2], target_pid); + [r1?, r2?, r3?] + }; + trace!( + "item handles: {:?} remote_handles: {:?}", + handles, + remote_handles + ); + item.take_platform_handles(|| Some(remote_handles)); + } + + self.codec.encode(item, &mut self.write_buf)?; + + if got_handles { + // Enforce splitting sends on messages that contain file + // descriptors. + self.set_frame(); + } + + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), Self::SinkError> { + trace!("flushing framed transport"); + + try_ready!(self.do_write()); + + try_nb!(self.io.flush()); + + trace!("framed transport flushed"); + Ok(().into()) + } + + fn close(&mut self) -> Poll<(), Self::SinkError> { + if task::is_in_task() { + try_ready!(self.poll_complete()); + } + self.io.shutdown() + } +} + +pub fn framed_with_platformhandles<A, C>(io: A, codec: C) -> FramedWithPlatformHandles<A, C> { + FramedWithPlatformHandles { + io, + codec, + read_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + is_readable: false, + eof: false, + frames: VecDeque::new(), + write_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + } +} + +use super::PlatformHandleType; +use winapi::shared::minwindef::{DWORD, FALSE}; +use winapi::um::{handleapi, processthreadsapi, winnt}; + +// source_handle is effectively taken ownership of (consumed) and +// closed when duplicate_platformhandle is called. +// TODO: Make this transfer more explicit via the type system. +unsafe fn duplicate_platformhandle( + source_handle: PlatformHandleType, + target_pid: DWORD, +) -> Result<PlatformHandleType, std::io::Error> { + let source = processthreadsapi::GetCurrentProcess(); + let target = processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, target_pid); + if !super::valid_handle(target) { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "invalid target process", + )); + } + + let mut target_handle = std::ptr::null_mut(); + let ok = handleapi::DuplicateHandle( + source, + source_handle, + target, + &mut target_handle, + 0, + FALSE, + winnt::DUPLICATE_CLOSE_SOURCE | winnt::DUPLICATE_SAME_ACCESS, + ); + handleapi::CloseHandle(target); + if ok == FALSE { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "DuplicateHandle failed", + )); + } + Ok(target_handle) +} diff --git a/media/audioipc/audioipc/src/lib.rs b/media/audioipc/audioipc/src/lib.rs new file mode 100644 index 0000000000..1c056bc123 --- /dev/null +++ b/media/audioipc/audioipc/src/lib.rs @@ -0,0 +1,212 @@ +// 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; +#[macro_use] +extern crate serde_derive; +#[macro_use] +extern crate futures; +#[macro_use] +extern crate tokio_io; + +mod async_msg; +#[cfg(unix)] +mod cmsg; +pub mod codec; +pub mod core; +#[allow(deprecated)] +pub mod errors; +#[cfg(unix)] +pub mod fd_passing; +#[cfg(unix)] +pub use crate::fd_passing as platformhandle_passing; +#[cfg(windows)] +pub mod handle_passing; +#[cfg(windows)] +pub use handle_passing as platformhandle_passing; +pub mod frame; +pub mod messages; +#[cfg(unix)] +mod msg; +pub mod rpc; +pub mod shm; + +// TODO: Remove local fork when https://github.com/tokio-rs/tokio/pull/1294 is resolved. +#[cfg(unix)] +mod tokio_uds_stream; + +#[cfg(windows)] +mod tokio_named_pipes; + +pub use crate::messages::{ClientMessage, ServerMessage}; + +// TODO: Remove hardcoded size and allow allocation based on cubeb backend requirements. +pub const SHM_AREA_SIZE: usize = 2 * 1024 * 1024; + +#[cfg(unix)] +use std::os::unix::io::{FromRawFd, IntoRawFd}; +#[cfg(windows)] +use std::os::windows::io::{FromRawHandle, IntoRawHandle}; + +use std::cell::RefCell; + +// 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(Clone, Debug)] +pub struct PlatformHandle(RefCell<Inner>); + +#[derive(Clone, Debug)] +struct Inner { + handle: PlatformHandleType, + owned: bool, +} + +unsafe impl Send for PlatformHandle {} + +pub const INVALID_HANDLE_VALUE: PlatformHandleType = -1isize as PlatformHandleType; + +// Custom serialization to treat HANDLEs as i64. This is not valid in +// general, but after sending the HANDLE value to a remote process we +// use it to create a valid HANDLE via DuplicateHandle. +// To avoid duplicating the serialization code, we're lazy and treat +// file descriptors as i64 rather than i32. +impl serde::Serialize for PlatformHandle { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: serde::Serializer, + { + let h = self.0.borrow(); + serializer.serialize_i64(h.handle as i64) + } +} + +struct PlatformHandleVisitor; +impl<'de> serde::de::Visitor<'de> for PlatformHandleVisitor { + type Value = PlatformHandle; + + 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, + { + let owned = cfg!(windows); + Ok(PlatformHandle::new(value as PlatformHandleType, owned)) + } +} + +impl<'de> serde::Deserialize<'de> for PlatformHandle { + fn deserialize<D>(deserializer: D) -> Result<PlatformHandle, D::Error> + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_i64(PlatformHandleVisitor) + } +} + +#[cfg(unix)] +fn valid_handle(handle: PlatformHandleType) -> bool { + handle >= 0 +} + +#[cfg(windows)] +fn valid_handle(handle: PlatformHandleType) -> bool { + const NULL_HANDLE_VALUE: PlatformHandleType = 0isize as PlatformHandleType; + handle != INVALID_HANDLE_VALUE && handle != NULL_HANDLE_VALUE +} + +impl PlatformHandle { + pub fn new(raw: PlatformHandleType, owned: bool) -> PlatformHandle { + assert!(valid_handle(raw)); + let inner = Inner { + handle: raw, + owned: owned, + }; + PlatformHandle(RefCell::new(inner)) + } + + #[cfg(windows)] + pub fn from<T: IntoRawHandle>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_handle(), true) + } + + #[cfg(unix)] + pub fn from<T: IntoRawFd>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_fd(), true) + } + + #[cfg(windows)] + pub unsafe fn into_file(&self) -> std::fs::File { + std::fs::File::from_raw_handle(self.into_raw()) + } + + #[cfg(unix)] + pub unsafe fn into_file(&self) -> std::fs::File { + std::fs::File::from_raw_fd(self.into_raw()) + } + + pub unsafe fn into_raw(&self) -> PlatformHandleType { + let mut h = self.0.borrow_mut(); + assert!(h.owned); + h.owned = false; + h.handle + } +} + +impl Drop for PlatformHandle { + fn drop(&mut self) { + let inner = self.0.borrow(); + if inner.owned { + unsafe { close_platformhandle(inner.handle) } + } + } +} + +#[cfg(unix)] +unsafe fn close_platformhandle(handle: PlatformHandleType) { + libc::close(handle); +} + +#[cfg(windows)] +unsafe fn close_platformhandle(handle: PlatformHandleType) { + winapi::um::handleapi::CloseHandle(handle); +} + +#[cfg(unix)] +pub mod messagestream_unix; +#[cfg(unix)] +pub use crate::messagestream_unix::*; + +#[cfg(windows)] +pub mod messagestream_win; +#[cfg(windows)] +pub use messagestream_win::*; + +#[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/media/audioipc/audioipc/src/messages.rs b/media/audioipc/audioipc/src/messages.rs new file mode 100644 index 0000000000..94843e629c --- /dev/null +++ b/media/audioipc/audioipc/src/messages.rs @@ -0,0 +1,388 @@ +// 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; +#[cfg(target_os = "linux")] +use audio_thread_priority::RtPriorityThreadInfo; +use cubeb::{self, ffi}; +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 { + pub output_name: Option<Vec<u8>>, + 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, + pub device_id: Option<Vec<u8>>, + pub friendly_name: Option<Vec<u8>>, + pub group_id: Option<Vec<u8>>, + 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<'a> From<&'a 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 { + 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 platform_handles: [PlatformHandle; 3], + pub target_pid: u32, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RegisterDeviceCollectionChanged { + pub platform_handles: [PlatformHandle; 3], + pub target_pid: u32, +} + +// 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), + StreamResetDefaultDevice(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, + StreamDestroyed, + + StreamStarted, + StreamStopped, + StreamDefaultDeviceReset, + 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, +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionReq { + DeviceChange(ffi::cubeb_device_type), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionResp { + DeviceChange, +} + +pub trait AssocRawPlatformHandle { + fn platform_handles(&self) -> Option<([PlatformHandleType; 3], u32)> { + None + } + + fn take_platform_handles<F>(&mut self, f: F) + where + F: FnOnce() -> Option<[PlatformHandleType; 3]>, + { + assert!(f().is_none()); + } +} + +impl AssocRawPlatformHandle for ServerMessage {} + +impl AssocRawPlatformHandle for ClientMessage { + fn platform_handles(&self) -> Option<([PlatformHandleType; 3], u32)> { + unsafe { + match *self { + ClientMessage::StreamCreated(ref data) => Some(( + [ + data.platform_handles[0].into_raw(), + data.platform_handles[1].into_raw(), + data.platform_handles[2].into_raw(), + ], + data.target_pid, + )), + ClientMessage::ContextSetupDeviceCollectionCallback(ref data) => Some(( + [ + data.platform_handles[0].into_raw(), + data.platform_handles[1].into_raw(), + data.platform_handles[2].into_raw(), + ], + data.target_pid, + )), + _ => None, + } + } + } + + fn take_platform_handles<F>(&mut self, f: F) + where + F: FnOnce() -> Option<[PlatformHandleType; 3]>, + { + let owned = cfg!(unix); + match *self { + ClientMessage::StreamCreated(ref mut data) => { + let handles = + f().expect("platform_handles must be available when processing StreamCreated"); + data.platform_handles = [ + PlatformHandle::new(handles[0], owned), + PlatformHandle::new(handles[1], owned), + PlatformHandle::new(handles[2], owned), + ] + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + let handles = f().expect("platform_handles must be available when processing ContextSetupDeviceCollectionCallback"); + data.platform_handles = [ + PlatformHandle::new(handles[0], owned), + PlatformHandle::new(handles[1], owned), + PlatformHandle::new(handles[2], owned), + ] + } + _ => {} + } + } +} + +#[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 mut raw = ffi::cubeb_stream_params::default(); + raw.format = ffi::CUBEB_SAMPLE_FLOAT32BE; + raw.rate = 96_000; + raw.channels = 32; + raw.layout = ffi::CUBEB_LAYOUT_3F1_LFE; + raw.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/media/audioipc/audioipc/src/messagestream_unix.rs b/media/audioipc/audioipc/src/messagestream_unix.rs new file mode 100644 index 0000000000..4ce1e96d7b --- /dev/null +++ b/media/audioipc/audioipc/src/messagestream_unix.rs @@ -0,0 +1,105 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use super::tokio_uds_stream as tokio_uds; +use futures::Poll; +use mio::Ready; +use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd}; +use std::os::unix::net; +use tokio_io::{AsyncRead, AsyncWrite}; + +#[derive(Debug)] +pub struct MessageStream(net::UnixStream); +pub struct AsyncMessageStream(tokio_uds::UnixStream); + +impl MessageStream { + fn new(stream: net::UnixStream) -> MessageStream { + MessageStream(stream) + } + + pub fn anonymous_ipc_pair( + ) -> std::result::Result<(MessageStream, MessageStream), std::io::Error> { + let pair = net::UnixStream::pair()?; + Ok((MessageStream::new(pair.0), MessageStream::new(pair.1))) + } + + pub unsafe fn from_raw_fd(raw: super::PlatformHandleType) -> MessageStream { + MessageStream::new(net::UnixStream::from_raw_fd(raw)) + } + + pub fn into_tokio_ipc( + self, + handle: &tokio::reactor::Handle, + ) -> std::result::Result<AsyncMessageStream, std::io::Error> { + Ok(AsyncMessageStream::new(tokio_uds::UnixStream::from_std( + self.0, handle, + )?)) + } +} + +impl IntoRawFd for MessageStream { + fn into_raw_fd(self) -> RawFd { + self.0.into_raw_fd() + } +} + +impl AsyncMessageStream { + fn new(stream: tokio_uds::UnixStream) -> AsyncMessageStream { + AsyncMessageStream(stream) + } + + pub fn poll_read_ready(&self, ready: Ready) -> Poll<Ready, std::io::Error> { + self.0.poll_read_ready(ready) + } + + pub fn clear_read_ready(&self, ready: Ready) -> Result<(), std::io::Error> { + self.0.clear_read_ready(ready) + } + + pub fn poll_write_ready(&self) -> Poll<Ready, std::io::Error> { + self.0.poll_write_ready() + } + + pub fn clear_write_ready(&self) -> Result<(), std::io::Error> { + self.0.clear_write_ready() + } +} + +impl std::io::Read for AsyncMessageStream { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { + self.0.read(buf) + } +} + +impl std::io::Write for AsyncMessageStream { + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { + self.0.write(buf) + } + fn flush(&mut self) -> std::io::Result<()> { + self.0.flush() + } +} + +impl AsyncRead for AsyncMessageStream { + fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> { + <&tokio_uds::UnixStream>::read_buf(&mut &self.0, buf) + } +} + +impl AsyncWrite for AsyncMessageStream { + fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> { + <&tokio_uds::UnixStream>::shutdown(&mut &self.0) + } + + fn write_buf<B: bytes::Buf>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> { + <&tokio_uds::UnixStream>::write_buf(&mut &self.0, buf) + } +} + +impl AsRawFd for AsyncMessageStream { + fn as_raw_fd(&self) -> RawFd { + self.0.as_raw_fd() + } +} diff --git a/media/audioipc/audioipc/src/messagestream_win.rs b/media/audioipc/audioipc/src/messagestream_win.rs new file mode 100644 index 0000000000..9f2a6f0bd8 --- /dev/null +++ b/media/audioipc/audioipc/src/messagestream_win.rs @@ -0,0 +1,111 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use super::tokio_named_pipes; +use mio_named_pipes; +use std::os::windows::fs::*; +use std::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use tokio_io::{AsyncRead, AsyncWrite}; +use winapi::um::winbase::FILE_FLAG_OVERLAPPED; + +#[derive(Debug)] +pub struct MessageStream(miow::pipe::NamedPipe); +pub struct AsyncMessageStream(tokio_named_pipes::NamedPipe); + +impl MessageStream { + fn new(stream: miow::pipe::NamedPipe) -> MessageStream { + MessageStream(stream) + } + + pub fn anonymous_ipc_pair( + ) -> std::result::Result<(MessageStream, MessageStream), std::io::Error> { + let pipe_name = get_pipe_name(); + let pipe_server = miow::pipe::NamedPipe::new(&pipe_name)?; + let pipe_client = { + let mut opts = std::fs::OpenOptions::new(); + opts.read(true) + .write(true) + .custom_flags(FILE_FLAG_OVERLAPPED); + let file = opts.open(&pipe_name)?; + unsafe { miow::pipe::NamedPipe::from_raw_handle(file.into_raw_handle()) } + }; + Ok(( + MessageStream::new(pipe_server), + MessageStream::new(pipe_client), + )) + } + + pub unsafe fn from_raw_fd(raw: super::PlatformHandleType) -> MessageStream { + MessageStream::new(miow::pipe::NamedPipe::from_raw_handle(raw)) + } + + pub fn into_tokio_ipc( + self, + handle: &tokio::reactor::Handle, + ) -> std::result::Result<AsyncMessageStream, std::io::Error> { + let pipe = unsafe { mio_named_pipes::NamedPipe::from_raw_handle(self.into_raw_handle()) }; + Ok(AsyncMessageStream::new( + tokio_named_pipes::NamedPipe::from_pipe(pipe, handle)?, + )) + } +} + +impl IntoRawHandle for MessageStream { + fn into_raw_handle(self) -> RawHandle { + self.0.into_raw_handle() + } +} + +impl AsyncMessageStream { + fn new(stream: tokio_named_pipes::NamedPipe) -> AsyncMessageStream { + AsyncMessageStream(stream) + } +} + +impl std::io::Read for AsyncMessageStream { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { + self.0.read(buf) + } +} + +impl std::io::Write for AsyncMessageStream { + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { + self.0.write(buf) + } + fn flush(&mut self) -> std::io::Result<()> { + self.0.flush() + } +} + +impl AsyncRead for AsyncMessageStream { + fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> { + <tokio_named_pipes::NamedPipe>::read_buf(&mut self.0, buf) + } +} + +impl AsyncWrite for AsyncMessageStream { + fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> { + <tokio_named_pipes::NamedPipe>::shutdown(&mut self.0) + } + + fn write_buf<B: bytes::Buf>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> { + <tokio_named_pipes::NamedPipe>::write_buf(&mut self.0, buf) + } +} + +impl AsRawHandle for AsyncMessageStream { + fn as_raw_handle(&self) -> RawHandle { + self.0.as_raw_handle() + } +} + +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::SeqCst); + format!("\\\\.\\pipe\\cubeb-pipe-{}-{}", pid, pipe_id) +} diff --git a/media/audioipc/audioipc/src/msg.rs b/media/audioipc/audioipc/src/msg.rs new file mode 100644 index 0000000000..d41cc58703 --- /dev/null +++ b/media/audioipc/audioipc/src/msg.rs @@ -0,0 +1,116 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +use iovec::unix; +use iovec::IoVec; +use libc; +use std::os::unix::io::{AsRawFd, RawFd}; +use std::{cmp, io, mem, ptr}; + +// Extend sys::os::unix::net::UnixStream to support sending and receiving a single file desc. +// We can extend UnixStream by using traits, eliminating the need to introduce a new wrapped +// UnixStream type. +pub trait RecvMsg { + fn recv_msg( + &mut self, + iov: &mut [&mut IoVec], + cmsg: &mut [u8], + ) -> io::Result<(usize, usize, i32)>; +} + +pub trait SendMsg { + fn send_msg(&mut self, iov: &[&IoVec], cmsg: &[u8]) -> io::Result<usize>; +} + +impl<T: AsRawFd> RecvMsg for T { + fn recv_msg( + &mut self, + iov: &mut [&mut IoVec], + cmsg: &mut [u8], + ) -> io::Result<(usize, usize, i32)> { + #[cfg(target_os = "linux")] + let flags = libc::MSG_CMSG_CLOEXEC; + #[cfg(not(target_os = "linux"))] + let flags = 0; + recv_msg_with_flags(self.as_raw_fd(), iov, cmsg, flags) + } +} + +impl<T: AsRawFd> SendMsg for T { + fn send_msg(&mut self, iov: &[&IoVec], cmsg: &[u8]) -> io::Result<usize> { + send_msg_with_flags(self.as_raw_fd(), iov, cmsg, 0) + } +} + +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 fn recv_msg_with_flags( + socket: RawFd, + bufs: &mut [&mut IoVec], + cmsg: &mut [u8], + 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.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_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) })?; + + let controllen = msghdr.msg_controllen as usize; + Ok((n, controllen, msghdr.msg_flags)) +} + +pub 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/media/audioipc/audioipc/src/rpc/client/mod.rs b/media/audioipc/audioipc/src/rpc/client/mod.rs new file mode 100644 index 0000000000..0fb8aed565 --- /dev/null +++ b/media/audioipc/audioipc/src/rpc/client/mod.rs @@ -0,0 +1,162 @@ +// This is a derived version of simple/pipeline/client.rs from +// tokio_proto crate used under MIT license. +// +// Original version of client.rs: +// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3 +// +// The following modifications were made: +// * Simplify the code to implement RPC for pipeline requests that +// contain simple request/response messages: +// * Remove `Error` types, +// * Remove `bind_transport` fn & `BindTransport` type, +// * Remove all "Lift"ing functionality. +// * Remove `Service` trait since audioipc doesn't use `tokio_service` +// crate. +// +// Copyright (c) 2016 Tokio contributors +// +// Permission is hereby granted, free of charge, to any +// person obtaining a copy of this software and associated +// documentation files (the "Software"), to deal in the +// Software without restriction, including without +// limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of +// the Software, and to permit persons to whom the Software +// is furnished to do so, subject to the following +// conditions: +// +// The above copyright notice and this permission notice +// shall be included in all copies or substantial portions +// of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// DEALINGS IN THE SOFTWARE. + +use crate::rpc::driver::Driver; +use crate::rpc::Handler; +use futures::sync::oneshot; +use futures::{Async, Future, Poll, Sink, Stream}; +use std::collections::VecDeque; +use std::io; +use tokio::runtime::current_thread; + +mod proxy; + +pub use self::proxy::{ClientProxy, Response}; + +pub fn bind_client<C>(transport: C::Transport) -> proxy::ClientProxy<C::Request, C::Response> +where + C: Client, +{ + let (tx, rx) = proxy::channel(); + + let fut = { + let handler = ClientHandler::<C> { + transport, + requests: rx, + in_flight: VecDeque::with_capacity(32), + }; + Driver::new(handler) + }; + + // Spawn the RPC driver into task + current_thread::spawn(fut.map_err(|_| ())); + + tx +} + +pub trait Client: 'static { + /// Request + type Request: 'static; + + /// Response + type Response: 'static; + + /// The message transport, which works with async I/O objects of type `A` + type Transport: 'static + + Stream<Item = Self::Response, Error = io::Error> + + Sink<SinkItem = Self::Request, SinkError = io::Error>; +} + +//////////////////////////////////////////////////////////////////////////////// + +struct ClientHandler<C> +where + C: Client, +{ + transport: C::Transport, + requests: proxy::Receiver<C::Request, C::Response>, + in_flight: VecDeque<oneshot::Sender<C::Response>>, +} + +impl<C> Handler for ClientHandler<C> +where + C: Client, +{ + type In = C::Response; + type Out = C::Request; + type Transport = C::Transport; + + fn transport(&mut self) -> &mut Self::Transport { + &mut self.transport + } + + fn consume(&mut self, response: Self::In) -> io::Result<()> { + trace!("ClientHandler::consume"); + if let Some(complete) = self.in_flight.pop_front() { + drop(complete.send(response)); + } else { + return Err(io::Error::new( + io::ErrorKind::Other, + "request / response mismatch", + )); + } + + Ok(()) + } + + /// Produce a message + fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error> { + trace!("ClientHandler::produce"); + + // Try to get a new request + match self.requests.poll() { + Ok(Async::Ready(Some((request, complete)))) => { + trace!(" --> received request"); + + // Track complete handle + self.in_flight.push_back(complete); + + Ok(Some(request).into()) + } + Ok(Async::Ready(None)) => { + trace!(" --> client dropped"); + Ok(None.into()) + } + Ok(Async::NotReady) => { + trace!(" --> not ready"); + Ok(Async::NotReady) + } + Err(_) => unreachable!(), + } + } + + /// RPC currently in flight + fn has_in_flight(&self) -> bool { + !self.in_flight.is_empty() + } +} + +impl<C: Client> Drop for ClientHandler<C> { + fn drop(&mut self) { + let _ = self.transport.close(); + self.in_flight.clear(); + } +} diff --git a/media/audioipc/audioipc/src/rpc/client/proxy.rs b/media/audioipc/audioipc/src/rpc/client/proxy.rs new file mode 100644 index 0000000000..bd44110d59 --- /dev/null +++ b/media/audioipc/audioipc/src/rpc/client/proxy.rs @@ -0,0 +1,136 @@ +// This is a derived version of client_proxy.rs from +// tokio_proto crate used under MIT license. +// +// Original version of client_proxy.rs: +// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3 +// +// The following modifications were made: +// * Remove `Service` trait since audioipc doesn't use `tokio_service` +// crate. +// * Remove `RefCell` from `ClientProxy` since cubeb is called from +// multiple threads. `mpsc::UnboundedSender` is thread safe. +// * Simplify the interface to just request (`R`) and response (`Q`) +// removing error (`E`). +// * Remove the `Envelope` type. +// * Renamed `pair` to `channel` to represent that an `rpc::channel` +// is being created. +// +// Original License: +// +// Copyright (c) 2016 Tokio contributors +// +// Permission is hereby granted, free of charge, to any +// person obtaining a copy of this software and associated +// documentation files (the "Software"), to deal in the +// Software without restriction, including without +// limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of +// the Software, and to permit persons to whom the Software +// is furnished to do so, subject to the following +// conditions: +// +// The above copyright notice and this permission notice +// shall be included in all copies or substantial portions +// of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// DEALINGS IN THE SOFTWARE. + +use futures::sync::{mpsc, oneshot}; +use futures::{Async, Future, Poll}; +use std::fmt; +use std::io; + +/// Message used to dispatch requests to the task managing the +/// client connection. +pub type Request<R, Q> = (R, oneshot::Sender<Q>); + +/// Receive requests submitted to the client +pub type Receiver<R, Q> = mpsc::UnboundedReceiver<Request<R, Q>>; + +/// Response future returned from a client +pub struct Response<Q> { + inner: oneshot::Receiver<Q>, +} + +pub struct ClientProxy<R, Q> { + tx: mpsc::UnboundedSender<Request<R, Q>>, +} + +impl<R, Q> Clone for ClientProxy<R, Q> { + fn clone(&self) -> Self { + ClientProxy { + tx: self.tx.clone(), + } + } +} + +pub fn channel<R, Q>() -> (ClientProxy<R, Q>, Receiver<R, Q>) { + // Create a channel to send the requests to client-side of rpc. + let (tx, rx) = mpsc::unbounded(); + + // Wrap the `tx` part in ClientProxy so the rpc call interface + // can be implemented. + let client = ClientProxy { tx }; + + (client, rx) +} + +impl<R, Q> ClientProxy<R, Q> { + pub fn call(&self, request: R) -> Response<Q> { + // The response to returned from the rpc client task over a + // oneshot channel. + let (tx, rx) = oneshot::channel(); + + // If send returns an Err, its because the other side has been dropped. + // By ignoring it, we are just dropping the `tx`, which will mean the + // rx will return Canceled when polled. In turn, that is translated + // into a BrokenPipe, which conveys the proper error. + let _ = self.tx.unbounded_send((request, tx)); + + Response { inner: rx } + } +} + +impl<R, Q> fmt::Debug for ClientProxy<R, Q> +where + R: fmt::Debug, + Q: fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "ClientProxy {{ ... }}") + } +} + +impl<Q> Future for Response<Q> { + type Item = Q; + type Error = io::Error; + + fn poll(&mut self) -> Poll<Q, io::Error> { + match self.inner.poll() { + Ok(Async::Ready(res)) => Ok(Async::Ready(res)), + Ok(Async::NotReady) => Ok(Async::NotReady), + // Convert oneshot::Canceled into io::Error + Err(_) => { + let e = io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe"); + Err(e) + } + } + } +} + +impl<Q> fmt::Debug for Response<Q> +where + Q: fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Response {{ ... }}") + } +} diff --git a/media/audioipc/audioipc/src/rpc/driver.rs b/media/audioipc/audioipc/src/rpc/driver.rs new file mode 100644 index 0000000000..91463deaa7 --- /dev/null +++ b/media/audioipc/audioipc/src/rpc/driver.rs @@ -0,0 +1,173 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use crate::rpc::Handler; +use futures::{Async, AsyncSink, Future, Poll, Sink, Stream}; +use std::fmt; +use std::io; + +pub struct Driver<T> +where + T: Handler, +{ + // Glue + handler: T, + + // True as long as the connection has more request frames to read. + run: bool, + + // True when the transport is fully flushed + is_flushed: bool, +} + +impl<T> Driver<T> +where + T: Handler, +{ + /// Create a new rpc driver with the given service and transport. + pub fn new(handler: T) -> Driver<T> { + Driver { + handler, + run: true, + is_flushed: true, + } + } + + /// Returns true if the driver has nothing left to do + fn is_done(&self) -> bool { + !self.run && self.is_flushed && !self.has_in_flight() + } + + /// Process incoming messages off the transport. + fn receive_incoming(&mut self) -> io::Result<()> { + while self.run { + if let Async::Ready(req) = self.handler.transport().poll()? { + self.process_incoming(req)?; + } else { + break; + } + } + Ok(()) + } + + /// Process an incoming message + fn process_incoming(&mut self, req: Option<T::In>) -> io::Result<()> { + trace!("process_incoming"); + // At this point, the service & transport are ready to process the + // request, no matter what it is. + match req { + Some(message) => { + trace!("received message"); + + if let Err(e) = self.handler.consume(message) { + // TODO: Should handler be infalliable? + panic!("unimplemented error handling: {:?}", e); + } + } + None => { + trace!("received None"); + // At this point, we just return. This works + // because poll with be called again and go + // through the receive-cycle again. + self.run = false; + } + } + + Ok(()) + } + + /// Send outgoing messages to the transport. + fn send_outgoing(&mut self) -> io::Result<()> { + trace!("send_responses"); + loop { + match self.handler.produce()? { + Async::Ready(Some(message)) => { + trace!(" --> got message"); + self.process_outgoing(message)?; + } + Async::Ready(None) => { + trace!(" --> got None"); + // The service is done with the connection. + self.run = false; + break; + } + // Nothing to dispatch + Async::NotReady => break, + } + } + + Ok(()) + } + + fn process_outgoing(&mut self, message: T::Out) -> io::Result<()> { + trace!("process_outgoing"); + assert_send(&mut self.handler.transport(), message)?; + + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.is_flushed = self.handler.transport().poll_complete()?.is_ready(); + + // TODO: + Ok(()) + } + + fn has_in_flight(&self) -> bool { + self.handler.has_in_flight() + } +} + +impl<T> Future for Driver<T> +where + T: Handler, +{ + type Item = (); + type Error = io::Error; + + fn poll(&mut self) -> Poll<(), Self::Error> { + trace!("rpc::Driver::tick"); + + // First read off data from the socket + self.receive_incoming()?; + + // Handle completed responses + self.send_outgoing()?; + + // Try flushing buffered writes + self.flush()?; + + if self.is_done() { + trace!(" --> is done."); + return Ok(().into()); + } + + // Tick again later + Ok(Async::NotReady) + } +} + +fn assert_send<S: Sink>(s: &mut S, item: S::SinkItem) -> Result<(), S::SinkError> { + match s.start_send(item)? { + AsyncSink::Ready => Ok(()), + AsyncSink::NotReady(_) => panic!( + "sink reported itself as ready after `poll_ready` but was \ + then unable to accept a message" + ), + } +} + +impl<T> fmt::Debug for Driver<T> +where + T: Handler + fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("rpc::Handler") + .field("handler", &self.handler) + .field("run", &self.run) + .field("is_flushed", &self.is_flushed) + .finish() + } +} diff --git a/media/audioipc/audioipc/src/rpc/mod.rs b/media/audioipc/audioipc/src/rpc/mod.rs new file mode 100644 index 0000000000..8d54fc6e00 --- /dev/null +++ b/media/audioipc/audioipc/src/rpc/mod.rs @@ -0,0 +1,36 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details + +use futures::{Poll, Sink, Stream}; +use std::io; + +mod client; +mod driver; +mod server; + +pub use self::client::{bind_client, Client, ClientProxy, Response}; +pub use self::server::{bind_server, Server}; + +pub trait Handler { + /// Message type read from transport + type In; + /// Message type written to transport + type Out; + type Transport: 'static + + Stream<Item = Self::In, Error = io::Error> + + Sink<SinkItem = Self::Out, SinkError = io::Error>; + + /// Mutable reference to the transport + fn transport(&mut self) -> &mut Self::Transport; + + /// Consume a request + fn consume(&mut self, message: Self::In) -> io::Result<()>; + + /// Produce a response + fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error>; + + /// RPC currently in flight + fn has_in_flight(&self) -> bool; +} diff --git a/media/audioipc/audioipc/src/rpc/server.rs b/media/audioipc/audioipc/src/rpc/server.rs new file mode 100644 index 0000000000..1cb037bd8a --- /dev/null +++ b/media/audioipc/audioipc/src/rpc/server.rs @@ -0,0 +1,184 @@ +// This is a derived version of simple/pipeline/server.rs from +// tokio_proto crate used under MIT license. +// +// Original version of server.rs: +// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3 +// +// The following modifications were made: +// * Simplify the code to implement RPC for pipeline requests that +// contain simple request/response messages: +// * Remove `Error` types, +// * Remove `bind_transport` fn & `BindTransport` type, +// * Remove all "Lift"ing functionality. +// * Remove `Service` trait since audioipc doesn't use `tokio_service` +// crate. +// +// Copyright (c) 2016 Tokio contributors +// +// Permission is hereby granted, free of charge, to any +// person obtaining a copy of this software and associated +// documentation files (the "Software"), to deal in the +// Software without restriction, including without +// limitation the rights to use, copy, modify, merge, +// publish, distribute, sublicense, and/or sell copies of +// the Software, and to permit persons to whom the Software +// is furnished to do so, subject to the following +// conditions: +// +// The above copyright notice and this permission notice +// shall be included in all copies or substantial portions +// of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +// DEALINGS IN THE SOFTWARE. + +use crate::rpc::driver::Driver; +use crate::rpc::Handler; +use futures::{Async, Future, Poll, Sink, Stream}; +use std::collections::VecDeque; +use std::io; +use tokio::runtime::current_thread; + +/// Bind an async I/O object `io` to the `server`. +pub fn bind_server<S>(transport: S::Transport, server: S) +where + S: Server, +{ + let fut = { + let handler = ServerHandler { + server, + transport, + in_flight: VecDeque::with_capacity(32), + }; + Driver::new(handler) + }; + + // Spawn the RPC driver into task + current_thread::spawn(fut.map_err(|_| ())) +} + +pub trait Server: 'static { + /// Request + type Request: 'static; + + /// Response + type Response: 'static; + + /// Future + type Future: Future<Item = Self::Response, Error = ()>; + + /// The message transport, which works with async I/O objects of + /// type `A`. + type Transport: 'static + + Stream<Item = Self::Request, Error = io::Error> + + Sink<SinkItem = Self::Response, SinkError = io::Error>; + + /// Process the request and return the response asynchronously. + fn process(&mut self, req: Self::Request) -> Self::Future; +} + +//////////////////////////////////////////////////////////////////////////////// + +struct ServerHandler<S> +where + S: Server, +{ + // The service handling the connection + server: S, + // The transport responsible for sending/receving messages over the wire + transport: S::Transport, + // FIFO of "in flight" responses to requests. + in_flight: VecDeque<InFlight<S::Future>>, +} + +impl<S> Handler for ServerHandler<S> +where + S: Server, +{ + type In = S::Request; + type Out = S::Response; + type Transport = S::Transport; + + /// Mutable reference to the transport + fn transport(&mut self) -> &mut Self::Transport { + &mut self.transport + } + + /// Consume a message + fn consume(&mut self, request: Self::In) -> io::Result<()> { + trace!("ServerHandler::consume"); + let response = self.server.process(request); + self.in_flight.push_back(InFlight::Active(response)); + + // TODO: Should the error be handled differently? + Ok(()) + } + + /// Produce a message + fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error> { + trace!("ServerHandler::produce"); + + // Make progress on pending responses + for pending in &mut self.in_flight { + pending.poll(); + } + + // Is the head of the queue ready? + match self.in_flight.front() { + Some(&InFlight::Done(_)) => {} + _ => { + trace!(" --> not ready"); + return Ok(Async::NotReady); + } + } + + // Return the ready response + match self.in_flight.pop_front() { + Some(InFlight::Done(res)) => { + trace!(" --> received response"); + Ok(Async::Ready(Some(res))) + } + _ => panic!(), + } + } + + /// RPC currently in flight + fn has_in_flight(&self) -> bool { + !self.in_flight.is_empty() + } +} + +impl<S: Server> Drop for ServerHandler<S> { + fn drop(&mut self) { + let _ = self.transport.close(); + self.in_flight.clear(); + } +} + +//////////////////////////////////////////////////////////////////////////////// + +enum InFlight<F: Future<Error = ()>> { + Active(F), + Done(F::Item), +} + +impl<F: Future<Error = ()>> InFlight<F> { + fn poll(&mut self) { + let res = match *self { + InFlight::Active(ref mut f) => match f.poll() { + Ok(Async::Ready(e)) => e, + Err(_) => unreachable!(), + Ok(Async::NotReady) => return, + }, + _ => return, + }; + *self = InFlight::Done(res); + } +} diff --git a/media/audioipc/audioipc/src/shm.rs b/media/audioipc/audioipc/src/shm.rs new file mode 100644 index 0000000000..99012bfab4 --- /dev/null +++ b/media/audioipc/audioipc/src/shm.rs @@ -0,0 +1,266 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +use crate::errors::*; +use memmap::{Mmap, MmapMut, MmapOptions}; +use std::cell::UnsafeCell; +use std::convert::TryInto; +use std::env::temp_dir; +use std::fs::{remove_file, File, OpenOptions}; +use std::sync::{atomic, Arc}; + +fn open_shm_file(id: &str) -> Result<File> { + #[cfg(target_os = "linux")] + { + let id_cstring = std::ffi::CString::new(id).unwrap(); + unsafe { + let r = libc::syscall(libc::SYS_memfd_create, id_cstring.as_ptr(), 0); + if r >= 0 { + use std::os::unix::io::FromRawFd as _; + 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); + } + } + + 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(unix)] +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(unix)] +fn allocate_file(file: &File, size: usize) -> Result<()> { + use std::os::unix::io::AsRawFd; + + // 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(()) +} + +#[cfg(windows)] +fn allocate_file(file: &File, size: usize) -> Result<()> { + // CreateFileMapping will ensure the entire file is allocated + // before it's mapped in, so we simply set the size here. + file.set_len(size.try_into().unwrap())?; + Ok(()) +} + +pub struct SharedMemReader { + mmap: Mmap, +} + +impl SharedMemReader { + pub fn new(id: &str, size: usize) -> Result<(SharedMemReader, File)> { + let file = open_shm_file(id)?; + allocate_file(&file, size)?; + let mmap = unsafe { MmapOptions::new().map(&file)? }; + assert_eq!(mmap.len(), size); + Ok((SharedMemReader { mmap }, file)) + } + + pub fn read(&self, buf: &mut [u8]) -> Result<()> { + if buf.is_empty() { + return Ok(()); + } + // TODO: Track how much is in the shm area. + if buf.len() <= self.mmap.len() { + atomic::fence(atomic::Ordering::Acquire); + let len = buf.len(); + buf.copy_from_slice(&self.mmap[..len]); + Ok(()) + } else { + bail!("mmap size"); + } + } +} + +pub struct SharedMemSlice { + mmap: Arc<Mmap>, +} + +impl SharedMemSlice { + pub fn from(file: &File, size: usize) -> Result<SharedMemSlice> { + let mmap = unsafe { MmapOptions::new().map(file)? }; + assert_eq!(mmap.len(), size); + let mmap = Arc::new(mmap); + Ok(SharedMemSlice { mmap }) + } + + pub fn get_slice(&self, size: usize) -> Result<&[u8]> { + if size == 0 { + return Ok(&[]); + } + // TODO: Track how much is in the shm area. + if size <= self.mmap.len() { + atomic::fence(atomic::Ordering::Acquire); + let buf = &self.mmap[..size]; + Ok(buf) + } else { + bail!("mmap size"); + } + } + + /// Clones the memory map. + /// + /// The underlying memory map is shared, and thus the caller must + /// ensure that the memory is not illegally aliased. + pub unsafe fn unsafe_clone(&self) -> Self { + SharedMemSlice { + mmap: self.mmap.clone(), + } + } +} + +unsafe impl Send for SharedMemSlice {} + +pub struct SharedMemWriter { + mmap: MmapMut, +} + +impl SharedMemWriter { + pub fn new(id: &str, size: usize) -> Result<(SharedMemWriter, File)> { + let file = open_shm_file(id)?; + allocate_file(&file, size)?; + let mmap = unsafe { MmapOptions::new().map_mut(&file)? }; + assert_eq!(mmap.len(), size); + Ok((SharedMemWriter { mmap }, file)) + } + + pub fn write(&mut self, buf: &[u8]) -> Result<()> { + if buf.is_empty() { + return Ok(()); + } + // TODO: Track how much is in the shm area. + if buf.len() <= self.mmap.len() { + self.mmap[..buf.len()].copy_from_slice(buf); + atomic::fence(atomic::Ordering::Release); + Ok(()) + } else { + bail!("mmap size"); + } + } +} + +pub struct SharedMemMutSlice { + mmap: Arc<UnsafeCell<MmapMut>>, +} + +impl SharedMemMutSlice { + pub fn from(file: &File, size: usize) -> Result<SharedMemMutSlice> { + let mmap = unsafe { MmapOptions::new().map_mut(file)? }; + assert_eq!(mmap.len(), size); + let mmap = Arc::new(UnsafeCell::new(mmap)); + Ok(SharedMemMutSlice { mmap }) + } + + pub fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + if size == 0 { + return Ok(&mut []); + } + // TODO: Track how much is in the shm area. + if size <= self.inner().len() { + let buf = &mut self.inner_mut()[..size]; + atomic::fence(atomic::Ordering::Release); + Ok(buf) + } else { + bail!("mmap size"); + } + } + + /// Clones the memory map. + /// + /// The underlying memory map is shared, and thus the caller must + /// ensure that the memory is not illegally aliased. + pub unsafe fn unsafe_clone(&self) -> Self { + SharedMemMutSlice { + mmap: self.mmap.clone(), + } + } + + fn inner(&self) -> &MmapMut { + unsafe { &*self.mmap.get() } + } + + fn inner_mut(&mut self) -> &mut MmapMut { + unsafe { &mut *self.mmap.get() } + } +} + +unsafe impl Send for SharedMemMutSlice {} diff --git a/media/audioipc/audioipc/src/tokio_named_pipes.rs b/media/audioipc/audioipc/src/tokio_named_pipes.rs new file mode 100644 index 0000000000..d402337fc7 --- /dev/null +++ b/media/audioipc/audioipc/src/tokio_named_pipes.rs @@ -0,0 +1,155 @@ +// Copied from tokio-named-pipes/src/lib.rs revision 49ec1ba8bbc94ab6fc9636af2a00dfb3204080c8 (tokio-named-pipes 0.2.0) +// This file is dual licensed under the MIT and Apache-2.0 per upstream: https://github.com/NikVolf/tokio-named-pipes/blob/stable/LICENSE-MIT and https://github.com/NikVolf/tokio-named-pipes/blob/stable/LICENSE-APACHE +// - Implement AsyncWrite::shutdown + +#![cfg(windows)] + +use std::ffi::OsStr; +use std::fmt; +use std::io::{Read, Write}; +use std::os::windows::io::*; + +use bytes::{Buf, BufMut}; +use futures::{Async, Poll}; +use mio::Ready; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::reactor::{Handle, PollEvented2}; + +pub struct NamedPipe { + io: PollEvented2<mio_named_pipes::NamedPipe>, +} + +impl NamedPipe { + pub fn new<P: AsRef<OsStr>>(p: P, handle: &Handle) -> std::io::Result<NamedPipe> { + NamedPipe::_new(p.as_ref(), handle) + } + + fn _new(p: &OsStr, handle: &Handle) -> std::io::Result<NamedPipe> { + let inner = mio_named_pipes::NamedPipe::new(p)?; + NamedPipe::from_pipe(inner, handle) + } + + pub fn from_pipe( + pipe: mio_named_pipes::NamedPipe, + handle: &Handle, + ) -> std::io::Result<NamedPipe> { + Ok(NamedPipe { + io: PollEvented2::new_with_handle(pipe, handle)?, + }) + } + + pub fn connect(&self) -> std::io::Result<()> { + self.io.get_ref().connect() + } + + pub fn disconnect(&self) -> std::io::Result<()> { + self.io.get_ref().disconnect() + } + + pub fn poll_read_ready_readable(&mut self) -> tokio::io::Result<Async<Ready>> { + self.io.poll_read_ready(Ready::readable()) + } + + pub fn poll_write_ready(&mut self) -> tokio::io::Result<Async<Ready>> { + self.io.poll_write_ready() + } + + fn io_mut(&mut self) -> &mut PollEvented2<mio_named_pipes::NamedPipe> { + &mut self.io + } +} + +impl Read for NamedPipe { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { + self.io.read(buf) + } +} + +impl Write for NamedPipe { + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { + self.io.write(buf) + } + fn flush(&mut self) -> std::io::Result<()> { + self.io.flush() + } +} + +impl<'a> Read for &'a NamedPipe { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { + (&self.io).read(buf) + } +} + +impl<'a> Write for &'a NamedPipe { + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { + (&self.io).write(buf) + } + + fn flush(&mut self) -> std::io::Result<()> { + (&self.io).flush() + } +} + +impl AsyncRead for NamedPipe { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + + fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, std::io::Error> { + if let Async::NotReady = self.io.poll_read_ready(Ready::readable())? { + return Ok(Async::NotReady); + } + + let mut stack_buf = [0u8; 1024]; + let bytes_read = self.io_mut().read(&mut stack_buf); + match bytes_read { + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + self.io_mut().clear_read_ready(Ready::readable())?; + return Ok(Async::NotReady); + } + Err(e) => Err(e), + Ok(bytes_read) => { + buf.put_slice(&stack_buf[0..bytes_read]); + Ok(Async::Ready(bytes_read)) + } + } + } +} + +impl AsyncWrite for NamedPipe { + fn shutdown(&mut self) -> Poll<(), std::io::Error> { + let _ = self.disconnect(); + Ok(().into()) + } + + fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, std::io::Error> { + if let Async::NotReady = self.io.poll_write_ready()? { + return Ok(Async::NotReady); + } + + let bytes_wrt = self.io_mut().write(buf.bytes()); + match bytes_wrt { + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + self.io_mut().clear_write_ready()?; + return Ok(Async::NotReady); + } + Err(e) => Err(e), + Ok(bytes_wrt) => { + buf.advance(bytes_wrt); + Ok(Async::Ready(bytes_wrt)) + } + } + } +} + +impl fmt::Debug for NamedPipe { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + self.io.get_ref().fmt(f) + } +} + +impl AsRawHandle for NamedPipe { + fn as_raw_handle(&self) -> RawHandle { + self.io.get_ref().as_raw_handle() + } +} diff --git a/media/audioipc/audioipc/src/tokio_uds_stream.rs b/media/audioipc/audioipc/src/tokio_uds_stream.rs new file mode 100644 index 0000000000..db005d7287 --- /dev/null +++ b/media/audioipc/audioipc/src/tokio_uds_stream.rs @@ -0,0 +1,363 @@ +// Copied from tokio-uds/src/stream.rs revision 25e835c5b7e2cfeb9c22b1fd576844f6814a9477 (tokio-uds 0.2.5) +// License MIT per upstream: https://github.com/tokio-rs/tokio/blob/master/tokio-uds/LICENSE +// - Removed ucred for build simplicity +// - Added clear_{read,write}_ready per: https://github.com/tokio-rs/tokio/pull/1294 + +use tokio_io::{AsyncRead, AsyncWrite}; +use tokio_reactor::{Handle, PollEvented}; + +use bytes::{Buf, BufMut}; +use futures::{Async, Future, Poll}; +use iovec::{self, IoVec}; +use libc; +use mio::Ready; +use mio_uds; + +use std::fmt; +use std::io::{self, Read, Write}; +use std::net::Shutdown; +use std::os::unix::io::{AsRawFd, RawFd}; +use std::os::unix::net::{self, SocketAddr}; +use std::path::Path; + +/// A structure representing a connected Unix socket. +/// +/// This socket can be connected directly with `UnixStream::connect` or accepted +/// from a listener with `UnixListener::incoming`. Additionally, a pair of +/// anonymous Unix sockets can be created with `UnixStream::pair`. +pub struct UnixStream { + io: PollEvented<mio_uds::UnixStream>, +} + +/// Future returned by `UnixStream::connect` which will resolve to a +/// `UnixStream` when the stream is connected. +#[derive(Debug)] +pub struct ConnectFuture { + inner: State, +} + +#[derive(Debug)] +enum State { + Waiting(UnixStream), + Error(io::Error), + Empty, +} + +impl UnixStream { + /// Connects to the socket named by `path`. + /// + /// This function will create a new Unix socket and connect to the path + /// specified, associating the returned stream with the default event loop's + /// handle. + pub fn connect<P>(path: P) -> ConnectFuture + where + P: AsRef<Path>, + { + let res = mio_uds::UnixStream::connect(path).map(UnixStream::new); + + let inner = match res { + Ok(stream) => State::Waiting(stream), + Err(e) => State::Error(e), + }; + + ConnectFuture { inner } + } + + /// Consumes a `UnixStream` in the standard library and returns a + /// nonblocking `UnixStream` from this crate. + /// + /// The returned stream will be associated with the given event loop + /// specified by `handle` and is ready to perform I/O. + pub fn from_std(stream: net::UnixStream, handle: &Handle) -> io::Result<UnixStream> { + let stream = mio_uds::UnixStream::from_stream(stream)?; + let io = PollEvented::new_with_handle(stream, handle)?; + + Ok(UnixStream { io }) + } + + /// Creates an unnamed pair of connected sockets. + /// + /// This function will create a pair of interconnected Unix sockets for + /// communicating back and forth between one another. Each socket will + /// be associated with the default event loop's handle. + pub fn pair() -> io::Result<(UnixStream, UnixStream)> { + let (a, b) = mio_uds::UnixStream::pair()?; + let a = UnixStream::new(a); + let b = UnixStream::new(b); + + Ok((a, b)) + } + + pub(crate) fn new(stream: mio_uds::UnixStream) -> UnixStream { + let io = PollEvented::new(stream); + UnixStream { io } + } + + /// Test whether this socket is ready to be read or not. + pub fn poll_read_ready(&self, ready: Ready) -> Poll<Ready, io::Error> { + self.io.poll_read_ready(ready) + } + + /// Clear read ready state. + pub fn clear_read_ready(&self, ready: mio::Ready) -> io::Result<()> { + self.io.clear_read_ready(ready) + } + + /// Test whether this socket is ready to be written to or not. + pub fn poll_write_ready(&self) -> Poll<Ready, io::Error> { + self.io.poll_write_ready() + } + + /// Clear write ready state. + pub fn clear_write_ready(&self) -> io::Result<()> { + self.io.clear_write_ready() + } + + /// Returns the socket address of the local half of this connection. + pub fn local_addr(&self) -> io::Result<SocketAddr> { + self.io.get_ref().local_addr() + } + + /// Returns the socket address of the remote half of this connection. + pub fn peer_addr(&self) -> io::Result<SocketAddr> { + self.io.get_ref().peer_addr() + } + + /// Returns the value of the `SO_ERROR` option. + pub fn take_error(&self) -> io::Result<Option<io::Error>> { + self.io.get_ref().take_error() + } + + /// Shuts down the read, write, or both halves of this connection. + /// + /// This function will cause all pending and future I/O calls on the + /// specified portions to immediately return with an appropriate value + /// (see the documentation of `Shutdown`). + pub fn shutdown(&self, how: Shutdown) -> io::Result<()> { + self.io.get_ref().shutdown(how) + } +} + +impl Read for UnixStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { + self.io.read(buf) + } +} + +impl Write for UnixStream { + fn write(&mut self, buf: &[u8]) -> io::Result<usize> { + self.io.write(buf) + } + fn flush(&mut self) -> io::Result<()> { + self.io.flush() + } +} + +impl AsyncRead for UnixStream { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + + fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> { + <&UnixStream>::read_buf(&mut &*self, buf) + } +} + +impl AsyncWrite for UnixStream { + fn shutdown(&mut self) -> Poll<(), io::Error> { + <&UnixStream>::shutdown(&mut &*self) + } + + fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> { + <&UnixStream>::write_buf(&mut &*self, buf) + } +} + +impl<'a> Read for &'a UnixStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { + (&self.io).read(buf) + } +} + +impl<'a> Write for &'a UnixStream { + fn write(&mut self, buf: &[u8]) -> io::Result<usize> { + (&self.io).write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + (&self.io).flush() + } +} + +impl<'a> AsyncRead for &'a UnixStream { + unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool { + false + } + + fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> { + if let Async::NotReady = <UnixStream>::poll_read_ready(self, Ready::readable())? { + return Ok(Async::NotReady); + } + unsafe { + let r = read_ready(buf, self.as_raw_fd()); + if r == -1 { + let e = io::Error::last_os_error(); + if e.kind() == io::ErrorKind::WouldBlock { + self.io.clear_read_ready(Ready::readable())?; + Ok(Async::NotReady) + } else { + Err(e) + } + } else { + let r = r as usize; + buf.advance_mut(r); + Ok(r.into()) + } + } + } +} + +impl<'a> AsyncWrite for &'a UnixStream { + fn shutdown(&mut self) -> Poll<(), io::Error> { + Ok(().into()) + } + + fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> { + if let Async::NotReady = <UnixStream>::poll_write_ready(self)? { + return Ok(Async::NotReady); + } + unsafe { + let r = write_ready(buf, self.as_raw_fd()); + if r == -1 { + let e = io::Error::last_os_error(); + if e.kind() == io::ErrorKind::WouldBlock { + self.io.clear_write_ready()?; + Ok(Async::NotReady) + } else { + Err(e) + } + } else { + let r = r as usize; + buf.advance(r); + Ok(r.into()) + } + } + } +} + +impl fmt::Debug for UnixStream { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.io.get_ref().fmt(f) + } +} + +impl AsRawFd for UnixStream { + fn as_raw_fd(&self) -> RawFd { + self.io.get_ref().as_raw_fd() + } +} + +impl Future for ConnectFuture { + type Item = UnixStream; + type Error = io::Error; + + fn poll(&mut self) -> Poll<UnixStream, io::Error> { + use std::mem; + + match self.inner { + State::Waiting(ref mut stream) => { + if let Async::NotReady = stream.io.poll_write_ready()? { + return Ok(Async::NotReady); + } + + if let Some(e) = stream.io.get_ref().take_error()? { + return Err(e); + } + } + State::Error(_) => { + let e = match mem::replace(&mut self.inner, State::Empty) { + State::Error(e) => e, + _ => unreachable!(), + }; + + return Err(e); + } + State::Empty => panic!("can't poll stream twice"), + } + + match mem::replace(&mut self.inner, State::Empty) { + State::Waiting(stream) => Ok(Async::Ready(stream)), + _ => unreachable!(), + } + } +} + +unsafe fn read_ready<B: BufMut>(buf: &mut B, raw_fd: RawFd) -> isize { + // The `IoVec` type can't have a 0-length size, so we create a bunch + // of dummy versions on the stack with 1 length which we'll quickly + // overwrite. + let b1: &mut [u8] = &mut [0]; + let b2: &mut [u8] = &mut [0]; + let b3: &mut [u8] = &mut [0]; + let b4: &mut [u8] = &mut [0]; + let b5: &mut [u8] = &mut [0]; + let b6: &mut [u8] = &mut [0]; + let b7: &mut [u8] = &mut [0]; + let b8: &mut [u8] = &mut [0]; + let b9: &mut [u8] = &mut [0]; + let b10: &mut [u8] = &mut [0]; + let b11: &mut [u8] = &mut [0]; + let b12: &mut [u8] = &mut [0]; + let b13: &mut [u8] = &mut [0]; + let b14: &mut [u8] = &mut [0]; + let b15: &mut [u8] = &mut [0]; + let b16: &mut [u8] = &mut [0]; + let mut bufs: [&mut IoVec; 16] = [ + b1.into(), + b2.into(), + b3.into(), + b4.into(), + b5.into(), + b6.into(), + b7.into(), + b8.into(), + b9.into(), + b10.into(), + b11.into(), + b12.into(), + b13.into(), + b14.into(), + b15.into(), + b16.into(), + ]; + + let n = buf.bytes_vec_mut(&mut bufs); + read_ready_vecs(&mut bufs[..n], raw_fd) +} + +unsafe fn read_ready_vecs(bufs: &mut [&mut IoVec], raw_fd: RawFd) -> isize { + let iovecs = iovec::unix::as_os_slice_mut(bufs); + + libc::readv(raw_fd, iovecs.as_ptr(), iovecs.len() as i32) +} + +unsafe fn write_ready<B: Buf>(buf: &mut B, raw_fd: RawFd) -> isize { + // The `IoVec` type can't have a zero-length size, so create a dummy + // version from a 1-length slice which we'll overwrite with the + // `bytes_vec` method. + static DUMMY: &[u8] = &[0]; + let iovec = <&IoVec>::from(DUMMY); + let mut bufs = [ + iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, + iovec, iovec, iovec, + ]; + + let n = buf.bytes_vec(&mut bufs); + write_ready_vecs(&bufs[..n], raw_fd) +} + +unsafe fn write_ready_vecs(bufs: &[&IoVec], raw_fd: RawFd) -> isize { + let iovecs = iovec::unix::as_os_slice(bufs); + + libc::writev(raw_fd, iovecs.as_ptr(), iovecs.len() as i32) +} diff --git a/media/audioipc/client/Cargo.toml b/media/audioipc/client/Cargo.toml new file mode 100644 index 0000000000..eff18b1a02 --- /dev/null +++ b/media/audioipc/client/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "audioipc-client" +version = "0.4.0" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>" + ] +description = "Cubeb Backend for talking to remote cubeb server." +edition = "2018" + +[dependencies] +audio_thread_priority = "0.23.4" +audioipc = { path="../audioipc" } +cubeb-backend = "0.8" +futures = { version="0.1.18", default-features=false, features=["use_std"] } +futures-cpupool = { version="0.1.8", default-features=false } +log = "0.4" +tokio = "0.1" diff --git a/media/audioipc/client/cbindgen.toml b/media/audioipc/client/cbindgen.toml new file mode 100644 index 0000000000..9a7c204fe4 --- /dev/null +++ b/media/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/media/audioipc/client/src/context.rs b/media/audioipc/client/src/context.rs new file mode 100644 index 0000000000..7441d0c0e0 --- /dev/null +++ b/media/audioipc/client/src/context.rs @@ -0,0 +1,451 @@ +// 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::frame::{framed, Framed}; +use audioipc::platformhandle_passing::{framed_with_platformhandles, FramedWithPlatformHandles}; +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, io, 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 = FramedWithPlatformHandles< + audioipc::AsyncMessageStream, + LengthDelimitedCodec<Self::Request, Self::Response>, + >; +} + +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<ServerMessage, ClientMessage>, + core: core::CoreThread, + cpu_pool: CpuPool, + 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 handle(&self) -> current_thread::Handle { + self.core.handle() + } + + #[doc(hidden)] + pub fn rpc(&self) -> rpc::ClientProxy<ServerMessage, ClientMessage> { + 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<ServerMessage, ClientMessage>) { + 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<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: &rpc::ClientProxy<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>>, + cpu_pool: CpuPool, +} + +impl rpc::Server for DeviceCollectionServer { + type Request = DeviceCollectionReq; + type Response = DeviceCollectionResp; + type Future = CpuFuture<Self::Response, ()>; + type Transport = + Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Response, Self::Request>>; + + 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<Context> { + fn bind_and_send_client( + stream: audioipc::AsyncMessageStream, + tx_rpc: &mpsc::Sender<rpc::ClientProxy<ServerMessage, ClientMessage>>, + ) -> io::Result<()> { + let transport = framed_with_platformhandles(stream, Default::default()); + let rpc = rpc::bind_client::<CubebClient>(transport); + // If send fails then the rx end has closed + // which is unlikely here. + let _ = tx_rpc.send(rpc); + Ok(()) + } + + 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_fd(params.server_connection) }; + + let core = core::spawn_thread( + "AudioIPC Client RPC", + move || { + let handle = reactor::Handle::default(); + + register_thread(thread_create_callback); + + server_stream + .into_tokio_ipc(&handle) + .and_then(|stream| bind_and_send_client(stream, &tx_rpc)) + }, + 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<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> = match send_recv!(self.rpc(), + ContextGetDeviceEnumeration(devtype.bits()) => + ContextEnumeratedDevices()) + { + Ok(mut v) => v.drain(..).map(|i| i.into()).collect(), + Err(e) => return Err(e), + }; + 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(); + + fn opt_stream_params(p: Option<&StreamParamsRef>) -> Option<messages::StreamParams> { + match p { + Some(p) => Some(messages::StreamParams::from(p)), + None => None, + } + } + + let stream_name = match stream_name { + Some(s) => Some(s.to_bytes_with_nul().to_vec()), + None => None, + }; + + let input_stream_params = opt_stream_params(input_stream_params); + let output_stream_params = opt_stream_params(output_stream_params); + + 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 fds = send_recv!(self.rpc(), + ContextSetupDeviceCollectionCallback => + ContextSetupDeviceCollectionCallback())?; + + let stream = + unsafe { audioipc::MessageStream::from_raw_fd(fds.platform_handles[0].into_raw()) }; + + // TODO: The lowest comms layer expects exactly 3 PlatformHandles, but we only + // need one here. Drop the dummy handles the other side sent us to discard. + unsafe { + fds.platform_handles[1].into_file(); + fds.platform_handles[2].into_file(); + } + + 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/media/audioipc/client/src/lib.rs b/media/audioipc/client/src/lib.rs new file mode 100644 index 0000000000..b9363ea2df --- /dev/null +++ b/media/audioipc/client/src/lib.rs @@ -0,0 +1,84 @@ +// 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<bool> = std::cell::RefCell::new(false)); +thread_local!(static AUDIOIPC_INIT_PARAMS: std::cell::RefCell<Option<AudioIpcInitParams>> = std::cell::RefCell::new(None)); + +// This must match the definition of AudioIpcInitParams in +// dom/media/CubebUtils.cpp in Gecko. +#[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_eq!(*b.borrow(), false); + }); +} + +#[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::<ClientContext>(c, context_name) +} diff --git a/media/audioipc/client/src/send_recv.rs b/media/audioipc/client/src/send_recv.rs new file mode 100644 index 0000000000..482b4fa6e3 --- /dev/null +++ b/media/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: E) -> Error +where + E: Into<Option<c_int>>, +{ + match e.into() { + Some(e) => unsafe { 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/media/audioipc/client/src/stream.rs b/media/audioipc/client/src/stream.rs new file mode 100644 index 0000000000..00a335c720 --- /dev/null +++ b/media/audioipc/client/src/stream.rs @@ -0,0 +1,357 @@ +// 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::frame::{framed, Framed}; +use audioipc::messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage}; +use audioipc::rpc; +use audioipc::shm::{SharedMemMutSlice, SharedMemSlice}; +use audioipc::{codec::LengthDelimitedCodec, messages::StreamCreateParams}; +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<Mutex<ffi::cubeb_device_changed_callback>>, +} + +struct CallbackServer { + input_shm: Option<SharedMemSlice>, + output_shm: Option<SharedMemMutSlice>, + data_cb: ffi::cubeb_data_callback, + state_cb: ffi::cubeb_state_callback, + user_ptr: usize, + cpu_pool: CpuPool, + device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>, +} + +impl rpc::Server for CallbackServer { + type Request = CallbackReq; + type Response = CallbackResp; + type Future = CpuFuture<Self::Response, ()>; + type Transport = + Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Response, Self::Request>>; + + 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, + ); + + // Clone values that need to be moved into the cpu pool thread. + let input_shm = match self.input_shm { + Some(ref shm) => unsafe { Some(shm.unsafe_clone()) }, + None => None, + }; + let mut output_shm = match self.output_shm { + Some(ref shm) => unsafe { Some(shm.unsafe_clone()) }, + None => None, + }; + let user_ptr = self.user_ptr; + let cb = self.data_cb.unwrap(); + + self.cpu_pool.spawn_fn(move || { + // TODO: This is proof-of-concept. Make it better. + let input_ptr: *const u8 = match input_shm { + Some(shm) => shm + .get_slice(nframes as usize * input_frame_size) + .unwrap() + .as_ptr(), + None => ptr::null(), + }; + let output_ptr: *mut u8 = match output_shm { + Some(ref mut shm) => shm + .get_mut_slice(nframes as usize * output_frame_size) + .unwrap() + .as_mut_ptr(), + None => ptr::null_mut(), + }; + + run_in_callback(|| { + let nframes = unsafe { + cb( + ptr::null_mut(), + 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) + }) + } + } + } +} + +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 data = send_recv!(rpc, StreamCreate(create_params) => StreamCreated())?; + + debug!( + "token = {}, handles = {:?}", + data.token, data.platform_handles + ); + + let has_input = init_params.input_stream_params.is_some(); + let has_output = init_params.output_stream_params.is_some(); + + let stream = + unsafe { audioipc::MessageStream::from_raw_fd(data.platform_handles[0].into_raw()) }; + + let input_file = unsafe { data.platform_handles[1].into_file() }; + let input_shm = if has_input { + match SharedMemSlice::from(&input_file, audioipc::SHM_AREA_SIZE) { + Ok(shm) => Some(shm), + Err(e) => { + debug!("Client failed to set up input shmem: {}", e); + return Err(Error::error()); + } + } + } else { + None + }; + + let output_file = unsafe { data.platform_handles[2].into_file() }; + let output_shm = if has_output { + match SharedMemMutSlice::from(&output_file, audioipc::SHM_AREA_SIZE) { + Ok(shm) => Some(shm), + Err(e) => { + debug!("Client failed to set up output shmem: {}", e); + return Err(Error::error()); + } + } + } else { + None + }; + + 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 server = CallbackServer { + input_shm, + output_shm, + data_cb: data_callback, + state_cb: state_callback, + user_ptr: user_data, + cpu_pool, + device_change_cb: device_change_cb.clone(), + }; + + 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, + })); + Ok(unsafe { Stream::from_ptr(stream as *mut _) }) + } +} + +impl<'ctx> Drop for ClientStream<'ctx> { + fn drop(&mut self) { + debug!("ClientStream dropped..."); + let rpc = self.context.rpc(); + let _ = send_recv!(rpc, StreamDestroy(self.token) => StreamDestroyed); + } +} + +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 reset_default_device(&mut self) -> Result<()> { + assert_not_in_callback(); + let rpc = self.context.rpc(); + send_recv!(rpc, StreamResetDefaultDevice(self.token) => StreamDefaultDeviceReset) + } + + 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(); + // It's all unsafe... + 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/media/audioipc/gecko.patch b/media/audioipc/gecko.patch new file mode 100644 index 0000000000..8a7d46532a --- /dev/null +++ b/media/audioipc/gecko.patch @@ -0,0 +1,15 @@ +From 2c1d5189aedc7f90b603807a36a57254dc24471c Mon Sep 17 00:00:00 2001 +From: Dan Glastonbury <dglastonbury@mozilla.com> +Date: Fri, 19 Jan 2018 11:56:58 +1000 +Subject: gecko: Change paths to vendored crates. + + +diff --git a/media/audioipc/Cargo.toml b/media/audioipc/Cargo.toml +index ede6064..d0a1979 100644 +--- a/media/audioipc/Cargo.toml ++++ b/media/audioipc/Cargo.toml +@@ -1,2 +1,2 @@ + [workspace] +-members = ["audioipc", "client", "server", "ipctest"] ++members = ["audioipc", "client", "server"] + diff --git a/media/audioipc/rustfmt.toml b/media/audioipc/rustfmt.toml new file mode 100644 index 0000000000..9e4912fff7 --- /dev/null +++ b/media/audioipc/rustfmt.toml @@ -0,0 +1,10 @@ +ideal_width = 80 +match_block_trailing_comma = true +max_width = 120 +newline_style = "Unix" +normalize_comments = false +struct_lit_multiline_style = "ForceMulti" +where_trailing_comma = true +reorder_imports = true +reorder_imported_names = true +trailing_comma = "Never"
\ No newline at end of file diff --git a/media/audioipc/server/Cargo.toml b/media/audioipc/server/Cargo.toml new file mode 100644 index 0000000000..6ae2877030 --- /dev/null +++ b/media/audioipc/server/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "audioipc-server" +version = "0.2.3" +authors = [ + "Matthew Gregan <kinetik@flim.org>", + "Dan Glastonbury <dan.glastonbury@gmail.com>" + ] +description = "Remote cubeb server" +edition = "2018" + +[dependencies] +audio_thread_priority = "0.23.4" +audioipc = { path = "../audioipc" } +cubeb-core = "0.8.0" +futures = "0.1.29" +once_cell = "1.2.0" +log = "0.4" +slab = "0.4" +tokio = "0.1" + +[dependencies.error-chain] +version = "0.11.0" +default-features = false diff --git a/media/audioipc/server/cbindgen.toml b/media/audioipc/server/cbindgen.toml new file mode 100644 index 0000000000..9a7c204fe4 --- /dev/null +++ b/media/audioipc/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", "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/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<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 { + 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<ServerWrapper> { + 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::<ServerWrapper>::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<Vec<Rc<RefCell<CubebServerCallbacks>>>>, +} + +impl CubebDeviceCollectionManager { + fn new() -> CubebDeviceCollectionManager { + CubebDeviceCollectionManager { + servers: Mutex::new(Vec::new()), + } + } + + fn register( + &mut self, + context: &cubeb::Context, + server: &Rc<RefCell<CubebServerCallbacks>>, + 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<RefCell<CubebServerCallbacks>>, + 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<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 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<cubeb::Context>, + manager: CubebDeviceCollectionManager, +} + +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 = 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<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 { + 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<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Request, Self::Response>>; +} + +struct CallbackClient; + +impl rpc::Client for CallbackClient { + type Request = CallbackReq; + type Response = CallbackResp; + type Transport = + Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Request, Self::Response>>; +} + +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<CallbackReq, CallbackResp>, +} + +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<cubeb::Stream>, + cbs: Box<ServerStreamCallbacks>, +} + +impl Drop for ServerStream { + fn drop(&mut self) { + // `stream` *must* be dropped before `cbs`. + drop(self.stream.take()); + } +} + +type StreamSlab = slab::Slab<ServerStream>; + +struct CubebServerCallbacks { + rpc: rpc::ClientProxy<DeviceCollectionReq, DeviceCollectionResp>, + 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<u32>, + cbs: Option<Rc<RefCell<CubebServerCallbacks>>>, + devidmap: DevIdMap, +} + +impl rpc::Server for CubebServer { + type Request = ServerMessage; + type Response = ClientMessage; + type Future = FutureResult<Self::Response, ()>; + type Transport = FramedWithPlatformHandles< + audioipc::AsyncMessageStream, + LengthDelimitedCodec<Self::Response, Self::Request>, + >; + + 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<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.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::<DeviceCollectionClient>(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<ClientMessage> { + 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<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()); + + 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::<CallbackClient>(transport); + drop(tx.send(rpc)); + Ok(()) + })) + .expect("Failed to spawn CallbackClient"); + + let rpc: rpc::ClientProxy<CallbackReq, CallbackResp> = 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<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.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::<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) => 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"); +} diff --git a/media/audioipc/update.sh b/media/audioipc/update.sh new file mode 100644 index 0000000000..72c871bd4a --- /dev/null +++ b/media/audioipc/update.sh @@ -0,0 +1,37 @@ +# Usage: sh update.sh <upstream_src_directory> +set -e + +cp -p $1/README.md . +cp -p $1/Cargo.toml . + +for crate in audioipc client server; do + rm -fr $crate + mkdir $crate + cp -pr $1/$crate/Cargo.toml $crate + [ -e $1/$crate/cbindgen.toml ] && cp -pr $1/$crate/cbindgen.toml $crate + cp -pr $1/$crate/src $crate +done + +rm -f audioipc/src/cmsghdr.c + +if [ -d $1/.git ]; then + rev=$(cd $1 && git rev-parse --verify HEAD) + date=$(cd $1 && git show -s --format=%ci HEAD) + dirty=$(cd $1 && git diff-index --name-only HEAD) +fi + +if [ -n "$rev" ]; then + version=$rev + if [ -n "$dirty" ]; then + version=$version-dirty + echo "WARNING: updating from a dirty git repository." + fi + sed -i.bak -e "/The git commit ID used was/ s/[0-9a-f]\{40\}\(-dirty\)\{0,1\} .\{1,100\}/$version ($date)/" README_MOZILLA + rm -f README_MOZILLA.bak +else + echo "Remember to update README_MOZILLA with the version details." +fi + +echo "Applying gecko.patch on top of $rev" +patch -p3 < gecko.patch + |