diff options
Diffstat (limited to '')
20 files changed, 3716 insertions, 0 deletions
diff --git a/third_party/rust/audioipc/src/async_msg.rs b/third_party/rust/audioipc/src/async_msg.rs new file mode 100644 index 0000000000..963cf2d533 --- /dev/null +++ b/third_party/rust/audioipc/src/async_msg.rs @@ -0,0 +1,194 @@ +// 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}; +use futures::{Async, 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), + } + } +} + +// AsyncRecvMsg/AsyncSendMsg are implemented for convenience on Windows and don't use the _cmsg parameter. +#[cfg(windows)] +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, + { + // _cmsg unused on Windows. Pass through to read_buf. + match <super::AsyncMessageStream>::read_buf(self, buf) { + Ok(Async::Ready(n)) => Ok(Async::Ready((n, 0))), + Ok(Async::NotReady) => Ok(Async::NotReady), + Err(e) => Err(e), + } + } +} + +#[cfg(windows)] +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, + { + // _cmsg unused on Windows. Pass through to write_buf. + <super::AsyncMessageStream>::write_buf(self, buf) + } +} diff --git a/third_party/rust/audioipc/src/cmsg.rs b/third_party/rust/audioipc/src/cmsg.rs new file mode 100644 index 0000000000..70cb579755 --- /dev/null +++ b/third_party/rust/audioipc/src/cmsg.rs @@ -0,0 +1,180 @@ +// 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)] + // `cmsg_len` is `usize` on some platforms, `u32` on others. + #[allow(clippy::useless_conversion)] + let cmsghdr = cmsghdr { + cmsg_len: len(msg.len()).try_into().unwrap(), + cmsg_level: 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/third_party/rust/audioipc/src/cmsghdr.c b/third_party/rust/audioipc/src/cmsghdr.c new file mode 100644 index 0000000000..82d7852867 --- /dev/null +++ b/third_party/rust/audioipc/src/cmsghdr.c @@ -0,0 +1,23 @@ +#include <sys/socket.h> +#include <inttypes.h> +#include <string.h> + +const uint8_t* +cmsghdr_bytes(size_t* size) +{ + int myfd = 0; + + static union { + uint8_t buf[CMSG_SPACE(sizeof(myfd))]; + struct cmsghdr align; + } u; + + u.align.cmsg_len = CMSG_LEN(sizeof(myfd)); + u.align.cmsg_level = SOL_SOCKET; + u.align.cmsg_type = SCM_RIGHTS; + + memcpy(CMSG_DATA(&u.align), &myfd, sizeof(myfd)); + + *size = sizeof(u); + return (const uint8_t*)&u.buf; +} diff --git a/third_party/rust/audioipc/src/codec.rs b/third_party/rust/audioipc/src/codec.rs new file mode 100644 index 0000000000..f53a13779d --- /dev/null +++ b/third_party/rust/audioipc/src/codec.rs @@ -0,0 +1,187 @@ +// 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, Options}; +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) -> Option<usize> { + if buf.len() < MESSAGE_LENGTH_SIZE { + // Not enough data + return None; + } + + let n = LittleEndian::read_u32(buf.as_ref()); + + // Consume the length field + let _ = buf.split_to(MESSAGE_LENGTH_SIZE); + + 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 = bincode::options() + .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 = bincode::options().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::options() + .with_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/third_party/rust/audioipc/src/core.rs b/third_party/rust/audioipc/src/core.rs new file mode 100644 index 0000000000..0afc1f12f3 --- /dev/null +++ b/third_party/rust/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)); + + rt.spawn(futures::future::lazy(|| { + let _ = f(); + Ok(()) + })); + + let _ = rt.block_on(shutdown_rx); + trace!("thread shutdown..."); + d(); + })?; + + let handle = handle_rx.recv().map_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/third_party/rust/audioipc/src/errors.rs b/third_party/rust/audioipc/src/errors.rs new file mode 100644 index 0000000000..f4f3e32f5f --- /dev/null +++ b/third_party/rust/audioipc/src/errors.rs @@ -0,0 +1,18 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +error_chain! { + // Maybe replace with chain_err to improve the error info. + foreign_links { + Bincode(bincode::Error); + Io(std::io::Error); + Cubeb(cubeb::Error); + } + + // Replace bail!(str) with explicit errors. + errors { + Disconnected + } +} diff --git a/third_party/rust/audioipc/src/framing.rs b/third_party/rust/audioipc/src/framing.rs new file mode 100644 index 0000000000..3af16fb2b3 --- /dev/null +++ b/third_party/rust/audioipc/src/framing.rs @@ -0,0 +1,396 @@ +// 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}; +#[cfg(unix)] +use crate::cmsg; +use crate::codec::Codec; +#[cfg(windows)] +use crate::duplicate_platform_handle; +use crate::messages::AssocRawPlatformHandle; +use bytes::{Bytes, BytesMut, IntoBuf}; +use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream}; +use std::collections::VecDeque; +#[cfg(unix)] +use std::mem; +#[cfg(unix)] +use std::os::unix::io::RawFd; +use std::{fmt, io}; + +const INITIAL_CAPACITY: usize = 1024; +const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY; + +#[cfg(unix)] +struct IncomingFd { + cmsg: BytesMut, + recv_fd: Option<cmsg::ControlMsgIter>, +} + +#[cfg(unix)] +impl IncomingFd { + pub fn new() -> Self { + IncomingFd { + cmsg: BytesMut::with_capacity(cmsg::space(mem::size_of::<RawFd>())), + recv_fd: None, + } + } + + pub fn take_fd(&mut self) -> Option<RawFd> { + loop { + let fd = self + .recv_fd + .as_mut() + .and_then(|recv_fd| recv_fd.next()) + .map(|fd| { + assert_eq!(fd.len(), 1); + fd[0] + }); + + if fd.is_some() { + return fd; + } + + if self.cmsg.is_empty() { + return None; + } + + self.recv_fd = Some(cmsg::iterator(self.cmsg.take().freeze())); + } + } + + pub fn cmsg(&mut self) -> &mut BytesMut { + self.cmsg.reserve(cmsg::space(mem::size_of::<RawFd>())); + &mut self.cmsg + } +} + +#[derive(Debug)] +struct Frame { + msgs: Bytes, + handle: 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 Framed<A, C> { + io: A, + codec: C, + // Stream + read_buf: BytesMut, + #[cfg(unix)] + incoming_fd: IncomingFd, + is_readable: bool, + eof: bool, + // Sink + frames: VecDeque<Frame>, + write_buf: BytesMut, + #[cfg(unix)] + outgoing_fd: BytesMut, +} + +impl<A, C> Framed<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 {:?}, handle {:?}", frame.msgs, frame.handle); + let mut msgs = frame.msgs.clone().into_buf(); + let handle = match frame.handle { + Some(ref handle) => handle.clone(), + None => Bytes::new(), + } + .into_buf(); + try_ready!(self.io.send_msg_buf(&mut msgs, &handle)) + } + _ => { + // No pending frames. + return Ok(().into()); + } + }; + + match self.frames.pop_front() { + Some(mut frame) => { + processed += 1; + + // Close any handle that was sent. The handle is + // encoded in cmsg format inside frame.handle. Use + // the cmsg iterator to access msg and extract + // raw handle to close. + #[cfg(unix)] + if let Some(cmsg) = frame.handle.take() { + for handle in cmsg::iterator(cmsg) { + assert_eq!(handle.len(), 1); + unsafe { + super::close_platform_handle(handle[0]); + } + } + } + + 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 handle + // since it has 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, handle: Option<Bytes>) { + if self.write_buf.is_empty() { + assert!(handle.is_none()); + trace!("set_frame: No pending messages..."); + return; + } + + let msgs = self.write_buf.take().freeze(); + trace!("set_frame: msgs={:?} handle={:?}", msgs, handle); + + self.frames.push_back(Frame { msgs, handle }); + } +} + +impl<A, C> Stream for Framed<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 { + #[allow(unused_mut)] + let mut item = self.codec.decode_eof(&mut self.read_buf)?; + #[cfg(unix)] + item.set_owned_handle(|| self.incoming_fd.take_fd()); + return Ok(Some(item).into()); + } + + trace!("attempting to decode a frame"); + + #[allow(unused_mut)] + if let Some(mut item) = self.codec.decode(&mut self.read_buf)? { + trace!("frame decoded from buffer"); + #[cfg(unix)] + item.set_owned_handle(|| self.incoming_fd.take_fd()); + return Ok(Some(item).into()); + } + + self.is_readable = false; + } + + assert!(!self.eof); + + // XXX(kinetik): work around tokio_named_pipes assuming at least 1kB available. + #[cfg(windows)] + 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 + #[cfg(unix)] + let incoming_handle = self.incoming_fd.cmsg(); + #[cfg(windows)] + let incoming_handle = &mut BytesMut::new(); + let (n, _) = try_ready!(self.io.recv_msg_buf(&mut self.read_buf, incoming_handle)); + + if n == 0 { + self.eof = true; + } + + self.is_readable = true; + } + } +} + +impl<A, C> Sink for Framed<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, + 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)); + } + } + + // Take handle ownership here for `set_frame` to keep handle alive until `do_write`, + // otherwise handle would be closed too early (i.e. when `item` is dropped). + let handle = item.take_handle_for_send(); + + // On Windows, the handle is transferred by duplicating it into the target remote process during message send. + #[cfg(windows)] + if let Some((handle, target_pid)) = handle { + let remote_handle = unsafe { duplicate_platform_handle(handle, Some(target_pid))? }; + trace!( + "item handle: {:?} remote_handle: {:?}", + handle, + remote_handle + ); + // The new handle in the remote process is indicated by updating the handle stored in the item with the expected + // value on the remote. + item.set_remote_handle_value(|| Some(remote_handle)); + } + // On Unix, the handle is encoded into a cmsg buffer for out-of-band transport via sendmsg. + #[cfg(unix)] + if let Some((handle, _)) = handle { + item.set_remote_handle_value(|| Some(handle)); + } + + self.codec.encode(item, &mut self.write_buf)?; + + if handle.is_some() { + // Handle ownership is transferred into the cmsg buffer here; the local handle is closed + // after sending in `do_write`. + #[cfg(unix)] + let handle = handle.and_then(|handle| { + cmsg::builder(&mut self.outgoing_fd) + .rights(&[handle.0]) + .finish() + .ok() + }); + + // No cmsg buffer on Windows, but we still want to trigger `set_frame`, so just pass `None`. + #[cfg(windows)] + let handle = None; + + // Enforce splitting sends on messages that contain handles. + self.set_frame(handle); + } + + 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), + #[cfg(unix)] + incoming_fd: IncomingFd::new(), + is_readable: false, + eof: false, + frames: VecDeque::new(), + write_buf: BytesMut::with_capacity(INITIAL_CAPACITY), + #[cfg(unix)] + outgoing_fd: BytesMut::with_capacity(cmsg::space(mem::size_of::<RawFd>())), + } +} + +#[cfg(all(test, unix))] +mod tests { + use bytes::BufMut; + + 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::IncomingFd::new(); + + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fd().is_some()); + assert!(incoming.take_fd().is_none()); + } + + #[test] + fn multiple_cmsg_1() { + let mut incoming = super::IncomingFd::new(); + + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fd().is_some()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fd().is_some()); + assert!(incoming.take_fd().is_none()); + } + + #[test] + fn multiple_cmsg_2() { + let mut incoming = super::IncomingFd::new(); + + incoming.cmsg().put_slice(cmsg_bytes()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fd().is_some()); + incoming.cmsg().put_slice(cmsg_bytes()); + assert!(incoming.take_fd().is_some()); + assert!(incoming.take_fd().is_some()); + assert!(incoming.take_fd().is_none()); + } +} diff --git a/third_party/rust/audioipc/src/lib.rs b/third_party/rust/audioipc/src/lib.rs new file mode 100644 index 0000000000..02f7b2457b --- /dev/null +++ b/third_party/rust/audioipc/src/lib.rs @@ -0,0 +1,207 @@ +// 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; +pub mod framing; +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}; + +#[cfg(unix)] +use std::os::unix::io::IntoRawFd; +#[cfg(windows)] +use std::os::windows::io::IntoRawHandle; + +// This must match the definition of +// ipc::FileDescriptor::PlatformHandleType in Gecko. +#[cfg(windows)] +pub type PlatformHandleType = std::os::windows::raw::HANDLE; +#[cfg(unix)] +pub type PlatformHandleType = libc::c_int; + +// This stands in for RawFd/RawHandle. +#[derive(Debug)] +pub struct PlatformHandle(PlatformHandleType); + +#[cfg(unix)] +pub const INVALID_HANDLE_VALUE: PlatformHandleType = -1isize as PlatformHandleType; + +#[cfg(windows)] +pub const INVALID_HANDLE_VALUE: PlatformHandleType = winapi::um::handleapi::INVALID_HANDLE_VALUE; + +#[cfg(unix)] +fn valid_handle(handle: PlatformHandleType) -> bool { + handle >= 0 +} + +#[cfg(windows)] +fn valid_handle(handle: PlatformHandleType) -> bool { + handle != INVALID_HANDLE_VALUE && !handle.is_null() +} + +impl PlatformHandle { + pub fn new(raw: PlatformHandleType) -> PlatformHandle { + assert!(valid_handle(raw)); + PlatformHandle(raw) + } + + #[cfg(windows)] + pub fn from<T: IntoRawHandle>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_handle()) + } + + #[cfg(unix)] + pub fn from<T: IntoRawFd>(from: T) -> PlatformHandle { + PlatformHandle::new(from.into_raw_fd()) + } + + #[allow(clippy::missing_safety_doc)] + pub unsafe fn into_raw(self) -> PlatformHandleType { + let handle = self.0; + std::mem::forget(self); + handle + } + + #[cfg(unix)] + pub fn duplicate(h: PlatformHandleType) -> Result<PlatformHandle, std::io::Error> { + unsafe { + let newfd = libc::dup(h); + if !valid_handle(newfd) { + return Err(std::io::Error::last_os_error()); + } + Ok(PlatformHandle::from(newfd)) + } + } + + #[allow(clippy::missing_safety_doc)] + #[cfg(windows)] + pub unsafe fn duplicate(h: PlatformHandleType) -> Result<PlatformHandle, std::io::Error> { + let dup = duplicate_platform_handle(h, None)?; + Ok(PlatformHandle::new(dup)) + } +} + +impl Drop for PlatformHandle { + fn drop(&mut self) { + unsafe { close_platform_handle(self.0) } + } +} + +#[cfg(unix)] +unsafe fn close_platform_handle(handle: PlatformHandleType) { + libc::close(handle); +} + +#[cfg(windows)] +unsafe fn close_platform_handle(handle: PlatformHandleType) { + winapi::um::handleapi::CloseHandle(handle); +} + +#[cfg(windows)] +use winapi::shared::minwindef::DWORD; + +// Duplicate `source_handle`. +// - If `target_pid` is `Some(...)`, `source_handle` is closed. +// - If `target_pid` is `None`, `source_handle` is not closed. +#[cfg(windows)] +pub(crate) unsafe fn duplicate_platform_handle( + source_handle: PlatformHandleType, + target_pid: Option<DWORD>, +) -> Result<PlatformHandleType, std::io::Error> { + use winapi::shared::minwindef::FALSE; + use winapi::um::{handleapi, processthreadsapi, winnt}; + + let source = processthreadsapi::GetCurrentProcess(); + let (target, close_source) = if let Some(pid) = target_pid { + let target = processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, pid); + if !valid_handle(target) { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "invalid target process", + )); + } + (target, true) + } else { + (source, false) + }; + + let mut target_handle = std::ptr::null_mut(); + let mut options = winnt::DUPLICATE_SAME_ACCESS; + if close_source { + options |= winnt::DUPLICATE_CLOSE_SOURCE; + } + let ok = handleapi::DuplicateHandle( + source, + source_handle, + target, + &mut target_handle, + 0, + FALSE, + options, + ); + handleapi::CloseHandle(target); + if ok == FALSE { + return Err(std::io::Error::new( + std::io::ErrorKind::Other, + "DuplicateHandle failed", + )); + } + Ok(target_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/third_party/rust/audioipc/src/messages.rs b/third_party/rust/audioipc/src/messages.rs new file mode 100644 index 0000000000..55a46df631 --- /dev/null +++ b/third_party/rust/audioipc/src/messages.rs @@ -0,0 +1,555 @@ +// 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_handle: SerializableHandle, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RegisterDeviceCollectionChanged { + pub platform_handle: SerializableHandle, +} + +// Client -> Server messages. +// TODO: Callbacks should be different messages types so +// ServerConn::process_msg doesn't have a catch-all case. +#[derive(Debug, Serialize, Deserialize)] +pub enum ServerMessage { + ClientConnect(u32), + ClientDisconnect, + + ContextGetBackendId, + ContextGetMaxChannelCount, + ContextGetMinLatency(StreamParams), + ContextGetPreferredSampleRate, + ContextGetDeviceEnumeration(ffi::cubeb_device_type), + ContextSetupDeviceCollectionCallback, + ContextRegisterDeviceCollectionChanged(ffi::cubeb_device_type, bool), + + StreamCreate(StreamCreateParams), + StreamInit(usize, StreamInitParams), + StreamDestroy(usize), + + StreamStart(usize), + StreamStop(usize), + StreamGetPosition(usize), + StreamGetLatency(usize), + StreamGetInputLatency(usize), + StreamSetVolume(usize, f32), + StreamSetName(usize, CString), + StreamGetCurrentDevice(usize), + StreamRegisterDeviceChangeCallback(usize, bool), + + #[cfg(target_os = "linux")] + PromoteThreadToRealTime([u8; std::mem::size_of::<RtPriorityThreadInfo>()]), +} + +// Server -> Client messages. +// TODO: Streams need id. +#[derive(Debug, Serialize, Deserialize)] +pub enum ClientMessage { + ClientConnected, + ClientDisconnected, + + ContextBackendId(String), + ContextMaxChannelCount(u32), + ContextMinLatency(u32), + ContextPreferredSampleRate(u32), + ContextEnumeratedDevices(Vec<DeviceInfo>), + ContextSetupDeviceCollectionCallback(RegisterDeviceCollectionChanged), + ContextRegisteredDeviceCollectionChanged, + + StreamCreated(StreamCreate), + StreamInitialized, + StreamDestroyed, + + StreamStarted, + StreamStopped, + StreamPosition(u64), + StreamLatency(u32), + StreamInputLatency(u32), + StreamVolumeSet, + StreamNameSet, + StreamCurrentDevice(Device), + StreamRegisterDeviceChangeCallback, + + #[cfg(target_os = "linux")] + ThreadPromoted, + + Error(c_int), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum CallbackReq { + Data { + nframes: isize, + input_frame_size: usize, + output_frame_size: usize, + }, + State(ffi::cubeb_state), + DeviceChange, + SharedMem(SerializableHandle, usize), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum CallbackResp { + Data(isize), + State, + DeviceChange, + SharedMem, + Error(c_int), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionReq { + DeviceChange(ffi::cubeb_device_type), +} + +#[derive(Debug, Deserialize, Serialize)] +pub enum DeviceCollectionResp { + DeviceChange, +} + +// Represents a handle in various transitional states during serialization and remoting. +// The process of serializing and remoting handles and the ownership during various states differs +// between Windows and Unix. SerializableHandle changes during IPC is as follows: +// +// 1. The initial state `Owned`, with a valid `target_pid`. +// 2. Ownership is transferred out for processing during IPC send, becoming `Empty` temporarily. +// See `AssocRawPlatformHandle::take_handle_for_send`. +// 3. Message containing `SerializableHandleValue` is serialized and sent via IPC. +// - Windows: DuplicateHandle transfers the handle to the remote process. +// This produces a new value in the local process representing the remote handle. +// This value must be sent to the remote, so is recorded as `SerializableValue`. +// - Unix: Handle value (and ownership) is encoded into cmsg buffer via `cmsg::builder`. +// The handle is converted to a `SerializableValue` for convenience, but is otherwise unused. +// 4. Message received and deserialized in target process. +// - Windows: Deserialization converts the `SerializableValue` into `Owned`, ready for use. +// - Unix: Handle (with a new value in the target process) is received out-of-band via `recvmsg` +// and converted to `Owned` via `AssocRawPlatformHandle::set_owned_handle`. +#[derive(Debug)] +pub enum SerializableHandle { + // Owned handle, with optional target_pid on sending side. + Owned(PlatformHandle, Option<u32>), + // Transitional IPC states. + SerializableValue(PlatformHandleType), + Empty, +} + +// PlatformHandle is non-Send and containers a pointer (HANDLE) on Windows. +#[allow(clippy::non_send_fields_in_send_ty)] +unsafe impl Send for SerializableHandle {} + +impl SerializableHandle { + pub fn new(handle: PlatformHandle, target_pid: u32) -> SerializableHandle { + SerializableHandle::Owned(handle, Some(target_pid)) + } + + pub fn take_handle(&mut self) -> PlatformHandle { + match std::mem::replace(self, SerializableHandle::Empty) { + SerializableHandle::Owned(handle, target_pid) => { + assert!(target_pid.is_none()); + handle + } + _ => panic!("take_handle called in invalid state"), + } + } + + unsafe fn take_handle_for_send(&mut self) -> (PlatformHandleType, u32) { + match std::mem::replace(self, SerializableHandle::Empty) { + SerializableHandle::Owned(handle, target_pid) => ( + handle.into_raw(), + target_pid.expect("need valid target_pid"), + ), + _ => panic!("take_handle_with_target called in invalid state"), + } + } + + fn new_owned(handle: PlatformHandleType) -> SerializableHandle { + SerializableHandle::Owned(PlatformHandle::new(handle), None) + } + + fn new_serializable_value(handle: PlatformHandleType) -> SerializableHandle { + SerializableHandle::SerializableValue(handle) + } + + fn get_serializable_value(&self) -> PlatformHandleType { + match *self { + SerializableHandle::SerializableValue(handle) => handle, + _ => panic!("get_remote_handle called in invalid state"), + } + } +} + +// Raw handle values are serialized as i64. Additional handling external to (de)serialization is required during IPC +// send/receive to convert these raw values into valid handles. +impl serde::Serialize for SerializableHandle { + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> + where + S: serde::Serializer, + { + let handle = self.get_serializable_value(); + serializer.serialize_i64(handle as i64) + } +} + +impl<'de> serde::Deserialize<'de> for SerializableHandle { + fn deserialize<D>(deserializer: D) -> Result<SerializableHandle, D::Error> + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_i64(SerializableHandleVisitor) + } +} + +struct SerializableHandleVisitor; +impl<'de> serde::de::Visitor<'de> for SerializableHandleVisitor { + type Value = SerializableHandle; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("an integer between -2^63 and 2^63") + } + + fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> + where + E: serde::de::Error, + { + let value = value as PlatformHandleType; + Ok(if cfg!(windows) { + SerializableHandle::new_owned(value) + } else { + // On Unix, SerializableHandle becomes owned once `set_owned_handle` is called + // with the new local handle value during `recvmsg`. + SerializableHandle::new_serializable_value(value) + }) + } +} + +pub trait AssocRawPlatformHandle { + // Transfer ownership of handle (if any) to caller. + // The caller may then replace the handle using the platform-specific methods below. + fn take_handle_for_send(&mut self) -> Option<(PlatformHandleType, u32)> { + None + } + + // Update the item's handle to reflect the new remote handle value. + // Called on the sending side prior to serialization. + fn set_remote_handle_value<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + assert!(f().is_none()); + } + + // Update the item's handle with the received value, making it a valid owned handle. + // Called on the receiving side after deserialization. + #[cfg(unix)] + fn set_owned_handle<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + assert!(f().is_none()); + } +} + +impl AssocRawPlatformHandle for ServerMessage {} + +impl AssocRawPlatformHandle for ClientMessage { + fn take_handle_for_send(&mut self) -> Option<(PlatformHandleType, u32)> { + unsafe { + match *self { + ClientMessage::StreamCreated(ref mut data) => { + Some(data.platform_handle.take_handle_for_send()) + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + Some(data.platform_handle.take_handle_for_send()) + } + _ => None, + } + } + } + + fn set_remote_handle_value<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + match *self { + ClientMessage::StreamCreated(ref mut data) => { + let handle = + f().expect("platform_handle must be available when processing StreamCreated"); + data.platform_handle = SerializableHandle::new_serializable_value(handle); + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + let handle = f().expect("platform_handle must be available when processing ContextSetupDeviceCollectionCallback"); + data.platform_handle = SerializableHandle::new_serializable_value(handle); + } + _ => {} + } + } + + #[cfg(unix)] + fn set_owned_handle<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + match *self { + ClientMessage::StreamCreated(ref mut data) => { + let handle = + f().expect("platform_handle must be available when processing StreamCreated"); + data.platform_handle = SerializableHandle::new_owned(handle); + } + ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => { + let handle = f().expect("platform_handle must be available when processing ContextSetupDeviceCollectionCallback"); + data.platform_handle = SerializableHandle::new_owned(handle); + } + _ => {} + } + } +} + +impl AssocRawPlatformHandle for DeviceCollectionReq {} +impl AssocRawPlatformHandle for DeviceCollectionResp {} + +impl AssocRawPlatformHandle for CallbackReq { + fn take_handle_for_send(&mut self) -> Option<(PlatformHandleType, u32)> { + unsafe { + if let CallbackReq::SharedMem(ref mut data, _) = *self { + Some(data.take_handle_for_send()) + } else { + None + } + } + } + + fn set_remote_handle_value<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + if let CallbackReq::SharedMem(ref mut data, _) = *self { + let handle = f().expect("platform_handle must be available when processing SharedMem"); + *data = SerializableHandle::new_serializable_value(handle); + } + } + + #[cfg(unix)] + fn set_owned_handle<F>(&mut self, f: F) + where + F: FnOnce() -> Option<PlatformHandleType>, + { + if let CallbackReq::SharedMem(ref mut data, _) = *self { + let handle = f().expect("platform_handle must be available when processing SharedMem"); + *data = SerializableHandle::new_owned(handle); + } + } +} + +impl AssocRawPlatformHandle for CallbackResp {} + +#[cfg(test)] +mod test { + use super::StreamParams; + use cubeb::ffi; + use std::mem; + + #[test] + fn stream_params_size_check() { + assert_eq!( + mem::size_of::<StreamParams>(), + mem::size_of::<ffi::cubeb_stream_params>() + ) + } + + #[test] + fn stream_params_from() { + let raw = ffi::cubeb_stream_params { + format: ffi::CUBEB_SAMPLE_FLOAT32BE, + rate: 96_000, + channels: 32, + layout: ffi::CUBEB_LAYOUT_3F1_LFE, + prefs: ffi::CUBEB_STREAM_PREF_LOOPBACK, + }; + let wrapped = ::cubeb::StreamParams::from(raw); + let params = StreamParams::from(wrapped.as_ref()); + assert_eq!(params.format, raw.format); + assert_eq!(params.rate, raw.rate); + assert_eq!(params.channels, raw.channels); + assert_eq!(params.layout, raw.layout); + assert_eq!(params.prefs, raw.prefs); + } +} diff --git a/third_party/rust/audioipc/src/messagestream_unix.rs b/third_party/rust/audioipc/src/messagestream_unix.rs new file mode 100644 index 0000000000..61c60056a0 --- /dev/null +++ b/third_party/rust/audioipc/src/messagestream_unix.rs @@ -0,0 +1,106 @@ +// 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))) + } + + #[allow(clippy::missing_safety_doc)] + pub unsafe fn from_raw_handle(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/third_party/rust/audioipc/src/messagestream_win.rs b/third_party/rust/audioipc/src/messagestream_win.rs new file mode 100644 index 0000000000..708bbbfb8b --- /dev/null +++ b/third_party/rust/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 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), + )) + } + + #[allow(clippy::missing_safety_doc)] + pub unsafe fn from_raw_handle(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/third_party/rust/audioipc/src/msg.rs b/third_party/rust/audioipc/src/msg.rs new file mode 100644 index 0000000000..e50734d3e4 --- /dev/null +++ b/third_party/rust/audioipc/src/msg.rs @@ -0,0 +1,115 @@ +// 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 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/third_party/rust/audioipc/src/rpc/client/mod.rs b/third_party/rust/audioipc/src/rpc/client/mod.rs new file mode 100644 index 0000000000..0fb8aed565 --- /dev/null +++ b/third_party/rust/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/third_party/rust/audioipc/src/rpc/client/proxy.rs b/third_party/rust/audioipc/src/rpc/client/proxy.rs new file mode 100644 index 0000000000..bd44110d59 --- /dev/null +++ b/third_party/rust/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/third_party/rust/audioipc/src/rpc/driver.rs b/third_party/rust/audioipc/src/rpc/driver.rs new file mode 100644 index 0000000000..0890fd138e --- /dev/null +++ b/third_party/rust/audioipc/src/rpc/driver.rs @@ -0,0 +1,171 @@ +// 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>) { + 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; + } + } + } + + /// 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/third_party/rust/audioipc/src/rpc/mod.rs b/third_party/rust/audioipc/src/rpc/mod.rs new file mode 100644 index 0000000000..8d54fc6e00 --- /dev/null +++ b/third_party/rust/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/third_party/rust/audioipc/src/rpc/server.rs b/third_party/rust/audioipc/src/rpc/server.rs new file mode 100644 index 0000000000..1cb037bd8a --- /dev/null +++ b/third_party/rust/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/third_party/rust/audioipc/src/shm.rs b/third_party/rust/audioipc/src/shm.rs new file mode 100644 index 0000000000..ddcd14afe3 --- /dev/null +++ b/third_party/rust/audioipc/src/shm.rs @@ -0,0 +1,334 @@ +// Copyright © 2017 Mozilla Foundation +// +// This program is made available under an ISC-style license. See the +// accompanying file LICENSE for details. + +#![allow(clippy::missing_safety_doc)] + +use crate::errors::*; +use crate::PlatformHandle; +use std::{convert::TryInto, ffi::c_void, slice}; + +#[cfg(unix)] +pub use unix::SharedMem; +#[cfg(windows)] +pub use windows::SharedMem; + +#[derive(Copy, Clone)] +pub struct SharedMemView { + ptr: *mut c_void, + size: usize, +} + +unsafe impl Send for SharedMemView {} + +impl SharedMemView { + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + let map = slice::from_raw_parts(self.ptr as _, self.size); + if size <= self.size { + Ok(&map[..size]) + } else { + bail!("mmap size"); + } + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + let map = slice::from_raw_parts_mut(self.ptr as _, self.size); + if size <= self.size { + Ok(&mut map[..size]) + } else { + bail!("mmap size") + } + } +} + +#[cfg(unix)] +mod unix { + use super::*; + use memmap2::{MmapMut, MmapOptions}; + use std::fs::File; + use std::os::unix::io::{AsRawFd, FromRawFd}; + + #[cfg(target_os = "android")] + fn open_shm_file(_id: &str, size: usize) -> Result<File> { + unsafe { + let fd = ashmem::ASharedMemory_create(std::ptr::null(), size); + if fd >= 0 { + // Drop PROT_EXEC + let r = ashmem::ASharedMemory_setProt(fd, libc::PROT_READ | libc::PROT_WRITE); + assert_eq!(r, 0); + return Ok(File::from_raw_fd(fd.try_into().unwrap())); + } + Err(std::io::Error::last_os_error().into()) + } + } + + #[cfg(not(target_os = "android"))] + fn open_shm_file(id: &str, size: usize) -> Result<File> { + let file = open_shm_file_impl(id)?; + allocate_file(&file, size)?; + Ok(file) + } + + #[cfg(not(target_os = "android"))] + fn open_shm_file_impl(id: &str) -> Result<File> { + use std::env::temp_dir; + use std::fs::{remove_file, OpenOptions}; + + let id_cstring = std::ffi::CString::new(id).unwrap(); + + #[cfg(target_os = "linux")] + { + unsafe { + let r = libc::syscall(libc::SYS_memfd_create, id_cstring.as_ptr(), 0); + if r >= 0 { + return Ok(File::from_raw_fd(r.try_into().unwrap())); + } + } + + let mut path = std::path::PathBuf::from("/dev/shm"); + path.push(id); + + if let Ok(file) = OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .open(&path) + { + let _ = remove_file(&path); + return Ok(file); + } + } + + unsafe { + let fd = libc::shm_open( + id_cstring.as_ptr(), + libc::O_RDWR | libc::O_CREAT | libc::O_EXCL, + 0o600, + ); + if fd >= 0 { + libc::shm_unlink(id_cstring.as_ptr()); + return Ok(File::from_raw_fd(fd)); + } + } + + let mut path = temp_dir(); + path.push(id); + + let file = OpenOptions::new() + .read(true) + .write(true) + .create_new(true) + .open(&path)?; + + let _ = remove_file(&path); + Ok(file) + } + + #[cfg(not(target_os = "android"))] + fn handle_enospc(s: &str) -> Result<()> { + let err = std::io::Error::last_os_error(); + let errno = err.raw_os_error().unwrap_or(0); + assert_ne!(errno, 0); + debug!("allocate_file: {} failed errno={}", s, errno); + if errno == libc::ENOSPC { + return Err(err.into()); + } + Ok(()) + } + + #[cfg(not(target_os = "android"))] + fn allocate_file(file: &File, size: usize) -> Result<()> { + // First, set the file size. This may create a sparse file on + // many systems, which can fail with SIGBUS when accessed via a + // mapping and the lazy backing allocation fails due to low disk + // space. To avoid this, try to force the entire file to be + // preallocated before mapping using OS-specific approaches below. + + file.set_len(size.try_into().unwrap())?; + + let fd = file.as_raw_fd(); + let size: libc::off_t = size.try_into().unwrap(); + + // Try Linux-specific fallocate. + #[cfg(target_os = "linux")] + { + if unsafe { libc::fallocate(fd, 0, 0, size) } == 0 { + return Ok(()); + } + handle_enospc("fallocate()")?; + } + + // Try macOS-specific fcntl. + #[cfg(target_os = "macos")] + { + let params = libc::fstore_t { + fst_flags: libc::F_ALLOCATEALL, + fst_posmode: libc::F_PEOFPOSMODE, + fst_offset: 0, + fst_length: size, + fst_bytesalloc: 0, + }; + if unsafe { libc::fcntl(fd, libc::F_PREALLOCATE, ¶ms) } == 0 { + return Ok(()); + } + handle_enospc("fcntl(F_PREALLOCATE)")?; + } + + // Fall back to portable version, where available. + #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "dragonfly"))] + { + if unsafe { libc::posix_fallocate(fd, 0, size) } == 0 { + return Ok(()); + } + handle_enospc("posix_fallocate()")?; + } + + Ok(()) + } + + pub struct SharedMem { + file: File, + _mmap: MmapMut, + view: SharedMemView, + } + + impl SharedMem { + pub fn new(id: &str, size: usize) -> Result<SharedMem> { + let file = open_shm_file(id, size)?; + let mut mmap = unsafe { MmapOptions::new().map_mut(&file)? }; + assert_eq!(mmap.len(), size); + let view = SharedMemView { + ptr: mmap.as_mut_ptr() as _, + size, + }; + Ok(SharedMem { + file, + _mmap: mmap, + view, + }) + } + + pub unsafe fn make_handle(&self) -> Result<PlatformHandle> { + PlatformHandle::duplicate(self.file.as_raw_fd()).map_err(|e| e.into()) + } + + pub unsafe fn from(handle: PlatformHandle, size: usize) -> Result<SharedMem> { + let file = File::from_raw_fd(handle.into_raw()); + let mut mmap = MmapOptions::new().map_mut(&file)?; + assert_eq!(mmap.len(), size); + let view = SharedMemView { + ptr: mmap.as_mut_ptr() as _, + size, + }; + Ok(SharedMem { + file, + _mmap: mmap, + view, + }) + } + + pub unsafe fn unsafe_view(&self) -> SharedMemView { + self.view + } + + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + self.view.get_slice(size) + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + self.view.get_mut_slice(size) + } + } +} + +#[cfg(windows)] +mod windows { + use super::*; + use std::ptr; + use winapi::{ + shared::{minwindef::DWORD, ntdef::HANDLE}, + um::{ + handleapi::CloseHandle, + memoryapi::{MapViewOfFile, UnmapViewOfFile, FILE_MAP_ALL_ACCESS}, + winbase::CreateFileMappingA, + winnt::PAGE_READWRITE, + }, + }; + + use crate::INVALID_HANDLE_VALUE; + + pub struct SharedMem { + handle: HANDLE, + view: SharedMemView, + } + + unsafe impl Send for SharedMem {} + + impl Drop for SharedMem { + fn drop(&mut self) { + unsafe { + let ok = UnmapViewOfFile(self.view.ptr); + assert_ne!(ok, 0); + let ok = CloseHandle(self.handle); + assert_ne!(ok, 0); + } + } + } + + impl SharedMem { + pub fn new(_id: &str, size: usize) -> Result<SharedMem> { + unsafe { + let handle = CreateFileMappingA( + INVALID_HANDLE_VALUE, + ptr::null_mut(), + PAGE_READWRITE, + (size as u64 >> 32).try_into().unwrap(), + (size as u64 & (DWORD::MAX as u64)).try_into().unwrap(), + ptr::null(), + ); + if handle.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + + let ptr = MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, size); + if ptr.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + + Ok(SharedMem { + handle, + view: SharedMemView { ptr, size }, + }) + } + } + + pub unsafe fn make_handle(&self) -> Result<PlatformHandle> { + PlatformHandle::duplicate(self.handle).map_err(|e| e.into()) + } + + pub unsafe fn from(handle: PlatformHandle, size: usize) -> Result<SharedMem> { + let handle = handle.into_raw(); + let ptr = MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, size); + if ptr.is_null() { + return Err(std::io::Error::last_os_error().into()); + } + Ok(SharedMem { + handle, + view: SharedMemView { ptr, size }, + }) + } + + pub unsafe fn unsafe_view(&self) -> SharedMemView { + self.view + } + + pub unsafe fn get_slice(&self, size: usize) -> Result<&[u8]> { + self.view.get_slice(size) + } + + pub unsafe fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> { + self.view.get_mut_slice(size) + } + } +} diff --git a/third_party/rust/audioipc/src/tokio_named_pipes.rs b/third_party/rust/audioipc/src/tokio_named_pipes.rs new file mode 100644 index 0000000000..ed4343d759 --- /dev/null +++ b/third_party/rust/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())?; + 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()?; + 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/third_party/rust/audioipc/src/tokio_uds_stream.rs b/third_party/rust/audioipc/src/tokio_uds_stream.rs new file mode 100644 index 0000000000..4453c3d8f7 --- /dev/null +++ b/third_party/rust/audioipc/src/tokio_uds_stream.rs @@ -0,0 +1,361 @@ +// 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 mio::Ready; + +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> { + tokio_io::AsyncRead::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) +} |