Bug 1403048 - Update media/audioipc to b5559d28. r=kamidphish

--HG--
extra : rebase_source : aeb3f0b9a45729fa21248983d34ce956e98a9b79

media/audioipc/audioipc/Cargo.toml

@@ -8,13 +8,14 @@ authors = [
 description = "Remote Cubeb IPC"
 
 [dependencies]
+cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
+bincode = "0.8"
+bytes = "0.4"
 error-chain = "0.10.0"
+libc = "0.2"
 log = "^0.3.6"
+memmap = "0.5.2"
+mio = "0.6.7"
+mio-uds = "0.6.4"
 serde = "1.*.*"
 serde_derive = "1.*.*"
-bincode = "0.8"
-libc = "0.2"
-mio = "0.6.7"
-cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
-byteorder = "1"
-memmap = "0.5.2"

media/audioipc/audioipc/src/async.rs

@@ -0,0 +1,153 @@
+// 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.
+
+use {RecvFd, SendFd};
+use bytes::{Buf, BufMut};
+use mio_uds;
+use std::io as std_io;
+use std::os::unix::io::RawFd;
+use std::os::unix::net;
+
+/// A convenience macro for working with `io::Result<T>` from the
+/// `std::io::Read` and `std::io::Write` traits.
+///
+/// This macro takes `io::Result<T>` as input, and returns `T` as the output. If
+/// the input type is of the `Err` variant, then `Async::NotReady` is returned if
+/// it indicates `WouldBlock` or otherwise `Err` is returned.
+#[macro_export]
+macro_rules! try_nb {
+    ($e:expr) => (match $e {
+        Ok(t) => t,
+        Err(ref e) if e.kind() == ::std::io::ErrorKind::WouldBlock => {
+            return Ok(Async::NotReady)
+        }
+        Err(e) => return Err(e.into()),
+    })
+}
+
+/////////////////////////////////////////////////////////////////////////////////////////
+// Async support - Handle EWOULDBLOCK/EAGAIN from non-blocking I/O operations.
+
+/// Return type for async methods, indicates whether the operation was
+/// ready or not.
+///
+/// * `Ok(Async::Ready(t))` means that the operation has completed successfully.
+/// * `Ok(Async::NotReady)` means that the underlying system is not ready to handle operation.
+/// * `Err(e)` means that the operation has completed with the given error `e`.
+pub type AsyncResult<T, E> = Result<Async<T>, E>;
+
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum Async<T> {
+    /// Represents that a value is immediately ready.
+    Ready(T),
+    /// Represents that a value is not ready yet, but may be so later.
+    NotReady
+}
+
+impl<T> Async<T> {
+    pub fn is_ready(&self) -> bool {
+        match *self {
+            Async::Ready(_) => true,
+            Async::NotReady => false,
+        }
+    }
+
+    pub fn is_not_ready(&self) -> bool {
+        !self.is_ready()
+    }
+}
+
+/// Return type for an async attempt to send a value.
+///
+/// * `Ok(AsyncSend::Ready)` means that the operation has completed successfully.
+/// * `Ok(AsyncSend::NotReady(t))` means that the underlying system is not ready to handle
+///    send. returns the value that tried to be sent in `t`.
+/// * `Err(e)` means that operation has completed with the given error `e`.
+pub type AsyncSendResult<T, E> = Result<AsyncSend<T>, E>;
+
+#[derive(Copy, Clone, Debug, PartialEq)]
+pub enum AsyncSend<T> {
+    Ready,
+    NotReady(T)
+}
+
+pub trait AsyncRecvFd: RecvFd {
+    unsafe fn prepare_uninitialized_buffer(&self, bytes: &mut [u8]) -> bool {
+        for byte in bytes.iter_mut() {
+            *byte = 0;
+        }
+
+        true
+    }
+
+    /// 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_buf_fd<B>(&mut self, buf: &mut B) -> AsyncResult<(usize, Option<RawFd>), std_io::Error>
+    where
+        Self: Sized,
+        B: BufMut,
+    {
+        if !buf.has_remaining_mut() {
+            return Ok(Async::Ready((0, None)));
+        }
+
+        unsafe {
+            let (n, fd) = {
+                let bytes = buf.bytes_mut();
+                self.prepare_uninitialized_buffer(bytes);
+                try_nb!(self.recv_fd(bytes))
+            };
+
+            buf.advance_mut(n);
+            Ok(Async::Ready((n, fd)))
+        }
+    }
+}
+
+impl AsyncRecvFd for net::UnixStream {}
+impl AsyncRecvFd for mio_uds::UnixStream {}
+
+/// 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 AsyncSendFd: SendFd {
+    /// 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_buf_fd<B>(&mut self, buf: &mut B, fd: Option<RawFd>) -> AsyncResult<usize, std_io::Error>
+    where
+        Self: Sized,
+        B: Buf,
+    {
+        if !buf.has_remaining() {
+            return Ok(Async::Ready(0));
+        }
+
+        let n = try_nb!(self.send_fd(buf.bytes(), fd));
+        buf.advance(n);
+        Ok(Async::Ready(n))
+    }
+}
+
+impl AsyncSendFd for net::UnixStream {}
+impl AsyncSendFd for mio_uds::UnixStream {}

media/audioipc/audioipc/src/codec.rs

@@ -0,0 +1,141 @@
+// 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, Bounded, deserialize, serialize_into, serialized_size};
+use bytes::{Buf, BufMut, BytesMut, LittleEndian};
+use serde::de::DeserializeOwned;
+use serde::ser::Serialize;
+use std::io as std_io;
+use std::io::Cursor;
+use std::mem;
+
+////////////////////////////////////////////////////////////////////////////////
+// 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.
+
+#[derive(Debug)]
+enum FrameState {
+    Head,
+    Data(usize)
+}
+
+#[derive(Debug)]
+pub struct Decoder {
+    state: FrameState
+}
+
+impl Decoder {
+    pub fn new() -> Self {
+        Decoder {
+            state: FrameState::Head
+        }
+    }
+
+    fn decode_head(&mut self, src: &mut BytesMut) -> std_io::Result<Option<usize>> {
+        let head_size = mem::size_of::<u16>();
+        if src.len() < head_size {
+            // Not enough data
+            return Ok(None);
+        }
+
+        let n = {
+            let mut src = Cursor::new(&mut *src);
+
+            // match endianess
+            let n = src.get_uint::<LittleEndian>(head_size);
+
+            if n > u64::from(u16::max_value()) {
+                return Err(std_io::Error::new(
+                    std_io::ErrorKind::InvalidData,
+                    "frame size too big"
+                ));
+            }
+
+            // The check above ensures there is no overflow
+            n as usize
+        };
+
+        // Consume the length field
+        let _ = src.split_to(head_size);
+
+        Ok(Some(n))
+    }
+
+    fn decode_data(&self, n: usize, src: &mut BytesMut) -> std_io::Result<Option<BytesMut>> {
+        // At this point, the buffer has already had the required capacity
+        // reserved. All there is to do is read.
+        if src.len() < n {
+            return Ok(None);
+        }
+
+        Ok(Some(src.split_to(n)))
+    }
+
+    pub fn split_frame(&mut self, src: &mut BytesMut) -> std_io::Result<Option<BytesMut>> {
+        let n = match self.state {
+            FrameState::Head => {
+                match try!(self.decode_head(src)) {
+                    Some(n) => {
+                        self.state = FrameState::Data(n);
+
+                        // Ensure that the buffer has enough space to read the
+                        // incoming payload
+                        src.reserve(n);
+
+                        n
+                    },
+                    None => return Ok(None),
+                }
+            },
+            FrameState::Data(n) => n,
+        };
+
+        match try!(self.decode_data(n, src)) {
+            Some(data) => {
+                // Update the decode state
+                self.state = FrameState::Head;
+
+                // Make sure the buffer has enough space to read the next head
+                src.reserve(mem::size_of::<u16>());
+
+                Ok(Some(data))
+            },
+            None => Ok(None),
+        }
+    }
+
+    pub fn decode<ITEM: DeserializeOwned>(&mut self, src: &mut BytesMut) -> Result<Option<ITEM>, bincode::Error> {
+        match try!(self.split_frame(src)) {
+            Some(buf) => deserialize::<ITEM>(buf.as_ref()).and_then(|t| Ok(Some(t))),
+            None => Ok(None),
+        }
+    }
+}
+
+impl Default for Decoder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+pub fn encode<ITEM: Serialize>(dst: &mut BytesMut, item: &ITEM) -> Result<(), bincode::Error> {
+    let head_len = mem::size_of::<u16>() as u64;
+    let item_len = serialized_size(item);
+
+    if head_len + item_len > u64::from(u16::max_value()) {
+        return Err(Box::new(bincode::ErrorKind::IoError(std_io::Error::new(
+            std_io::ErrorKind::InvalidInput,
+            "frame too big"
+        ))));
+    }
+
+    let n = (head_len + item_len) as usize;
+    dst.reserve(n);
+    dst.put_u16::<LittleEndian>(item_len as u16);
+    serialize_into(&mut dst.writer(), item, Bounded(item_len))
+}

media/audioipc/audioipc/src/connection.rs

@@ -1,26 +1,19 @@
-use bincode::{self, deserialize, serialize};
+use {AutoCloseFd, RecvFd, SendFd};
+use async::{Async, AsyncRecvFd};
+use bytes::{BufMut, BytesMut};
+use codec::{Decoder, encode};
 use errors::*;
-use msg;
 use mio::{Poll, PollOpt, Ready, Token};
 use mio::event::Evented;
 use mio::unix::EventedFd;
 use serde::de::DeserializeOwned;
 use serde::ser::Serialize;
+use std::collections::VecDeque;
 use std::fmt::Debug;
 use std::io::{self, Read};
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::os::unix::net;
 use std::os::unix::prelude::*;
-use libc;
-use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
-
-pub trait RecvFd {
-    fn recv_fd(&mut self, bytes: &mut [u8]) -> io::Result<(usize, Option<RawFd>)>;
-}
-
-pub trait SendFd {
-    fn send_fd<FD: IntoRawFd>(&mut self, bytes: &[u8], fd: Option<FD>) -> io::Result<(usize)>;
-}
 
 // Because of the trait implementation rules in Rust, this needs to be
 // a wrapper class to allow implementation of a trait from another
@@ -30,14 +23,23 @@ pub trait SendFd {
 
 #[derive(Debug)]
 pub struct Connection {
-    stream: net::UnixStream
+    stream: net::UnixStream,
+    recv_buffer: BytesMut,
+    recv_fd: VecDeque<AutoCloseFd>,
+    send_buffer: BytesMut,
+    decoder: Decoder
 }
 
 impl Connection {
     pub fn new(stream: net::UnixStream) -> Connection {
         info!("Create new connection");
+        stream.set_nonblocking(false).unwrap();
         Connection {
-            stream: stream
+            stream: stream,
+            recv_buffer: BytesMut::with_capacity(1024),
+            recv_fd: VecDeque::new(),
+            send_buffer: BytesMut::with_capacity(1024),
+            decoder: Decoder::new()
         }
     }
 
@@ -60,51 +62,74 @@ impl Connection {
     /// ```
     pub fn pair() -> io::Result<(Connection, Connection)> {
         let (s1, s2) = net::UnixStream::pair()?;
-        Ok((
+        Ok((Connection::new(s1), Connection::new(s2)))
-            Connection {
-                stream: s1
-            },
-            Connection {
-                stream: s2
     }
-        ))
+
+    pub fn take_fd(&mut self) -> Option<RawFd> {
+        self.recv_fd.pop_front().map(|fd| fd.into_raw_fd())
     }
 
     pub fn receive<RT>(&mut self) -> Result<RT>
     where
         RT: DeserializeOwned + Debug,
     {
-        match self.receive_with_fd() {
+        self.receive_with_fd()
-            Ok((r, None)) => Ok(r),
-            Ok((_, Some(_))) => panic!("unexpected fd received"),
-            Err(e) => Err(e),
-        }
     }
 
-    pub fn receive_with_fd<RT>(&mut self) -> Result<(RT, Option<RawFd>)>
+    pub fn receive_with_fd<RT>(&mut self) -> Result<RT>
     where
         RT: DeserializeOwned + Debug,
     {
-        // TODO: Check deserialize_from and serialize_into.
+        trace!("received_with_fd...");
-        let mut encoded = vec![0; 32 * 1024]; // TODO: Get max size from bincode, or at least assert.
+        loop {
+            trace!("   recv_buffer = {:?}", self.recv_buffer);
+            if !self.recv_buffer.is_empty() {
+                let r = self.decoder.decode(&mut self.recv_buffer);
+                trace!("receive {:?}", r);
+                match r {
+                    Ok(Some(r)) => return Ok(r),
+                    Ok(None) => {
+                        /* Buffer doesn't contain enough data for a complete
+                         * message, so need to enter recv_buf_fd to get more. */
+                    },
+                    Err(e) => return Err(e).chain_err(|| "Failed to deserialize message"),
+                }
+            }
+
+            // 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
+
+            // The decoder.decode should have reserved an amount for
+            // the next bit it needs to read.  Check that we reserved
+            // enough space for, at least the 2 byte size prefix.
+            assert!(self.recv_buffer.remaining_mut() > 2);
+
             // TODO: Read until block, EOF, or error.
             // TODO: Switch back to recv_fd.
-        match self.stream.recv_fd(&mut encoded) {
+            match self.stream.recv_buf_fd(&mut self.recv_buffer) {
-            Ok((0, _)) => Err(ErrorKind::Disconnected.into()),
+                Ok(Async::Ready((0, _))) => return Err(ErrorKind::Disconnected.into()),
                 // TODO: Handle partial read?
-            Ok((n, fd)) => {
+                Ok(Async::Ready((_, fd))) => {
-                let r = deserialize(&encoded[..n]);
+                    trace!(
-                debug!("receive {:?}", r);
+                        "   recv_buf_fd: recv_buffer: {:?}, recv_fd: {:?}, fd: {:?}",
-                match r {
+                        self.recv_buffer,
-                    Ok(r) => Ok((r, fd)),
+                        self.recv_fd,
-                    Err(e) => Err(e).chain_err(|| "Failed to deserialize message"),
+                        fd
+                    );
+                    if let Some(fd) = fd {
+                        self.recv_fd.push_back(
+                            unsafe { AutoCloseFd::from_raw_fd(fd) }
+                        );
                     }
                 },
+                Ok(Async::NotReady) => bail!("Socket should be blocking."),
                 // TODO: Handle dropped message.
                 // Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => panic!("wouldblock"),
                 _ => bail!("socket write"),
             }
         }
+    }
 
     pub fn send<ST>(&mut self, msg: ST) -> Result<usize>
     where
@@ -118,9 +143,11 @@ impl Connection {
         ST: Serialize + Debug,
         FD: IntoRawFd + Debug,
     {
-        let encoded: Vec<u8> = serialize(&msg, bincode::Infinite)?;
+        trace!("send_with_fd {:?}, {:?}", msg, fd_to_send);
-        info!("send_with_fd {:?}, {:?}", msg, fd_to_send);
+        try!(encode(&mut self.send_buffer, &msg));
-        self.stream.send_fd(&encoded, fd_to_send).chain_err(
+        let fd_to_send = fd_to_send.map(|fd| fd.into_raw_fd());
+        let send = self.send_buffer.take().freeze();
+        self.stream.send_fd(send.as_ref(), fd_to_send).chain_err(
             || "Failed to send message with fd"
         )
     }
@@ -153,14 +180,6 @@ impl<'a> Read for &'a Connection {
     }
 }
 
-impl RecvFd for net::UnixStream {
-    fn recv_fd(&mut self, buf_to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd>)> {
-        let length = self.read_u32::<LittleEndian>()?;
-
-        msg::recvmsg(self.as_raw_fd(), &mut buf_to_recv[..length as usize])
-    }
-}
-
 impl RecvFd for Connection {
     fn recv_fd(&mut self, buf_to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd>)> {
         self.stream.recv_fd(buf_to_recv)
@@ -169,9 +188,7 @@ impl RecvFd for Connection {
 
 impl FromRawFd for Connection {
     unsafe fn from_raw_fd(fd: RawFd) -> Connection {
-        Connection {
+        Connection::new(net::UnixStream::from_raw_fd(fd))
-            stream: net::UnixStream::from_raw_fd(fd)
-        }
     }
 }
 
@@ -181,19 +198,8 @@ impl IntoRawFd for Connection {
     }
 }
 
-impl SendFd for net::UnixStream {
-    fn send_fd<FD: IntoRawFd>(&mut self, buf_to_send: &[u8], fd_to_send: Option<FD>) -> io::Result<usize> {
-        self.write_u32::<LittleEndian>(buf_to_send.len() as u32)?;
-
-        let fd_to_send = fd_to_send.map(|fd| fd.into_raw_fd());
-        let r = msg::sendmsg(self.as_raw_fd(), buf_to_send, fd_to_send);
-        fd_to_send.map(|fd| unsafe { libc::close(fd) });
-        r
-    }
-}
-
 impl SendFd for Connection {
-    fn send_fd<FD: IntoRawFd>(&mut self, buf_to_send: &[u8], fd_to_send: Option<FD>) -> io::Result<usize> {
+    fn send_fd(&mut self, buf_to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Result<usize> {
         self.stream.send_fd(buf_to_send, fd_to_send)
     }
 }

media/audioipc/audioipc/src/lib.rs

@@ -13,18 +13,18 @@ extern crate log;
 
 #[macro_use]
 extern crate serde_derive;
-extern crate serde;
+
 extern crate bincode;
-
+extern crate bytes;
-extern crate mio;
-
 extern crate cubeb_core;
-
 extern crate libc;
-extern crate byteorder;
-
 extern crate memmap;
+extern crate mio;
+extern crate mio_uds;
+extern crate serde;
 
+pub mod async;
+pub mod codec;
 mod connection;
 pub mod errors;
 pub mod messages;
@@ -33,9 +33,89 @@ pub mod shm;
 
 pub use connection::*;
 pub use messages::{ClientMessage, ServerMessage};
+
 use std::env::temp_dir;
+use std::io;
+use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+use std::os::unix::net;
 use std::path::PathBuf;
 
+// 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 RecvFd {
+    fn recv_fd(&mut self, bytes: &mut [u8]) -> io::Result<(usize, Option<RawFd>)>;
+}
+
+pub trait SendFd {
+    fn send_fd(&mut self, bytes: &[u8], fd: Option<RawFd>) -> io::Result<(usize)>;
+}
+
+impl RecvFd for net::UnixStream {
+    fn recv_fd(&mut self, buf_to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd>)> {
+        msg::recvmsg(self.as_raw_fd(), buf_to_recv)
+    }
+}
+
+impl RecvFd for mio_uds::UnixStream {
+    fn recv_fd(&mut self, buf_to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd>)> {
+        msg::recvmsg(self.as_raw_fd(), buf_to_recv)
+    }
+}
+
+impl SendFd for net::UnixStream {
+    fn send_fd(&mut self, buf_to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Result<usize> {
+        msg::sendmsg(self.as_raw_fd(), buf_to_send, fd_to_send)
+    }
+}
+
+impl SendFd for mio_uds::UnixStream {
+    fn send_fd(&mut self, buf_to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Result<usize> {
+        msg::sendmsg(self.as_raw_fd(), buf_to_send, fd_to_send)
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+#[derive(Debug)]
+pub struct AutoCloseFd(RawFd);
+
+impl Drop for AutoCloseFd {
+    fn drop(&mut self) {
+        unsafe {
+            libc::close(self.0);
+        }
+    }
+}
+
+impl FromRawFd for AutoCloseFd {
+    unsafe fn from_raw_fd(fd: RawFd) -> Self {
+        AutoCloseFd(fd)
+    }
+}
+
+impl IntoRawFd for AutoCloseFd {
+    fn into_raw_fd(self) -> RawFd {
+        let fd = self.0;
+        ::std::mem::forget(self);
+        fd
+    }
+}
+
+impl AsRawFd for AutoCloseFd {
+    fn as_raw_fd(&self) -> RawFd {
+        self.0
+    }
+}
+
+impl<'a> AsRawFd for &'a AutoCloseFd {
+    fn as_raw_fd(&self) -> RawFd {
+        self.0
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 fn get_temp_path(name: &str) -> PathBuf {
     let mut path = temp_dir();
     path.push(name);

media/audioipc/audioipc/src/messages.rs

@@ -126,7 +126,7 @@ pub struct StreamParams {
 
 impl<'a> From<&'a ffi::cubeb_stream_params> for StreamParams {
     fn from(params: &'a ffi::cubeb_stream_params) -> Self {
-        assert!(params.channels <= u8::max_value() as u32);
+        assert!(params.channels <= u32::from(u8::max_value()));
 
         StreamParams {
             format: params.format,
@@ -141,8 +141,8 @@ impl<'a> From<&'a StreamParams> for ffi::cubeb_stream_params {
     fn from(params: &StreamParams) -> Self {
         ffi::cubeb_stream_params {
             format: params.format,
-            rate: params.rate as u32,
+            rate: u32::from(params.rate),
-            channels: params.channels as u32,
+            channels: u32::from(params.channels),
             layout: params.layout
         }
     }

media/audioipc/audioipc/src/msg.rs

@@ -1,27 +1,36 @@
 use libc;
 use std::io;
 use std::mem;
-use std::ptr;
 use std::os::unix::io::RawFd;
-use std;
+use std::ptr;
 
 // Note: The following fields must be laid out together, the OS expects them
 // to be part of a single allocation.
 #[repr(C)]
 struct CmsgSpace {
     cmsghdr: libc::cmsghdr,
+    #[cfg(not(target_os = "macos"))]
+    __padding: [usize; 0],
     data: libc::c_int,
 }
 
-unsafe fn sendmsg_retry(fd: libc::c_int, msg: *const libc::msghdr, flags: libc::c_int) -> libc::ssize_t {
+#[cfg(not(target_os = "macos"))]
-    loop {
+fn cmsg_align(len: usize) -> usize {
-        let r = libc::sendmsg(fd, msg, flags);
+    let align_bytes = mem::size_of::<usize>() - 1;
-        if r == -1 && io::Error::last_os_error().raw_os_error().unwrap() == libc::EAGAIN {
+    (len + align_bytes) & !align_bytes
-            std::thread::yield_now();
-            continue;
 }
-        return r;
+
+#[cfg(target_os = "macos")]
+fn cmsg_align(len: usize) -> usize {
+    len
 }
+
+fn cmsg_space() -> usize {
+    mem::size_of::<CmsgSpace>()
+}
+
+fn cmsg_len() -> usize {
+    cmsg_align(mem::size_of::<libc::cmsghdr>()) + mem::size_of::<libc::c_int>()
 }
 
 pub fn sendmsg(fd: RawFd, to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Result<usize> {
@@ -33,7 +42,7 @@ pub fn sendmsg(fd: RawFd, to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Resu
     msghdr.msg_iovlen = 1;
     if fd_to_send.is_some() {
         msghdr.msg_control = &mut cmsg.cmsghdr as *mut _ as *mut _;
-        msghdr.msg_controllen = mem::size_of::<CmsgSpace>() as _;
+        msghdr.msg_controllen = cmsg_space() as _;
     }
 
     iovec.iov_base = if to_send.is_empty() {
@@ -44,13 +53,13 @@ pub fn sendmsg(fd: RawFd, to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Resu
     };
     iovec.iov_len = to_send.len();
 
-    cmsg.cmsghdr.cmsg_len = msghdr.msg_controllen;
+    cmsg.cmsghdr.cmsg_len = cmsg_len() as _;
     cmsg.cmsghdr.cmsg_level = libc::SOL_SOCKET;
     cmsg.cmsghdr.cmsg_type = libc::SCM_RIGHTS;
 
     cmsg.data = fd_to_send.unwrap_or(-1);
 
-    let result = unsafe { sendmsg_retry(fd, &msghdr, 0) };
+    let result = unsafe { libc::sendmsg(fd, &msghdr, 0) };
     if result >= 0 {
         Ok(result as usize)
     } else {
@@ -58,17 +67,6 @@ pub fn sendmsg(fd: RawFd, to_send: &[u8], fd_to_send: Option<RawFd>) -> io::Resu
     }
 }
 
-unsafe fn recvmsg_retry(fd: libc::c_int, msg: *mut libc::msghdr, flags: libc::c_int) -> libc::ssize_t {
-    loop {
-        let r = libc::recvmsg(fd, msg, flags);
-        if r == -1 && io::Error::last_os_error().raw_os_error().unwrap() == libc::EAGAIN {
-            std::thread::yield_now();
-            continue;
-        }
-        return r;
-    }
-}
-
 pub fn recvmsg(fd: RawFd, to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd>)> {
     let mut msghdr: libc::msghdr = unsafe { mem::zeroed() };
     let mut iovec: libc::iovec = unsafe { mem::zeroed() };
@@ -77,7 +75,7 @@ pub fn recvmsg(fd: RawFd, to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd
     msghdr.msg_iov = &mut iovec as *mut _;
     msghdr.msg_iovlen = 1;
     msghdr.msg_control = &mut cmsg.cmsghdr as *mut _ as *mut _;
-    msghdr.msg_controllen = mem::size_of::<CmsgSpace>() as _;
+    msghdr.msg_controllen = cmsg_space() as _;
 
     iovec.iov_base = if to_recv.is_empty() {
         // Empty Vecs have a non-null pointer.
@@ -87,10 +85,10 @@ pub fn recvmsg(fd: RawFd, to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd
     };
     iovec.iov_len = to_recv.len();
 
-    let result = unsafe { recvmsg_retry(fd, &mut msghdr, 0) };
+    let result = unsafe { libc::recvmsg(fd, &mut msghdr, 0) };
     if result >= 0 {
-        let fd = if msghdr.msg_controllen == mem::size_of::<CmsgSpace>() as _ &&
+        let fd = if msghdr.msg_controllen == cmsg_space() as _ &&
-            cmsg.cmsghdr.cmsg_len == mem::size_of::<CmsgSpace>() as _ &&
+            cmsg.cmsghdr.cmsg_len == cmsg_len() as _ &&
             cmsg.cmsghdr.cmsg_level == libc::SOL_SOCKET &&
             cmsg.cmsghdr.cmsg_type == libc::SCM_RIGHTS {
                 Some(cmsg.data)
@@ -103,3 +101,61 @@ pub fn recvmsg(fd: RawFd, to_recv: &mut [u8]) -> io::Result<(usize, Option<RawFd
         Err(io::Error::last_os_error())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use libc;
+    use std::mem;
+    use std::os::unix::net::UnixStream;
+    use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd};
+    use std::io::{Read, Write};
+    use super::{cmsg_len, cmsg_space, sendmsg, recvmsg};
+
+    #[test]
+    fn portable_sizes() {
+        if cfg!(all(target_os = "linux", target_pointer_width = "64")) {
+            assert_eq!(mem::size_of::<libc::cmsghdr>(), 16);
+            assert_eq!(cmsg_len(), 20);
+            assert_eq!(cmsg_space(), 24);
+        } else if cfg!(all(target_os = "linux", target_pointer_width = "32")) {
+            assert_eq!(mem::size_of::<libc::cmsghdr>(), 12);
+            assert_eq!(cmsg_len(), 16);
+            assert_eq!(cmsg_space(), 16);
+        } else if cfg!(target_os = "macos") {
+            assert_eq!(mem::size_of::<libc::cmsghdr>(), 12);
+            assert_eq!(cmsg_len(), 16);
+            assert_eq!(cmsg_space(), 16);
+        } else if cfg!(target_pointer_width = "64") {
+            assert_eq!(mem::size_of::<libc::cmsghdr>(), 12);
+            assert_eq!(cmsg_len(), 20);
+            assert_eq!(cmsg_space(), 24);
+        } else {
+            assert_eq!(mem::size_of::<libc::cmsghdr>(), 12);
+            assert_eq!(cmsg_len(), 16);
+            assert_eq!(cmsg_space(), 16);
+        }
+    }
+
+    #[test]
+    fn fd_passing() {
+        let (tx, rx) = UnixStream::pair().unwrap();
+
+        let (send_tx, mut send_rx) = UnixStream::pair().unwrap();
+
+        let fd = send_tx.into_raw_fd();
+        assert_eq!(sendmsg(tx.as_raw_fd(), b"a", Some(fd)).unwrap(), 1);
+        unsafe { libc::close(fd) };
+
+        let mut buf = [0u8];
+        let (got, fd) = recvmsg(rx.as_raw_fd(), &mut buf).unwrap();
+        assert_eq!(got, 1);
+        assert_eq!(&buf, b"a");
+
+        let mut send_tx = unsafe { UnixStream::from_raw_fd(fd.unwrap()) };
+        assert_eq!(send_tx.write(b"b").unwrap(), 1);
+
+        let mut buf = [0u8];
+        assert_eq!(send_rx.read(&mut buf).unwrap(), 1);
+        assert_eq!(&buf, b"b");
+    }
+}

media/audioipc/client/Cargo.toml

@@ -6,6 +6,6 @@ description = "Cubeb Backend for talking to remote cubeb server."
 
 [dependencies]
 audioipc = { path="../audioipc" }
-cubeb-core = { path="../../cubeb-rs/cubeb-core" }
 cubeb-backend = { path="../../cubeb-rs/cubeb-backend" }
+cubeb-core = { path="../../cubeb-rs/cubeb-core" }
 log = "^0.3.6"

media/audioipc/client/src/context.rs

@@ -4,9 +4,10 @@
 // accompanying file LICENSE for details
 
 use ClientStream;
+use assert_not_in_callback;
 use audioipc::{self, ClientMessage, Connection, ServerMessage, messages};
 use cubeb_backend::{Context, Ops};
-use cubeb_core::{DeviceId, DeviceType, Error, Result, StreamParams, ffi};
+use cubeb_core::{DeviceId, DeviceType, Error, ErrorCode, Result, StreamParams, ffi};
 use cubeb_core::binding::Binding;
 use std::ffi::{CStr, CString};
 use std::mem;
@@ -33,13 +34,14 @@ pub const CLIENT_OPS: Ops = capi_new!(ClientContext, ClientStream);
 
 impl ClientContext {
     #[doc(hidden)]
-    pub fn conn(&self) -> MutexGuard<Connection> {
+    pub fn connection(&self) -> MutexGuard<Connection> {
         self.connection.lock().unwrap()
     }
 }
 
 impl Context for ClientContext {
     fn init(_context_name: Option<&CStr>) -> Result<*mut ffi::cubeb> {
+        assert_not_in_callback();
         // TODO: encapsulate connect, etc inside audioipc.
         let stream = t!(UnixStream::connect(audioipc::get_uds_path()));
         let ctx = Box::new(ClientContext {
@@ -50,29 +52,44 @@ impl Context for ClientContext {
     }
 
     fn backend_id(&self) -> &'static CStr {
+        // HACK: This is called reentrantly from Gecko's AudioStream::DataCallback.
+        //assert_not_in_callback();
         unsafe { CStr::from_ptr(b"remote\0".as_ptr() as *const _) }
     }
 
     fn max_channel_count(&self) -> Result<u32> {
-        send_recv!(self.conn(), ContextGetMaxChannelCount => ContextMaxChannelCount())
+        // HACK: This needs to be reentrant as MSG calls it from within data_callback.
+        //assert_not_in_callback();
+        //let mut conn = self.connection();
+        //send_recv!(conn, ContextGetMaxChannelCount => ContextMaxChannelCount())
+        warn!("Context::max_channel_count lying about result until reentrancy issues resolved.");
+        Ok(2)
     }
 
     fn min_latency(&self, params: &StreamParams) -> Result<u32> {
+        assert_not_in_callback();
         let params = messages::StreamParams::from(unsafe { &*params.raw() });
-        send_recv!(self.conn(), ContextGetMinLatency(params) => ContextMinLatency())
+        let mut conn = self.connection();
+        send_recv!(conn, ContextGetMinLatency(params) => ContextMinLatency())
     }
 
     fn preferred_sample_rate(&self) -> Result<u32> {
-        send_recv!(self.conn(), ContextGetPreferredSampleRate => ContextPreferredSampleRate())
+        assert_not_in_callback();
+        let mut conn = self.connection();
+        send_recv!(conn, ContextGetPreferredSampleRate => ContextPreferredSampleRate())
     }
 
     fn preferred_channel_layout(&self) -> Result<ffi::cubeb_channel_layout> {
-        send_recv!(self.conn(), ContextGetPreferredChannelLayout => ContextPreferredChannelLayout())
+        assert_not_in_callback();
+        let mut conn = self.connection();
+        send_recv!(conn, ContextGetPreferredChannelLayout => ContextPreferredChannelLayout())
     }
 
     fn enumerate_devices(&self, devtype: DeviceType) -> Result<ffi::cubeb_device_collection> {
+        assert_not_in_callback();
+        let mut conn = self.connection();
         let v: Vec<ffi::cubeb_device_info> =
-            match send_recv!(self.conn(), ContextGetDeviceEnumeration(devtype.bits()) => ContextEnumeratedDevices()) {
+            match send_recv!(conn, ContextGetDeviceEnumeration(devtype.bits()) => ContextEnumeratedDevices()) {
                 Ok(mut v) => v.drain(..).map(|i| i.into()).collect(),
                 Err(e) => return Err(e),
             };
@@ -88,6 +105,7 @@ impl Context for ClientContext {
     }
 
     fn device_collection_destroy(&self, collection: *mut ffi::cubeb_device_collection) {
+        assert_not_in_callback();
         unsafe {
             let coll = &*collection;
             let mut devices = Vec::from_raw_parts(
@@ -95,7 +113,7 @@ impl Context for ClientContext {
                 coll.count,
                 coll.count
             );
-            for dev in devices.iter_mut() {
+            for dev in &mut devices {
                 if !dev.device_id.is_null() {
                     let _ = CString::from_raw(dev.device_id as *mut _);
                 }
@@ -125,6 +143,7 @@ impl Context for ClientContext {
         state_callback: ffi::cubeb_state_callback,
         user_ptr: *mut c_void,
     ) -> Result<*mut ffi::cubeb_stream> {
+        assert_not_in_callback();
 
         fn opt_stream_params(p: Option<&ffi::cubeb_stream_params>) -> Option<messages::StreamParams> {
             match p {
@@ -149,7 +168,7 @@ impl Context for ClientContext {
             output_stream_params: output_stream_params,
             latency_frames: latency_frame
         };
-        stream::init(&self, init_params, data_callback, state_callback, user_ptr)
+        stream::init(self, init_params, data_callback, state_callback, user_ptr)
     }
 
     fn register_device_collection_changed(
@@ -158,13 +177,24 @@ impl Context for ClientContext {
         _collection_changed_callback: ffi::cubeb_device_collection_changed_callback,
         _user_ptr: *mut c_void,
     ) -> Result<()> {
+        assert_not_in_callback();
         Ok(())
     }
 }
 
 impl Drop for ClientContext {
     fn drop(&mut self) {
+        let mut conn = self.connection();
         info!("ClientContext drop...");
-        let _: Result<()> = send_recv!(self.conn(), ClientDisconnect => ClientDisconnected);
+        let r = conn.send(ServerMessage::ClientDisconnect);
+        if r.is_err() {
+            debug!("ClientContext::Drop send error={:?}", r);
+        } else {
+            let r = conn.receive();
+            if let Ok(ClientMessage::ClientDisconnected) = r {
+            } else {
+                debug!("ClientContext::Drop receive error={:?}", r);
+            }
+        }
     }
 }

media/audioipc/client/src/lib.rs

@@ -21,6 +21,21 @@ use cubeb_core::ffi;
 use std::os::raw::{c_char, c_int};
 use stream::ClientStream;
 
+thread_local!(static IN_CALLBACK: std::cell::RefCell<bool> = std::cell::RefCell::new(false));
+
+fn set_in_callback(in_callback: bool) {
+    IN_CALLBACK.with(|b| {
+        assert_eq!(*b.borrow(), !in_callback);
+        *b.borrow_mut() = in_callback;
+    });
+}
+
+fn assert_not_in_callback() {
+    IN_CALLBACK.with(|b| {
+        assert_eq!(*b.borrow(), false);
+    });
+}
+
 #[no_mangle]
 /// Entry point from C code.
 pub unsafe extern "C" fn audioipc_client_init(c: *mut *mut ffi::cubeb, context_name: *const c_char) -> c_int {

media/audioipc/client/src/send_recv.rs

@@ -17,26 +17,36 @@ macro_rules! send_recv {
         send_recv!(__recv $conn, $rmsg __result)
     }};
     //
-    (__send $conn:expr, $smsg:ident) => (
+    (__send $conn:expr, $smsg:ident) => ({
-        $conn.send(ServerMessage::$smsg)
+        let r = $conn.send(ServerMessage::$smsg);
-            .unwrap();
+        if r.is_err() {
-    );
+            debug!("send error - got={:?}", r);
-    (__send $conn:expr, $smsg:ident, $($a:expr),*) => (
+            return Err(ErrorCode::Error.into());
-        $conn.send(ServerMessage::$smsg($($a),*))
+        }
-            .unwrap();
+    });
-    );
+    (__send $conn:expr, $smsg:ident, $($a:expr),*) => ({
-    (__recv $conn:expr, $rmsg:ident) => (
+        let r = $conn.send(ServerMessage::$smsg($($a),*));
-        if let ClientMessage::$rmsg = $conn.receive().unwrap() {
+        if r.is_err() {
+            debug!("send error - got={:?}", r);
+            return Err(ErrorCode::Error.into());
+        }
+    });
+    (__recv $conn:expr, $rmsg:ident) => ({
+        let r = $conn.receive().unwrap();
+        if let ClientMessage::$rmsg = r {
             Ok(())
         } else {
-            panic!("wrong message received");
+            debug!("receive error - got={:?}", r);
+            Err(ErrorCode::Error.into())
         }
-    );
+    });
-    (__recv $conn:expr, $rmsg:ident __result) => (
+    (__recv $conn:expr, $rmsg:ident __result) => ({
-        if let ClientMessage::$rmsg(v) = $conn.receive().unwrap() {
+        let r = $conn.receive().unwrap();
+        if let ClientMessage::$rmsg(v) = r {
             Ok(v)
         } else {
-            panic!("wrong message received");
+            debug!("receive error - got={:?}", r);
+            Err(ErrorCode::Error.into())
         }
-    )
+    })
 }

media/audioipc/client/src/stream.rs

@@ -4,14 +4,15 @@
 // accompanying file LICENSE for details
 
 use ClientContext;
+use {set_in_callback, assert_not_in_callback};
 use audioipc::{ClientMessage, Connection, ServerMessage, messages};
-use audioipc::shm::{SharedMemSlice, SharedMemMutSlice};
+use audioipc::shm::{SharedMemMutSlice, SharedMemSlice};
 use cubeb_backend::Stream;
 use cubeb_core::{ErrorCode, Result, ffi};
 use std::ffi::CString;
+use std::fs::File;
 use std::os::raw::c_void;
 use std::os::unix::io::FromRawFd;
-use std::fs::File;
 use std::ptr;
 use std::thread;
 
@@ -27,7 +28,7 @@ pub struct ClientStream<'ctx> {
 
 fn stream_thread(
     mut conn: Connection,
-    input_shm: SharedMemSlice,
+    input_shm: &SharedMemSlice,
     mut output_shm: SharedMemMutSlice,
     data_cb: ffi::cubeb_data_callback,
     state_cb: ffi::cubeb_state_callback,
@@ -38,7 +39,7 @@ fn stream_thread(
             Ok(r) => r,
             Err(e) => {
                 debug!("stream_thread: Failed to receive message: {:?}", e);
-                continue;
+                return;
             },
         };
 
@@ -48,14 +49,21 @@ fn stream_thread(
                 return;
             },
             ClientMessage::StreamDataCallback(nframes, frame_size) => {
-                info!(
+                trace!(
                     "stream_thread: Data Callback: nframes={} frame_size={}",
                     nframes,
                     frame_size
                 );
                 // TODO: This is proof-of-concept. Make it better.
-                let input_ptr: *const u8 = input_shm.get_slice(nframes as usize * frame_size).unwrap().as_ptr();
+                let input_ptr: *const u8 = input_shm
-                let output_ptr: *mut u8 = output_shm.get_mut_slice(nframes as usize * frame_size).unwrap().as_mut_ptr();
+                    .get_slice(nframes as usize * frame_size)
+                    .unwrap()
+                    .as_ptr();
+                let output_ptr: *mut u8 = output_shm
+                    .get_mut_slice(nframes as usize * frame_size)
+                    .unwrap()
+                    .as_mut_ptr();
+                set_in_callback(true);
                 let nframes = data_cb(
                     ptr::null_mut(),
                     user_ptr as *mut c_void,
@@ -63,11 +71,18 @@ fn stream_thread(
                     output_ptr as *mut _,
                     nframes as _
                 );
-                conn.send(ServerMessage::StreamDataCallback(nframes as isize)).unwrap();
+                set_in_callback(false);
+                let r = conn.send(ServerMessage::StreamDataCallback(nframes as isize));
+                if r.is_err() {
+                    debug!("stream_thread: Failed to send StreamDataCallback: {:?}", r);
+                    return;
+                }
             },
             ClientMessage::StreamStateCallback(state) => {
                 info!("stream_thread: State Callback: {:?}", state);
+                set_in_callback(true);
                 state_cb(ptr::null_mut(), user_ptr as *mut _, state);
+                set_in_callback(false);
             },
             m => {
                 info!("Unexpected ClientMessage: {:?}", m);
@@ -84,25 +99,30 @@ impl<'ctx> ClientStream<'ctx> {
         state_callback: ffi::cubeb_state_callback,
         user_ptr: *mut c_void,
     ) -> Result<*mut ffi::cubeb_stream> {
+        assert_not_in_callback();
+        let mut conn = ctx.connection();
 
-        ctx.conn()
+        let r = conn.send(ServerMessage::StreamInit(init_params));
-            .send(ServerMessage::StreamInit(init_params))
+        if r.is_err() {
-            .unwrap();
+            debug!("ClientStream::init: Failed to send StreamInit: {:?}", r);
+            return Err(ErrorCode::Error.into());
+        }
 
-        let r = match ctx.conn().receive_with_fd::<ClientMessage>() {
+        let r = match conn.receive_with_fd::<ClientMessage>() {
             Ok(r) => r,
             Err(_) => return Err(ErrorCode::Error.into()),
         };
 
-        let (token, conn) = match r {
+        let (token, conn2) = match r {
-            (ClientMessage::StreamCreated(tok), Some(fd)) => (tok, unsafe {
+            ClientMessage::StreamCreated(tok) => {
-                Connection::from_raw_fd(fd)
+                let fd = conn.take_fd();
-            }),
+                if fd.is_none() {
-            (ClientMessage::StreamCreated(_), None) => {
                     debug!("Missing fd!");
                     return Err(ErrorCode::Error.into());
+                }
+                (tok, unsafe { Connection::from_raw_fd(fd.unwrap()) })
             },
-            (m, _) => {
+            m => {
                 debug!("Unexpected message: {:?}", m);
                 return Err(ErrorCode::Error.into());
             },
@@ -111,45 +131,60 @@ impl<'ctx> ClientStream<'ctx> {
         // TODO: It'd be nicer to receive these two fds as part of
         // StreamCreated, but that requires changing sendmsg/recvmsg to
         // support multiple fds.
-        let r = match ctx.conn().receive_with_fd::<ClientMessage>() {
+        let r = match conn.receive_with_fd::<ClientMessage>() {
             Ok(r) => r,
             Err(_) => return Err(ErrorCode::Error.into()),
         };
 
         let input_file = match r {
-            (ClientMessage::StreamCreatedInputShm, Some(fd)) => unsafe {
+            ClientMessage::StreamCreatedInputShm => {
-                File::from_raw_fd(fd)
+                let fd = conn.take_fd();
+                if fd.is_none() {
+                    debug!("Missing fd!");
+                    return Err(ErrorCode::Error.into());
+                }
+                unsafe { File::from_raw_fd(fd.unwrap()) }
             },
-            (m, _) => {
+            m => {
                 debug!("Unexpected message: {:?}", m);
                 return Err(ErrorCode::Error.into());
             },
         };
 
-        let input_shm = SharedMemSlice::from(input_file,
+        let input_shm = SharedMemSlice::from(input_file, SHM_AREA_SIZE).unwrap();
-                                             SHM_AREA_SIZE).unwrap();
 
-        let r = match ctx.conn().receive_with_fd::<ClientMessage>() {
+        let r = match conn.receive_with_fd::<ClientMessage>() {
             Ok(r) => r,
             Err(_) => return Err(ErrorCode::Error.into()),
         };
 
         let output_file = match r {
-            (ClientMessage::StreamCreatedOutputShm, Some(fd)) => unsafe {
+            ClientMessage::StreamCreatedOutputShm => {
-                File::from_raw_fd(fd)
+                let fd = conn.take_fd();
+                if fd.is_none() {
+                    debug!("Missing fd!");
+                    return Err(ErrorCode::Error.into());
+                }
+                unsafe { File::from_raw_fd(fd.unwrap()) }
             },
-            (m, _) => {
+            m => {
                 debug!("Unexpected message: {:?}", m);
                 return Err(ErrorCode::Error.into());
             },
         };
 
-        let output_shm = SharedMemMutSlice::from(output_file,
+        let output_shm = SharedMemMutSlice::from(output_file, SHM_AREA_SIZE).unwrap();
-                                                 SHM_AREA_SIZE).unwrap();
 
         let user_data = user_ptr as usize;
         let join_handle = thread::spawn(move || {
-            stream_thread(conn, input_shm, output_shm, data_callback, state_callback, user_data)
+            stream_thread(
+                conn2,
+                &input_shm,
+                output_shm,
+                data_callback,
+                state_callback,
+                user_data
+            )
         });
 
         Ok(Box::into_raw(Box::new(ClientStream {
@@ -162,48 +197,76 @@ impl<'ctx> ClientStream<'ctx> {
 
 impl<'ctx> Drop for ClientStream<'ctx> {
     fn drop(&mut self) {
-        let _: Result<()> = send_recv!(self.context.conn(), StreamDestroy(self.token) => StreamDestroyed);
+        let mut conn = self.context.connection();
+        let r = conn.send(ServerMessage::StreamDestroy(self.token));
+        if r.is_err() {
+            debug!("ClientStream::Drop send error={:?}", r);
+        } else {
+            let r = conn.receive();
+            if let Ok(ClientMessage::StreamDestroyed) = r {
+            } else {
+                debug!("ClientStream::Drop receive error={:?}", r);
+            }
+        }
+        // XXX: This is guaranteed to wait forever if the send failed.
         self.join_handle.take().unwrap().join().unwrap();
     }
 }
 
 impl<'ctx> Stream for ClientStream<'ctx> {
     fn start(&self) -> Result<()> {
-        send_recv!(self.context.conn(), StreamStart(self.token) => StreamStarted)
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamStart(self.token) => StreamStarted)
     }
 
     fn stop(&self) -> Result<()> {
-        send_recv!(self.context.conn(), StreamStop(self.token) => StreamStopped)
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamStop(self.token) => StreamStopped)
     }
 
     fn reset_default_device(&self) -> Result<()> {
-        send_recv!(self.context.conn(), StreamResetDefaultDevice(self.token) => StreamDefaultDeviceReset)
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamResetDefaultDevice(self.token) => StreamDefaultDeviceReset)
     }
 
     fn position(&self) -> Result<u64> {
-        send_recv!(self.context.conn(), StreamGetPosition(self.token) => StreamPosition())
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamGetPosition(self.token) => StreamPosition())
     }
 
     fn latency(&self) -> Result<u32> {
-        send_recv!(self.context.conn(), StreamGetLatency(self.token) => StreamLatency())
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamGetLatency(self.token) => StreamLatency())
     }
 
     fn set_volume(&self, volume: f32) -> Result<()> {
-        send_recv!(self.context.conn(), StreamSetVolume(self.token, volume) => StreamVolumeSet)
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamSetVolume(self.token, volume) => StreamVolumeSet)
     }
 
     fn set_panning(&self, panning: f32) -> Result<()> {
-        send_recv!(self.context.conn(), StreamSetPanning(self.token, panning) => StreamPanningSet)
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        send_recv!(conn, StreamSetPanning(self.token, panning) => StreamPanningSet)
     }
 
     fn current_device(&self) -> Result<*const ffi::cubeb_device> {
-        match send_recv!(self.context.conn(), StreamGetCurrentDevice(self.token) => StreamCurrentDevice()) {
+        assert_not_in_callback();
+        let mut conn = self.context.connection();
+        match send_recv!(conn, StreamGetCurrentDevice(self.token) => StreamCurrentDevice()) {
             Ok(d) => Ok(Box::into_raw(Box::new(d.into()))),
             Err(e) => Err(e),
         }
     }
 
     fn device_destroy(&self, device: *const ffi::cubeb_device) -> Result<()> {
+        assert_not_in_callback();
         // It's all unsafe...
         if !device.is_null() {
             unsafe {
@@ -224,6 +287,7 @@ impl<'ctx> Stream for ClientStream<'ctx> {
         &self,
         _device_changed_callback: ffi::cubeb_device_changed_callback,
     ) -> Result<()> {
+        assert_not_in_callback();
         Ok(())
     }
 }

media/audioipc/server/Cargo.toml

@@ -6,8 +6,9 @@ description = "Remote cubeb server"
 
 [dependencies]
 audioipc = { path = "../audioipc" }
-cubeb = { path = "../../cubeb-rs/cubeb-api" }
 cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
+cubeb = { path = "../../cubeb-rs/cubeb-api" }
+bytes = "0.4"
 error-chain = "0.10.0"
 lazycell = "^0.4"
 log = "^0.3.6"

media/audioipc/server/src/channel.rs

@@ -114,10 +114,10 @@ impl Evented for ReceiverCtl {
             let _ = set_readiness.set_readiness(Ready::readable());
         }
 
-        self.registration.fill(registration).ok().expect(
+        self.registration.fill(registration).expect(
             "unexpected state encountered"
         );
-        self.inner.set_readiness.fill(set_readiness).ok().expect(
+        self.inner.set_readiness.fill(set_readiness).expect(
             "unexpected state encountered"
         );
 
@@ -136,7 +136,7 @@ impl Evented for ReceiverCtl {
 
     fn deregister(&self, poll: &Poll) -> io::Result<()> {
         match self.registration.borrow() {
-            Some(registration) => <Registration as Evented>::deregister(&registration, poll),
+            Some(registration) => <Registration as Evented>::deregister(registration, poll),
             None => Err(io::Error::new(
                 io::ErrorKind::Other,
                 "receiver not registered"
@@ -192,19 +192,19 @@ impl<T> From<io::Error> for TrySendError<T> {
 
 impl<T: Any> error::Error for SendError<T> {
     fn description(&self) -> &str {
-        match self {
+        match *self {
-            &SendError::Io(ref io_err) => io_err.description(),
+            SendError::Io(ref io_err) => io_err.description(),
-            &SendError::Disconnected(..) => "Disconnected",
+            SendError::Disconnected(..) => "Disconnected",
         }
     }
 }
 
 impl<T: Any> error::Error for TrySendError<T> {
     fn description(&self) -> &str {
-        match self {
+        match *self {
-            &TrySendError::Io(ref io_err) => io_err.description(),
+            TrySendError::Io(ref io_err) => io_err.description(),
-            &TrySendError::Full(..) => "Full",
+            TrySendError::Full(..) => "Full",
-            &TrySendError::Disconnected(..) => "Disconnected",
+            TrySendError::Disconnected(..) => "Disconnected",
         }
     }
 }
@@ -235,17 +235,17 @@ impl<T> fmt::Display for TrySendError<T> {
 
 #[inline]
 fn format_send_error<T>(e: &SendError<T>, f: &mut fmt::Formatter) -> fmt::Result {
-    match e {
+    match *e {
-        &SendError::Io(ref io_err) => write!(f, "{}", io_err),
+        SendError::Io(ref io_err) => write!(f, "{}", io_err),
-        &SendError::Disconnected(..) => write!(f, "Disconnected"),
+        SendError::Disconnected(..) => write!(f, "Disconnected"),
     }
 }
 
 #[inline]
 fn format_try_send_error<T>(e: &TrySendError<T>, f: &mut fmt::Formatter) -> fmt::Result {
-    match e {
+    match *e {
-        &TrySendError::Io(ref io_err) => write!(f, "{}", io_err),
+        TrySendError::Io(ref io_err) => write!(f, "{}", io_err),
-        &TrySendError::Full(..) => write!(f, "Full"),
+        TrySendError::Full(..) => write!(f, "Full"),
-        &TrySendError::Disconnected(..) => write!(f, "Disconnected"),
+        TrySendError::Disconnected(..) => write!(f, "Disconnected"),
     }
 }

media/audioipc/server/src/lib.rs

@@ -5,6 +5,7 @@ extern crate error_chain;
 extern crate log;
 
 extern crate audioipc;
+extern crate bytes;
 extern crate cubeb;
 extern crate cubeb_core;
 extern crate lazycell;
@@ -12,14 +13,21 @@ extern crate mio;
 extern crate mio_uds;
 extern crate slab;
 
-use audioipc::messages::{ClientMessage, DeviceInfo, ServerMessage, StreamParams};
+use audioipc::AutoCloseFd;
+use audioipc::async::{Async, AsyncRecvFd, AsyncSendFd};
+use audioipc::codec::{Decoder, encode};
+use audioipc::messages::{ClientMessage, DeviceInfo, ServerMessage, StreamInitParams, StreamParams};
 use audioipc::shm::{SharedMemReader, SharedMemWriter};
+use bytes::{Bytes, BytesMut};
 use cubeb_core::binding::Binding;
 use cubeb_core::ffi;
-use mio::Token;
+use mio::{Ready, Token};
-use mio_uds::UnixListener;
+use mio_uds::{UnixListener, UnixStream};
 use std::{slice, thread};
+use std::collections::VecDeque;
+use std::collections::HashSet;
 use std::convert::From;
+use std::io::Cursor;
 use std::os::raw::c_void;
 use std::os::unix::prelude::*;
 use std::sync::Arc;
@@ -52,14 +60,14 @@ struct Callback {
     output_frame_size: u16,
     connection: audioipc::Connection,
     input_shm: SharedMemWriter,
-    output_shm: SharedMemReader,
+    output_shm: SharedMemReader
 }
 
 impl cubeb::StreamCallback for Callback {
     type Frame = u8;
 
     fn data_callback(&mut self, input: &[u8], output: &mut [u8]) -> isize {
-        info!("Stream data callback: {} {}", input.len(), output.len());
+        trace!("Stream data callback: {} {}", input.len(), output.len());
 
         // len is of input and output is frame len. Turn these into the real lengths.
         let real_input = unsafe {
@@ -68,18 +76,20 @@ impl cubeb::StreamCallback for Callback {
         };
         let real_output = unsafe {
             let size_bytes = output.len() * self.output_frame_size as usize;
-            info!("Resize output to {}", size_bytes);
+            trace!("Resize output to {}", size_bytes);
             slice::from_raw_parts_mut(output.as_mut_ptr(), size_bytes)
         };
 
-        self.input_shm.write(&real_input).unwrap();
+        self.input_shm.write(real_input).unwrap();
 
-        self.connection
+        let r = self.connection.send(ClientMessage::StreamDataCallback(
-            .send(ClientMessage::StreamDataCallback(
             output.len() as isize,
             self.output_frame_size as usize
-            ))
+        ));
-            .unwrap();
+        if r.is_err() {
+            debug!("data_callback: Failed to send to client - got={:?}", r);
+            return -1;
+        }
 
         let r = self.connection.receive();
         match r {
@@ -101,14 +111,28 @@ impl cubeb::StreamCallback for Callback {
 
     fn state_callback(&mut self, state: cubeb::State) {
         info!("Stream state callback: {:?}", state);
+        // TODO: Share this conversion with the same code in cubeb-rs?
+        let state = match state {
+            cubeb::State::Started => ffi::CUBEB_STATE_STARTED,
+            cubeb::State::Stopped => ffi::CUBEB_STATE_STOPPED,
+            cubeb::State::Drained => ffi::CUBEB_STATE_DRAINED,
+            cubeb::State::Error => ffi::CUBEB_STATE_ERROR,
+        };
+        let r = self.connection.send(
+            ClientMessage::StreamStateCallback(state)
+        );
+        if r.is_err() {
+            debug!("state_callback: Failed to send to client - got={:?}", r);
+        }
     }
 }
 
 impl Drop for Callback {
     fn drop(&mut self) {
-        self.connection
+        let r = self.connection.send(ClientMessage::StreamDestroyed);
-            .send(ClientMessage::StreamDestroyed)
+        if r.is_err() {
-            .unwrap();
+            debug!("Callback::drop failed to send StreamDestroyed = {:?}", r);
+        }
     }
 }
 
@@ -121,155 +145,228 @@ const QUIT: Token = Token(std::usize::MAX - 2);
 const SERVER: Token = Token(std::usize::MAX - 1);
 
 struct ServerConn {
-    connection: audioipc::Connection,
+    //connection: audioipc::Connection,
+    io: UnixStream,
     token: Option<Token>,
-    streams: StreamSlab
+    streams: StreamSlab,
+    decoder: Decoder,
+    recv_buffer: BytesMut,
+    send_buffer: BytesMut,
+    pending_send: VecDeque<(Bytes, Option<AutoCloseFd>)>,
+    device_ids: HashSet<usize>
 }
 
 impl ServerConn {
-    fn new<FD>(fd: FD) -> ServerConn
+    fn new(io: UnixStream) -> ServerConn {
-    where
+        let mut sc = ServerConn {
-        FD: IntoRawFd,
+            io: io,
-    {
-        ServerConn {
-            connection: unsafe { audioipc::Connection::from_raw_fd(fd.into_raw_fd()) },
             token: None,
             // TODO: Handle increasing slab size. Pick a good default size.
-            streams: StreamSlab::with_capacity(64)
+            streams: StreamSlab::with_capacity(64),
-        }
+            decoder: Decoder::new(),
+            recv_buffer: BytesMut::with_capacity(4096),
+            send_buffer: BytesMut::with_capacity(4096),
+            pending_send: VecDeque::new(),
+            device_ids: HashSet::new()
+        };
+        sc.device_ids.insert(0); // nullptr is always a valid (default) device id.
+        sc
     }
 
-    fn process(&mut self, poll: &mut mio::Poll, context: &Result<Option<cubeb::Context>>) -> Result<()> {
+    fn process_read(&mut self, context: &Result<cubeb::Context>) -> Result<Ready> {
-        let r = self.connection.receive();
+        // According to *something*, processing non-blocking stream
-        info!("ServerConn::process: got {:?}", r);
+        // should attempt to read until EWOULDBLOCK is returned.
+        while let Async::Ready((n, fd)) = try!(self.io.recv_buf_fd(&mut self.recv_buffer)) {
+            trace!("Received {} bytes and fd {:?}", n, fd);
 
-        if let &Ok(Some(ref ctx)) = context {
+            // Reading 0 signifies EOF
-            // TODO: Might need a simple state machine to deal with
+            if n == 0 {
-            // create/use/destroy ordering, etc.
+                return Err(
-            // TODO: receive() and all this handling should be moved out
+                    ::errors::ErrorKind::AudioIPC(::audioipc::errors::ErrorKind::Disconnected).into()
-            // of this event loop code.
+                );
-            let msg = try!(r);
-            let _ = try!(self.process_msg(&msg, ctx));
-        } else {
-            self.send_error(cubeb::Error::new());
             }
 
-        poll.reregister(
+            if let Some(fd) = fd {
-            &self.connection,
+                trace!("Unexpectedly received an fd from client.");
-            self.token.unwrap(),
+                let _ = unsafe { AutoCloseFd::from_raw_fd(fd) };
-            mio::Ready::readable(),
-            mio::PollOpt::edge() | mio::PollOpt::oneshot()
-        ).unwrap();
-
-        Ok(())
             }
 
-    fn process_msg(&mut self, msg: &ServerMessage, context: &cubeb::Context) -> Result<()> {
+            // Process all the complete messages contained in
-        match msg {
+            // send.recv_buffer.  It's possible that a read might not
-            &ServerMessage::ClientConnect => {
+            // return a complete message, so self.decoder.decode
-                panic!("already connected");
+            // returns Ok(None).
+            loop {
+                match self.decoder.decode::<ServerMessage>(&mut self.recv_buffer) {
+                    Ok(Some(msg)) => {
+                        info!("ServerConn::process: got {:?}", msg);
+                        try!(self.process_msg(&msg, context));
                     },
-            &ServerMessage::ClientDisconnect => {
+                    Ok(None) => {
-                // TODO:
+                        break;
-                //self.connection.client_disconnect();
-                self.connection
-                    .send(ClientMessage::ClientDisconnected)
-                    .unwrap();
-            },
-
-            &ServerMessage::ContextGetBackendId => {},
-
-            &ServerMessage::ContextGetMaxChannelCount => {
-                match context.max_channel_count() {
-                    Ok(channel_count) => {
-                        self.connection
-                            .send(ClientMessage::ContextMaxChannelCount(channel_count))
-                            .unwrap();
                     },
                     Err(e) => {
-                        self.send_error(e);
+                        return Err(e).chain_err(|| "Failed to decoder ServerMessage");
                     },
                 }
+            }
+        }
+
+        // Send any pending responses to client.
+        self.flush_pending_send()
+    }
+
+    // Process a request coming from the client.
+    fn process_msg(&mut self, msg: &ServerMessage, context: &Result<cubeb::Context>) -> Result<()> {
+        let resp: ClientMessage = if let Ok(ref context) = *context {
+            if let ServerMessage::StreamInit(ref params) = *msg {
+                return self.process_stream_init(context, params);
+            };
+
+            match *msg {
+                ServerMessage::ClientConnect => {
+                    panic!("already connected");
                 },
 
-            &ServerMessage::ContextGetMinLatency(ref params) => {
+                ServerMessage::ClientDisconnect => {
+                    // TODO:
+                    //self.connection.client_disconnect();
+                    ClientMessage::ClientDisconnected
+                },
 
+                ServerMessage::ContextGetBackendId => ClientMessage::ContextBackendId(),
+
+                ServerMessage::ContextGetMaxChannelCount => {
+                    context
+                        .max_channel_count()
+                        .map(ClientMessage::ContextMaxChannelCount)
+                        .unwrap_or_else(error)
+                },
+
+                ServerMessage::ContextGetMinLatency(ref params) => {
                     let format = cubeb::SampleFormat::from(params.format);
                     let layout = cubeb::ChannelLayout::from(params.layout);
 
                     let params = cubeb::StreamParamsBuilder::new()
                         .format(format)
-                    .rate(params.rate as _)
+                        .rate(u32::from(params.rate))
-                    .channels(params.channels as _)
+                        .channels(u32::from(params.channels))
                         .layout(layout)
                         .take();
 
-                match context.min_latency(&params) {
+                    context
-                    Ok(latency) => {
+                        .min_latency(&params)
-                        self.connection
+                        .map(ClientMessage::ContextMinLatency)
-                            .send(ClientMessage::ContextMinLatency(latency))
+                        .unwrap_or_else(error)
-                            .unwrap();
-                    },
-                    Err(e) => {
-                        self.send_error(e);
-                    },
-                }
                 },
 
-            &ServerMessage::ContextGetPreferredSampleRate => {
+                ServerMessage::ContextGetPreferredSampleRate => {
-                match context.preferred_sample_rate() {
+                    context
-                    Ok(rate) => {
+                        .preferred_sample_rate()
-                        self.connection
+                        .map(ClientMessage::ContextPreferredSampleRate)
-                            .send(ClientMessage::ContextPreferredSampleRate(rate))
+                        .unwrap_or_else(error)
-                            .unwrap();
-                    },
-                    Err(e) => {
-                        self.send_error(e);
-                    },
-                }
                 },
 
-            &ServerMessage::ContextGetPreferredChannelLayout => {
+                ServerMessage::ContextGetPreferredChannelLayout => {
-                match context.preferred_channel_layout() {
+                    context
-                    Ok(layout) => {
+                        .preferred_channel_layout()
-                        self.connection
+                        .map(|l| ClientMessage::ContextPreferredChannelLayout(l as _))
-                            .send(ClientMessage::ContextPreferredChannelLayout(layout as _))
+                        .unwrap_or_else(error)
-                            .unwrap();
-                    },
-                    Err(e) => {
-                        self.send_error(e);
-                    },
-                }
                 },
 
-            &ServerMessage::ContextGetDeviceEnumeration(device_type) => {
+                ServerMessage::ContextGetDeviceEnumeration(device_type) => {
-                match context.enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type)) {
+                    context
-                    Ok(devices) => {
+                        .enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type))
-                        let v: Vec<DeviceInfo> = devices.iter().map(|i| i.raw().into()).collect();
+                        .map(|devices| {
-                        self.connection
+                            let v: Vec<DeviceInfo> = devices.iter()
-                            .send(ClientMessage::ContextEnumeratedDevices(v))
+                                                            .map(|i| i.raw().into())
-                            .unwrap();
+                                                            .collect();
-                    },
+                            for i in &v {
-                    Err(e) => {
+                                self.device_ids.insert(i.devid);
-                        self.send_error(e);
-                    },
                             }
+                            ClientMessage::ContextEnumeratedDevices(v)
+                        })
+                        .unwrap_or_else(error)
                 },
 
-            &ServerMessage::StreamInit(ref params) => {
+                ServerMessage::StreamInit(_) => {
+                    panic!("StreamInit should have already been handled.");
+                },
+
+                ServerMessage::StreamDestroy(stm_tok) => {
+                    self.streams.remove(stm_tok);
+                    ClientMessage::StreamDestroyed
+                },
+
+                ServerMessage::StreamStart(stm_tok) => {
+                    let _ = self.streams[stm_tok].start();
+                    ClientMessage::StreamStarted
+                },
+
+                ServerMessage::StreamStop(stm_tok) => {
+                    let _ = self.streams[stm_tok].stop();
+                    ClientMessage::StreamStopped
+                },
+
+                ServerMessage::StreamGetPosition(stm_tok) => {
+                    self.streams[stm_tok]
+                        .position()
+                        .map(ClientMessage::StreamPosition)
+                        .unwrap_or_else(error)
+                },
+
+                ServerMessage::StreamGetLatency(stm_tok) => {
+                    self.streams[stm_tok]
+                        .latency()
+                        .map(ClientMessage::StreamLatency)
+                        .unwrap_or_else(error)
+                },
+
+                ServerMessage::StreamSetVolume(stm_tok, volume) => {
+                    self.streams[stm_tok]
+                        .set_volume(volume)
+                        .map(|_| ClientMessage::StreamVolumeSet)
+                        .unwrap_or_else(error)
+                },
+
+                ServerMessage::StreamSetPanning(stm_tok, panning) => {
+                    self.streams[stm_tok]
+                        .set_panning(panning)
+                        .map(|_| ClientMessage::StreamPanningSet)
+                        .unwrap_or_else(error)
+                },
+
+                ServerMessage::StreamGetCurrentDevice(stm_tok) => {
+                    self.streams[stm_tok]
+                        .current_device()
+                        .map(|device| ClientMessage::StreamCurrentDevice(device.into()))
+                        .unwrap_or_else(error)
+                },
+
+                _ => {
+                    bail!("Unexpected Message");
+                },
+            }
+        } else {
+            error(cubeb::Error::new())
+        };
+
+        debug!("process_msg: req={:?}, resp={:?}", msg, resp);
+
+        self.queue_message(resp)
+    }
+
+    // Stream init is special, so it's been separated from process_msg.
+    fn process_stream_init(&mut self, context: &cubeb::Context, params: &StreamInitParams) -> Result<()> {
         fn opt_stream_params(params: Option<&StreamParams>) -> Option<cubeb::StreamParams> {
-                    match params {
+            params.and_then(|p| {
-                        Some(p) => {
                 let raw = ffi::cubeb_stream_params::from(p);
                 Some(unsafe { cubeb::StreamParams::from_raw(&raw as *const _) })
-                        },
+            })
-                        None => None,
-                    }
         }
 
         fn frame_size_in_bytes(params: Option<cubeb::StreamParams>) -> u16 {
-                    match params.as_ref() {
+            params
-                        Some(p) => {
+                .map(|p| {
                     let sample_size = match p.format() {
                         cubeb::SampleFormat::S16LE |
                         cubeb::SampleFormat::S16BE |
@@ -280,14 +377,23 @@ impl ServerConn {
                     };
                     let channel_count = p.channels() as u16;
                     sample_size * channel_count
-                        },
+                })
-                        None => 0,
+                .unwrap_or(0u16)
-                    }
         }
 
+
+        if !self.device_ids.contains(&params.input_device) {
+            bail!("Invalid input_device passed to stream_init");
+        }
         // TODO: Yuck!
         let input_device = unsafe { cubeb::DeviceId::from_raw(params.input_device as *const _) };
+
+        if !self.device_ids.contains(&params.output_device) {
+            bail!("Invalid output_device passed to stream_init");
+        }
+        // TODO: Yuck!
         let output_device = unsafe { cubeb::DeviceId::from_raw(params.output_device as *const _) };
+
         let latency = params.latency_frames;
         let mut builder = cubeb::StreamInitOptionsBuilder::new();
         builder
@@ -314,19 +420,17 @@ impl ServerConn {
         let (conn1, conn2) = audioipc::Connection::pair()?;
         info!("Created connection pair: {:?}-{:?}", conn1, conn2);
 
-                let (input_shm, input_file) =
+        let (input_shm, input_file) = SharedMemWriter::new(&audioipc::get_shm_path("input"), SHM_AREA_SIZE)?;
-                    SharedMemWriter::new(&audioipc::get_shm_path("input"), SHM_AREA_SIZE)?;
+        let (output_shm, output_file) = SharedMemReader::new(&audioipc::get_shm_path("output"), SHM_AREA_SIZE)?;
-                let (output_shm, output_file) =
-                    SharedMemReader::new(&audioipc::get_shm_path("output"), SHM_AREA_SIZE)?;
 
-                match context.stream_init(
+        let err = match context.stream_init(
             &params,
             Callback {
                 input_frame_size: input_frame_size,
                 output_frame_size: output_frame_size,
                 connection: conn2,
                 input_shm: input_shm,
-                        output_shm: output_shm,
+                output_shm: output_shm
             }
         ) {
             Ok(stream) => {
@@ -345,100 +449,120 @@ impl ServerConn {
                     },
                 };
 
-                        self.connection
+                try!(self.queue_init_messages(
-                            .send_with_fd(ClientMessage::StreamCreated(stm_tok), Some(conn1))
+                    stm_tok,
-                            .unwrap();
+                    conn1,
-                        // TODO: It'd be nicer to send these as part of
+                    input_file,
-                        // StreamCreated, but that requires changing
+                    output_file
-                        // sendmsg/recvmsg to support multiple fds.
+                ));
-                        self.connection
-                            .send_with_fd(ClientMessage::StreamCreatedInputShm, Some(input_file))
-                            .unwrap();
-                        self.connection
-                            .send_with_fd(ClientMessage::StreamCreatedOutputShm, Some(output_file))
-                            .unwrap();
-                    },
-                    Err(e) => {
-                        self.send_error(e);
-                    },
-                }
-            },
-
-            &ServerMessage::StreamDestroy(stm_tok) => {
-                self.streams.remove(stm_tok);
-                self.connection
-                    .send(ClientMessage::StreamDestroyed)
-                    .unwrap();
-            },
-
-            &ServerMessage::StreamStart(stm_tok) => {
-                let _ = self.streams[stm_tok].start();
-                self.connection.send(ClientMessage::StreamStarted).unwrap();
-            },
-            &ServerMessage::StreamStop(stm_tok) => {
-                let _ = self.streams[stm_tok].stop();
-                self.connection.send(ClientMessage::StreamStopped).unwrap();
-            },
-            &ServerMessage::StreamGetPosition(stm_tok) => {
-                match self.streams[stm_tok].position() {
-                    Ok(position) => {
-                        self.connection
-                            .send(ClientMessage::StreamPosition(position))
-                            .unwrap();
-                    },
-                    Err(e) => {
-                        self.send_error(e);
-                    },
-                }
-            },
-            &ServerMessage::StreamGetLatency(stm_tok) => {
-                match self.streams[stm_tok].latency() {
-                    Ok(latency) => {
-                        self.connection
-                            .send(ClientMessage::StreamLatency(latency))
-                            .unwrap();
-                    },
-                    Err(e) => self.send_error(e),
-                }
-            },
-            &ServerMessage::StreamSetVolume(stm_tok, volume) => {
-                let _ = self.streams[stm_tok].set_volume(volume);
-                self.connection
-                    .send(ClientMessage::StreamVolumeSet)
-                    .unwrap();
-            },
-            &ServerMessage::StreamSetPanning(stm_tok, panning) => {
-                let _ = self.streams[stm_tok].set_panning(panning);
-                self.connection
-                    .send(ClientMessage::StreamPanningSet)
-                    .unwrap();
-            },
-            &ServerMessage::StreamGetCurrentDevice(stm_tok) => {
-                let err = match self.streams[stm_tok].current_device() {
-                    Ok(device) => {
-                        // TODO: Yuck!
-                        self.connection
-                            .send(ClientMessage::StreamCurrentDevice(device.into()))
-                            .unwrap();
                 None
             },
-                    Err(e) => Some(e),
+            Err(e) => Some(error(e)),
         };
-                if let Some(e) = err {
+
-                    self.send_error(e);
+        if let Some(err) = err {
-                }
+            try!(self.queue_message(err))
-            },
-            _ => {
-                bail!("Unexpected Message");
-            },
         }
+
         Ok(())
     }
 
-    fn send_error(&mut self, error: cubeb::Error) {
+    fn queue_init_messages<T, U, V>(&mut self, stm_tok: usize, conn: T, input_file: U, output_file: V) -> Result<()>
-        self.connection
+    where
-            .send(ClientMessage::ContextError(error.raw_code()))
+        T: IntoRawFd,
-            .unwrap();
+        U: IntoRawFd,
+        V: IntoRawFd,
+    {
+        try!(self.queue_message_fd(
+            ClientMessage::StreamCreated(stm_tok),
+            conn
+        ));
+        try!(self.queue_message_fd(
+            ClientMessage::StreamCreatedInputShm,
+            input_file
+        ));
+        try!(self.queue_message_fd(
+            ClientMessage::StreamCreatedOutputShm,
+            output_file
+        ));
+        Ok(())
+    }
+
+    fn queue_message(&mut self, msg: ClientMessage) -> Result<()> {
+        debug!("queue_message: {:?}", msg);
+        encode::<ClientMessage>(&mut self.send_buffer, &msg).or_else(|e| {
+            Err(e).chain_err(|| "Failed to encode msg into send buffer")
+        })
+    }
+
+    // Since send_fd supports sending one RawFd at a time, queuing a
+    // message with a RawFd forces use to take the current send_buffer
+    // and move it pending queue.
+    fn queue_message_fd<FD: IntoRawFd>(&mut self, msg: ClientMessage, fd: FD) -> Result<()> {
+        let fd = fd.into_raw_fd();
+        debug!("queue_message_fd: {:?} {:?}", msg, fd);
+        try!(self.queue_message(msg));
+        self.take_pending_send(Some(fd));
+        Ok(())
+    }
+
+    // Take the current messages in the send_buffer and move them to
+    // pending queue.
+    fn take_pending_send(&mut self, fd: Option<RawFd>) {
+        let pending = self.send_buffer.take().freeze();
+        debug!("take_pending_send: ({:?} {:?})", pending, fd);
+        self.pending_send.push_back((
+            pending,
+            fd.map(|fd| unsafe { AutoCloseFd::from_raw_fd(fd) })
+        ));
+    }
+
+    // Process the pending queue and send them to client.
+    fn flush_pending_send(&mut self) -> Result<Ready> {
+        debug!("flush_pending_send");
+        // take any pending messages in the send buffer.
+        if !self.send_buffer.is_empty() {
+            self.take_pending_send(None);
+        }
+
+        trace!("pending queue: {:?}", self.pending_send);
+
+        let mut result = Ready::readable();
+        let mut processed = 0;
+
+        for &mut (ref mut buf, ref mut fd) in &mut self.pending_send {
+            trace!("sending buf {:?}, fd {:?}", buf, fd);
+            let r = {
+                let mut src = Cursor::new(buf.as_ref());
+                let fd = match *fd {
+                    Some(ref fd) => Some(fd.as_raw_fd()),
+                    None => None,
+                };
+                try!(self.io.send_buf_fd(&mut src, fd))
+            };
+            match r {
+                Async::Ready(n) if n == buf.len() => {
+                    processed += 1;
+                },
+                Async::Ready(n) => {
+                    let _ = buf.split_to(n);
+                    let _ = fd.take();
+                    result.insert(Ready::writable());
+                    break;
+                },
+                Async::NotReady => {
+                    result.insert(Ready::writable());
+                },
+            }
+        }
+
+        debug!("processed {} buffers", processed);
+
+        self.pending_send = self.pending_send.split_off(processed);
+
+        trace!("pending queue: {:?}", self.pending_send);
+
+        Ok(result)
     }
 }
 
@@ -448,7 +572,7 @@ pub struct Server {
     // Ok(Some(ctx)) - Server has successfully created cubeb::Context.
     // Err(_)        - Server has tried and failed to create cubeb::Context.
     //                 Don't try again.
-    context: Result<Option<cubeb::Context>>,
+    context: Option<Result<cubeb::Context>>,
     conns: Slab<ServerConn>
 }
 
@@ -456,7 +580,7 @@ impl Server {
     pub fn new(socket: UnixListener) -> Server {
         Server {
             socket: socket,
-            context: Ok(None),
+            context: None,
             conns: Slab::with_capacity(16)
         }
     }
@@ -486,7 +610,7 @@ impl Server {
         // Register the connection
         self.conns[token].token = Some(token);
         poll.register(
-            &self.conns[token].connection,
+            &self.conns[token].io,
             token,
             mio::Ready::readable(),
             mio::PollOpt::edge() | mio::PollOpt::oneshot()
@@ -500,10 +624,10 @@ impl Server {
 
         // Since we have a connection try creating a cubeb context. If
         // it fails, mark the failure with Err.
-        if let Ok(None) = self.context {
+        if self.context.is_none() {
-            self.context = cubeb::Context::init("AudioIPC Server", None)
+            self.context = Some(cubeb::Context::init("AudioIPC Server", None).or_else(|e| {
-                .and_then(|ctx| Ok(Some(ctx)))
+                Err(e).chain_err(|| "Unable to create cubeb context.")
-                .or_else(|err| Err(err.into()));
+            }))
         }
 
         Ok(())
@@ -520,11 +644,8 @@ impl Server {
         for event in events.iter() {
             match event.token() {
                 SERVER => {
-                    match self.accept(poll) {
+                    if let Err(e) = self.accept(poll) {
-                        Err(e) => {
                         error!("server accept error: {}", e);
-                        },
-                        _ => {},
                     };
                 },
                 QUIT => {
@@ -532,27 +653,45 @@ impl Server {
                     bail!("quit");
                 },
                 token => {
-                    debug!("token {:?} ready", token);
+                    trace!("token {:?} ready", token);
 
-                    let r = self.conns[token].process(poll, &self.context);
+                    let context = self.context.as_ref().expect(
+                        "Shouldn't receive a message before accepting connection."
+                    );
 
-                    debug!("got {:?}", r);
+                    let mut readiness = Ready::readable();
+
+                    if event.readiness().is_readable() {
+                        let r = self.conns[token].process_read(context);
+                        trace!("got {:?}", r);
 
-                    // TODO: Handle disconnection etc.
-                    // TODO: Should be handled at a higher level by a
-                    // disconnect message.
                         if let Err(e) = r {
                             debug!("dropped client {:?} due to error {:?}", token, e);
                             self.conns.remove(token);
                             continue;
                         }
+                    };
 
-                    // poll.reregister(
+                    if event.readiness().is_writable() {
-                    //     &self.conn(token).connection,
+                        let r = self.conns[token].flush_pending_send();
-                    //     token,
+                        trace!("got {:?}", r);
-                    //     mio::Ready::readable(),
+
-                    //     mio::PollOpt::edge() | mio::PollOpt::oneshot()
+                        match r {
-                    // ).unwrap();
+                            Ok(r) => readiness = r,
+                            Err(e) => {
+                                debug!("dropped client {:?} due to error {:?}", token, e);
+                                self.conns.remove(token);
+                                continue;
+                            },
+                        }
+                    };
+
+                    poll.reregister(
+                        &self.conns[token].io,
+                        token,
+                        readiness,
+                        mio::PollOpt::edge() | mio::PollOpt::oneshot()
+                    ).unwrap();
                 },
             }
         }
@@ -587,18 +726,35 @@ pub fn run(running: Arc<AtomicBool>) -> Result<()> {
             bail!("server quit due to ctrl-c");
         }
 
-        let _ = try!(server.poll(&mut poll));
+        try!(server.poll(&mut poll));
     }
 
     //poll.deregister(&server.socket).unwrap();
 }
 
+fn error(error: cubeb::Error) -> ClientMessage {
+    ClientMessage::ContextError(error.raw_code())
+}
+
+struct ServerWrapper {
+    thread_handle: std::thread::JoinHandle<()>,
+    sender_ctl: channel::SenderCtl,
+}
+
+impl ServerWrapper {
+    fn shutdown(self) {
+        // Dropping SenderCtl here will notify the other end.
+        drop(self.sender_ctl);
+        self.thread_handle.join().unwrap();
+    }
+}
+
 #[no_mangle]
 pub extern "C" fn audioipc_server_start() -> *mut c_void {
 
     let (tx, rx) = channel::ctl_pair();
 
-    thread::spawn(move || {
+    let handle = thread::spawn(move || {
         // Ignore result.
         let _ = std::fs::remove_file(audioipc::get_uds_path());
 
@@ -617,20 +773,22 @@ pub extern "C" fn audioipc_server_start() -> *mut c_void {
             .unwrap();
 
         loop {
-            match server.poll(&mut poll) {
+            if server.poll(&mut poll).is_err() {
-                Err(_) => {
                 return;
-                },
-                _ => (),
             }
         }
     });
 
-    Box::into_raw(Box::new(tx)) as *mut _
+    let wrapper = ServerWrapper {
+        thread_handle: handle,
+        sender_ctl: tx
+    };
+
+    Box::into_raw(Box::new(wrapper)) as *mut _
 }
 
 #[no_mangle]
 pub extern "C" fn audioipc_server_stop(p: *mut c_void) {
-    // Dropping SenderCtl here will notify the other end.
+    let wrapper = unsafe { Box::<ServerWrapper>::from_raw(p as *mut _) };
-    let _ = unsafe { Box::<channel::SenderCtl>::from_raw(p as *mut _) };
+    wrapper.shutdown();
 }