Adding upstream version 115.7.0esr.upstream/115.7.0esrupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

29 files changed, 5636 insertions, 0 deletions

diff --git a/third_party/rust/audioipc2-client/.cargo-checksum.json b/third_party/rust/audioipc2-client/.cargo-checksum.json
new file mode 100644
index 0000000000..c33c0e16a1
--- /dev/null
+++ b/third_party/rust/audioipc2-client/.cargo-checksum.json
@@ -0,0 +1 @@
+{"files":{"Cargo.toml":"9a41bc278a815fdcbe8d07c1d32f5cedb17dec3375993351b942567b4a644f9e","cbindgen.toml":"fb6abe1671497f432a06e40b1db7ed7cd2cceecbd9a2382193ad7534e8855e34","src/context.rs":"93dc43233603409dbb0c5de20c4e838c88bb0711df65cd5afc882f5b5e39f659","src/lib.rs":"c4a6797734489280f6b97dd72c9e51a7bd7be4104592eece3929e29d45cbca4a","src/send_recv.rs":"859abe75b521eb4297c84b30423814b5b87f3c7741ad16fe72189212e123e1ac","src/stream.rs":"3101f4052e35c1600adafedc67aed92d7d5531ff0ff4845de0a5ea3cedef872d"},"package":null}
\ No newline at end of file
diff --git a/third_party/rust/audioipc2-client/Cargo.toml b/third_party/rust/audioipc2-client/Cargo.toml
new file mode 100644
index 0000000000..5ff80d59d3
--- /dev/null
+++ b/third_party/rust/audioipc2-client/Cargo.toml
@@ -0,0 +1,34 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2018"
+name = "audioipc2-client"
+version = "0.5.0"
+authors = [
+    "Matthew Gregan <kinetik@flim.org>",
+    "Dan Glastonbury <dan.glastonbury@gmail.com>",
+]
+description = "Cubeb Backend for talking to remote cubeb server."
+license = "ISC"
+
+[dependencies]
+cubeb-backend = "0.10"
+log = "0.4"
+
+[dependencies.audio_thread_priority]
+version = "0.26.1"
+features = ["winapi"]
+default-features = false
+
+[dependencies.audioipc]
+path = "../audioipc"
+package = "audioipc2"
diff --git a/third_party/rust/audioipc2-client/cbindgen.toml b/third_party/rust/audioipc2-client/cbindgen.toml
new file mode 100644
index 0000000000..b3267889c6
--- /dev/null
+++ b/third_party/rust/audioipc2-client/cbindgen.toml
@@ -0,0 +1,28 @@
+header = """/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */"""
+autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */"""
+include_version = true
+braces = "SameLine"
+line_length = 100
+tab_width = 2
+language = "C++"
+namespaces = ["mozilla", "audioipc2"]
+
+[export]
+item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"]
+
+[parse]
+parse_deps = true
+include = ["audioipc2"]
+
+[fn]
+args = "Vertical"
+rename_args = "GeckoCase"
+
+[struct]
+rename_fields = "GeckoCase"
+
+[defines]
+"windows" = "XP_WIN"
+"unix" = "XP_UNIX"
diff --git a/third_party/rust/audioipc2-client/src/context.rs b/third_party/rust/audioipc2-client/src/context.rs
new file mode 100644
index 0000000000..647d944724
--- /dev/null
+++ b/third_party/rust/audioipc2-client/src/context.rs
@@ -0,0 +1,383 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::stream;
+use crate::{assert_not_in_callback, run_in_callback};
+use crate::{ClientStream, AUDIOIPC_INIT_PARAMS};
+#[cfg(target_os = "linux")]
+use audio_thread_priority::get_current_thread_info;
+#[cfg(not(target_os = "linux"))]
+use audio_thread_priority::promote_current_thread_to_real_time;
+use audioipc::ipccore::EventLoopHandle;
+use audioipc::{ipccore, rpccore, sys, PlatformHandle};
+use audioipc::{
+    messages, messages::DeviceCollectionReq, messages::DeviceCollectionResp, ClientMessage,
+    ServerMessage,
+};
+use cubeb_backend::{
+    capi_new, ffi, Context, ContextOps, DeviceCollectionRef, DeviceId, DeviceType, Error, Ops,
+    Result, Stream, StreamParams, StreamParamsRef,
+};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::{fmt, mem, ptr};
+
+struct CubebClient;
+
+impl rpccore::Client for CubebClient {
+    type ServerMessage = ServerMessage;
+    type ClientMessage = ClientMessage;
+}
+
+pub const CLIENT_OPS: Ops = capi_new!(ClientContext, ClientStream);
+
+// ClientContext's layout *must* match cubeb.c's `struct cubeb` for the
+// common fields.
+#[repr(C)]
+pub struct ClientContext {
+    _ops: *const Ops,
+    rpc: rpccore::Proxy<ServerMessage, ClientMessage>,
+    rpc_thread: ipccore::EventLoopThread,
+    callback_thread: ipccore::EventLoopThread,
+    backend_id: CString,
+    device_collection_rpc: bool,
+    input_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+    output_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+}
+
+impl ClientContext {
+    #[doc(hidden)]
+    pub fn rpc_handle(&self) -> &EventLoopHandle {
+        self.rpc_thread.handle()
+    }
+
+    #[doc(hidden)]
+    pub fn rpc(&self) -> rpccore::Proxy<ServerMessage, ClientMessage> {
+        self.rpc.clone()
+    }
+
+    #[doc(hidden)]
+    pub fn callback_handle(&self) -> &EventLoopHandle {
+        self.callback_thread.handle()
+    }
+}
+
+#[cfg(target_os = "linux")]
+fn promote_thread(rpc: &rpccore::Proxy<ServerMessage, ClientMessage>) {
+    match get_current_thread_info() {
+        Ok(info) => {
+            let bytes = info.serialize();
+            let _ = rpc.call(ServerMessage::PromoteThreadToRealTime(bytes));
+        }
+        Err(_) => {
+            warn!("Could not remotely promote thread to RT.");
+        }
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+fn promote_thread(_rpc: &rpccore::Proxy<ServerMessage, ClientMessage>) {
+    match promote_current_thread_to_real_time(0, 48000) {
+        Ok(_) => {
+            info!("Audio thread promoted to real-time.");
+        }
+        Err(_) => {
+            warn!("Could not promote thread to real-time.");
+        }
+    }
+}
+
+fn register_thread(callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>) {
+    if let Some(func) = callback {
+        let thr = thread::current();
+        let name = CString::new(thr.name().unwrap()).unwrap();
+        func(name.as_ptr());
+    }
+}
+
+fn unregister_thread(callback: Option<extern "C" fn()>) {
+    if let Some(func) = callback {
+        func();
+    }
+}
+
+fn promote_and_register_thread(
+    rpc: &rpccore::Proxy<ServerMessage, ClientMessage>,
+    callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+) {
+    promote_thread(rpc);
+    register_thread(callback);
+}
+
+#[derive(Default)]
+struct DeviceCollectionCallback {
+    cb: ffi::cubeb_device_collection_changed_callback,
+    user_ptr: usize,
+}
+
+struct DeviceCollectionServer {
+    input_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+    output_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+}
+
+impl rpccore::Server for DeviceCollectionServer {
+    type ServerMessage = DeviceCollectionReq;
+    type ClientMessage = DeviceCollectionResp;
+
+    fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage {
+        match req {
+            DeviceCollectionReq::DeviceChange(device_type) => {
+                trace!(
+                    "ctx_thread: DeviceChange Callback: device_type={}",
+                    device_type
+                );
+
+                let devtype = cubeb_backend::DeviceType::from_bits_truncate(device_type);
+
+                let (input_cb, input_user_ptr) = {
+                    let dcb = self.input_device_callback.lock().unwrap();
+                    (dcb.cb, dcb.user_ptr)
+                };
+                let (output_cb, output_user_ptr) = {
+                    let dcb = self.output_device_callback.lock().unwrap();
+                    (dcb.cb, dcb.user_ptr)
+                };
+
+                run_in_callback(|| {
+                    if devtype.contains(cubeb_backend::DeviceType::INPUT) {
+                        unsafe { input_cb.unwrap()(ptr::null_mut(), input_user_ptr as *mut c_void) }
+                    }
+                    if devtype.contains(cubeb_backend::DeviceType::OUTPUT) {
+                        unsafe {
+                            output_cb.unwrap()(ptr::null_mut(), output_user_ptr as *mut c_void)
+                        }
+                    }
+                });
+
+                DeviceCollectionResp::DeviceChange
+            }
+        }
+    }
+}
+
+impl ContextOps for ClientContext {
+    fn init(_context_name: Option<&CStr>) -> Result<Context> {
+        assert_not_in_callback();
+
+        let params = AUDIOIPC_INIT_PARAMS.with(|p| p.replace(None).unwrap());
+        let thread_create_callback = params.thread_create_callback;
+        let thread_destroy_callback = params.thread_destroy_callback;
+
+        let server_connection =
+            unsafe { sys::Pipe::from_raw_handle(PlatformHandle::new(params.server_connection)) };
+
+        let rpc_thread = ipccore::EventLoopThread::new(
+            "AudioIPC Client RPC".to_string(),
+            None,
+            move || register_thread(thread_create_callback),
+            move || unregister_thread(thread_destroy_callback),
+        )
+        .map_err(|_| Error::default())?;
+        let rpc = rpc_thread
+            .handle()
+            .bind_client::<CubebClient>(server_connection)
+            .map_err(|_| Error::default())?;
+        let rpc2 = rpc.clone();
+
+        // Don't let errors bubble from here.  Later calls against this context
+        // will return errors the caller expects to handle.
+        let _ = send_recv!(rpc, ClientConnect(std::process::id()) => ClientConnected);
+
+        let backend_id = send_recv!(rpc, ContextGetBackendId => ContextBackendId())
+            .unwrap_or_else(|_| "(remote error)".to_string());
+        let backend_id = CString::new(backend_id).expect("backend_id query failed");
+
+        // TODO: remove params.pool_size from init params.
+        let callback_thread = ipccore::EventLoopThread::new(
+            "AudioIPC Client Callback".to_string(),
+            Some(params.stack_size),
+            move || promote_and_register_thread(&rpc2, thread_create_callback),
+            move || unregister_thread(thread_destroy_callback),
+        )
+        .map_err(|_| Error::default())?;
+
+        let ctx = Box::new(ClientContext {
+            _ops: &CLIENT_OPS as *const _,
+            rpc,
+            rpc_thread,
+            callback_thread,
+            backend_id,
+            device_collection_rpc: false,
+            input_device_callback: Arc::new(Mutex::new(Default::default())),
+            output_device_callback: Arc::new(Mutex::new(Default::default())),
+        });
+        Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) })
+    }
+
+    fn backend_id(&mut self) -> &CStr {
+        assert_not_in_callback();
+        self.backend_id.as_c_str()
+    }
+
+    fn max_channel_count(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        send_recv!(self.rpc(), ContextGetMaxChannelCount => ContextMaxChannelCount())
+    }
+
+    fn min_latency(&mut self, params: StreamParams) -> Result<u32> {
+        assert_not_in_callback();
+        let params = messages::StreamParams::from(params.as_ref());
+        send_recv!(self.rpc(), ContextGetMinLatency(params) => ContextMinLatency())
+    }
+
+    fn preferred_sample_rate(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        send_recv!(self.rpc(), ContextGetPreferredSampleRate => ContextPreferredSampleRate())
+    }
+
+    fn enumerate_devices(
+        &mut self,
+        devtype: DeviceType,
+        collection: &DeviceCollectionRef,
+    ) -> Result<()> {
+        assert_not_in_callback();
+        let v: Vec<ffi::cubeb_device_info> = send_recv!(
+            self.rpc(), ContextGetDeviceEnumeration(devtype.bits()) => ContextEnumeratedDevices())?
+        .into_iter()
+        .map(|i| i.into())
+        .collect();
+        let mut vs = v.into_boxed_slice();
+        let coll = unsafe { &mut *collection.as_ptr() };
+        coll.device = vs.as_mut_ptr();
+        coll.count = vs.len();
+        // Giving away the memory owned by vs.  Don't free it!
+        // Reclaimed in `device_collection_destroy`.
+        mem::forget(vs);
+        Ok(())
+    }
+
+    fn device_collection_destroy(&mut self, collection: &mut DeviceCollectionRef) -> Result<()> {
+        assert_not_in_callback();
+        unsafe {
+            let coll = &mut *collection.as_ptr();
+            let mut devices = Vec::from_raw_parts(
+                coll.device as *mut ffi::cubeb_device_info,
+                coll.count,
+                coll.count,
+            );
+            for dev in &mut devices {
+                if !dev.device_id.is_null() {
+                    let _ = CString::from_raw(dev.device_id as *mut _);
+                }
+                if !dev.group_id.is_null() {
+                    let _ = CString::from_raw(dev.group_id as *mut _);
+                }
+                if !dev.vendor_name.is_null() {
+                    let _ = CString::from_raw(dev.vendor_name as *mut _);
+                }
+                if !dev.friendly_name.is_null() {
+                    let _ = CString::from_raw(dev.friendly_name as *mut _);
+                }
+            }
+            coll.device = ptr::null_mut();
+            coll.count = 0;
+            Ok(())
+        }
+    }
+
+    fn stream_init(
+        &mut self,
+        stream_name: Option<&CStr>,
+        input_device: DeviceId,
+        input_stream_params: Option<&StreamParamsRef>,
+        output_device: DeviceId,
+        output_stream_params: Option<&StreamParamsRef>,
+        latency_frames: u32,
+        // These params aren't sent to the server
+        data_callback: ffi::cubeb_data_callback,
+        state_callback: ffi::cubeb_state_callback,
+        user_ptr: *mut c_void,
+    ) -> Result<Stream> {
+        assert_not_in_callback();
+
+        let stream_name = stream_name.map(|name| name.to_bytes_with_nul().to_vec());
+
+        let input_stream_params = input_stream_params.map(messages::StreamParams::from);
+        let output_stream_params = output_stream_params.map(messages::StreamParams::from);
+
+        let init_params = messages::StreamInitParams {
+            stream_name,
+            input_device: input_device as usize,
+            input_stream_params,
+            output_device: output_device as usize,
+            output_stream_params,
+            latency_frames,
+        };
+        stream::init(self, init_params, data_callback, state_callback, user_ptr)
+    }
+
+    fn register_device_collection_changed(
+        &mut self,
+        devtype: DeviceType,
+        collection_changed_callback: ffi::cubeb_device_collection_changed_callback,
+        user_ptr: *mut c_void,
+    ) -> Result<()> {
+        assert_not_in_callback();
+
+        if !self.device_collection_rpc {
+            let mut fd = send_recv!(self.rpc(),
+                                 ContextSetupDeviceCollectionCallback =>
+                                 ContextSetupDeviceCollectionCallback())?;
+
+            let stream = unsafe { sys::Pipe::from_raw_handle(fd.platform_handle.take_handle()) };
+
+            let server = DeviceCollectionServer {
+                input_device_callback: self.input_device_callback.clone(),
+                output_device_callback: self.output_device_callback.clone(),
+            };
+
+            self.rpc_handle()
+                .bind_server(server, stream)
+                .map_err(|_| Error::default())?;
+            self.device_collection_rpc = true;
+        }
+
+        if devtype.contains(cubeb_backend::DeviceType::INPUT) {
+            let mut cb = self.input_device_callback.lock().unwrap();
+            cb.cb = collection_changed_callback;
+            cb.user_ptr = user_ptr as usize;
+        }
+        if devtype.contains(cubeb_backend::DeviceType::OUTPUT) {
+            let mut cb = self.output_device_callback.lock().unwrap();
+            cb.cb = collection_changed_callback;
+            cb.user_ptr = user_ptr as usize;
+        }
+
+        let enable = collection_changed_callback.is_some();
+        send_recv!(self.rpc(),
+                   ContextRegisterDeviceCollectionChanged(devtype.bits(), enable) =>
+                   ContextRegisteredDeviceCollectionChanged)
+    }
+}
+
+impl Drop for ClientContext {
+    fn drop(&mut self) {
+        debug!("ClientContext dropped...");
+        let _ = send_recv!(self.rpc(), ClientDisconnect => ClientDisconnected);
+    }
+}
+
+impl fmt::Debug for ClientContext {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ClientContext")
+            .field("_ops", &self._ops)
+            .field("rpc", &self.rpc)
+            .field("core", &self.rpc_thread)
+            .field("cpu_pool", &"...")
+            .finish()
+    }
+}
diff --git a/third_party/rust/audioipc2-client/src/lib.rs b/third_party/rust/audioipc2-client/src/lib.rs
new file mode 100644
index 0000000000..51a68af709
--- /dev/null
+++ b/third_party/rust/audioipc2-client/src/lib.rs
@@ -0,0 +1,81 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+#![warn(unused_extern_crates)]
+
+#[macro_use]
+extern crate log;
+
+#[macro_use]
+mod send_recv;
+mod context;
+mod stream;
+
+use crate::context::ClientContext;
+use crate::stream::ClientStream;
+use audioipc::PlatformHandleType;
+use cubeb_backend::{capi, ffi};
+use std::os::raw::{c_char, c_int};
+
+thread_local!(static IN_CALLBACK: std::cell::RefCell<bool> = std::cell::RefCell::new(false));
+thread_local!(static AUDIOIPC_INIT_PARAMS: std::cell::RefCell<Option<AudioIpcInitParams>> = std::cell::RefCell::new(None));
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug)]
+pub struct AudioIpcInitParams {
+    // Fields only need to be public for ipctest.
+    pub server_connection: PlatformHandleType,
+    pub pool_size: usize,
+    pub stack_size: usize,
+    pub thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+    pub thread_destroy_callback: Option<extern "C" fn()>,
+}
+
+unsafe impl Send for AudioIpcInitParams {}
+
+fn set_in_callback(in_callback: bool) {
+    IN_CALLBACK.with(|b| {
+        assert_eq!(*b.borrow(), !in_callback);
+        *b.borrow_mut() = in_callback;
+    });
+}
+
+fn run_in_callback<F, R>(f: F) -> R
+where
+    F: FnOnce() -> R,
+{
+    set_in_callback(true);
+
+    let r = f();
+
+    set_in_callback(false);
+
+    r
+}
+
+fn assert_not_in_callback() {
+    IN_CALLBACK.with(|b| {
+        assert!(!*b.borrow());
+    });
+}
+
+#[allow(clippy::missing_safety_doc)]
+#[no_mangle]
+/// Entry point from C code.
+pub unsafe extern "C" fn audioipc2_client_init(
+    c: *mut *mut ffi::cubeb,
+    context_name: *const c_char,
+    init_params: *const AudioIpcInitParams,
+) -> c_int {
+    if init_params.is_null() {
+        return cubeb_backend::ffi::CUBEB_ERROR;
+    }
+
+    let init_params = &*init_params;
+
+    AUDIOIPC_INIT_PARAMS.with(|p| {
+        *p.borrow_mut() = Some(*init_params);
+    });
+    capi::capi_init::<ClientContext>(c, context_name)
+}
diff --git a/third_party/rust/audioipc2-client/src/send_recv.rs b/third_party/rust/audioipc2-client/src/send_recv.rs
new file mode 100644
index 0000000000..1134c99a49
--- /dev/null
+++ b/third_party/rust/audioipc2-client/src/send_recv.rs
@@ -0,0 +1,73 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use cubeb_backend::Error;
+use std::os::raw::c_int;
+
+#[doc(hidden)]
+pub fn _err<E>(e: E) -> Error
+where
+    E: Into<Option<c_int>>,
+{
+    match e.into() {
+        Some(e) => Error::from_raw(e),
+        None => Error::error(),
+    }
+}
+
+#[macro_export]
+macro_rules! send_recv {
+    ($rpc:expr, $smsg:ident => $rmsg:ident) => {{
+        let resp = send_recv!(__send $rpc, $smsg);
+        send_recv!(__recv resp, $rmsg)
+    }};
+    ($rpc:expr, $smsg:ident => $rmsg:ident()) => {{
+        let resp = send_recv!(__send $rpc, $smsg);
+        send_recv!(__recv resp, $rmsg __result)
+    }};
+    ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident) => {{
+        let resp = send_recv!(__send $rpc, $smsg, $($a),*);
+        send_recv!(__recv resp, $rmsg)
+    }};
+    ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident()) => {{
+        let resp = send_recv!(__send $rpc, $smsg, $($a),*);
+        send_recv!(__recv resp, $rmsg __result)
+    }};
+    //
+    (__send $rpc:expr, $smsg:ident) => ({
+        $rpc.call(ServerMessage::$smsg)
+    });
+    (__send $rpc:expr, $smsg:ident, $($a:expr),*) => ({
+        $rpc.call(ServerMessage::$smsg($($a),*))
+    });
+    (__recv $resp:expr, $rmsg:ident) => ({
+        match $resp {
+            Ok(ClientMessage::$rmsg) => Ok(()),
+            Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)),
+            Ok(m) => {
+                debug!("received wrong message - got={:?}", m);
+                Err($crate::send_recv::_err(None))
+            },
+            Err(e) => {
+                debug!("received error from rpc - got={:?}", e);
+                Err($crate::send_recv::_err(None))
+            },
+        }
+    });
+    (__recv $resp:expr, $rmsg:ident __result) => ({
+        match $resp {
+            Ok(ClientMessage::$rmsg(v)) => Ok(v),
+            Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)),
+            Ok(m) => {
+                debug!("received wrong message - got={:?}", m);
+                Err($crate::send_recv::_err(None))
+            },
+            Err(e) => {
+                debug!("received error - got={:?}", e);
+                Err($crate::send_recv::_err(None))
+            },
+        }
+    })
+}
diff --git a/third_party/rust/audioipc2-client/src/stream.rs b/third_party/rust/audioipc2-client/src/stream.rs
new file mode 100644
index 0000000000..ab702bc98c
--- /dev/null
+++ b/third_party/rust/audioipc2-client/src/stream.rs
@@ -0,0 +1,339 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::ClientContext;
+use crate::{assert_not_in_callback, run_in_callback};
+use audioipc::messages::StreamCreateParams;
+use audioipc::messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage};
+use audioipc::shm::SharedMem;
+use audioipc::{rpccore, sys};
+use cubeb_backend::{ffi, DeviceRef, Error, Result, Stream, StreamOps};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::ptr;
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+
+pub struct Device(ffi::cubeb_device);
+
+impl Drop for Device {
+    fn drop(&mut self) {
+        unsafe {
+            if !self.0.input_name.is_null() {
+                let _ = CString::from_raw(self.0.input_name as *mut _);
+            }
+            if !self.0.output_name.is_null() {
+                let _ = CString::from_raw(self.0.output_name as *mut _);
+            }
+        }
+    }
+}
+
+// ClientStream's layout *must* match cubeb.c's `struct cubeb_stream` for the
+// common fields.
+#[repr(C)]
+#[derive(Debug)]
+pub struct ClientStream<'ctx> {
+    // This must be a reference to Context for cubeb, cubeb accesses
+    // stream methods via stream->context->ops
+    context: &'ctx ClientContext,
+    user_ptr: *mut c_void,
+    token: usize,
+    device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>,
+    // Signals ClientStream that CallbackServer has dropped.
+    shutdown_rx: mpsc::Receiver<()>,
+}
+
+struct CallbackServer {
+    shm: SharedMem,
+    duplex_input: Option<Vec<u8>>,
+    data_cb: ffi::cubeb_data_callback,
+    state_cb: ffi::cubeb_state_callback,
+    user_ptr: usize,
+    device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>,
+    // Signals ClientStream that CallbackServer has dropped.
+    _shutdown_tx: mpsc::Sender<()>,
+}
+
+impl rpccore::Server for CallbackServer {
+    type ServerMessage = CallbackReq;
+    type ClientMessage = CallbackResp;
+
+    fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage {
+        match req {
+            CallbackReq::Data {
+                nframes,
+                input_frame_size,
+                output_frame_size,
+            } => {
+                trace!(
+                    "stream_thread: Data Callback: nframes={} input_fs={} output_fs={}",
+                    nframes,
+                    input_frame_size,
+                    output_frame_size,
+                );
+
+                let input_nbytes = nframes as usize * input_frame_size;
+                let output_nbytes = nframes as usize * output_frame_size;
+
+                // Input and output reuse the same shmem backing.  Unfortunately, cubeb's data_callback isn't
+                // specified in such a way that would require the callee to consume all of the input before
+                // writing to the output (i.e., it is passed as two pointers that aren't expected to alias).
+                // That means we need to copy the input here.
+                if let Some(buf) = &mut self.duplex_input {
+                    assert!(input_nbytes > 0);
+                    assert!(buf.capacity() >= input_nbytes);
+                    unsafe {
+                        let input = self.shm.get_slice(input_nbytes).unwrap();
+                        ptr::copy_nonoverlapping(input.as_ptr(), buf.as_mut_ptr(), input.len());
+                    }
+                }
+
+                run_in_callback(|| {
+                    let nframes = unsafe {
+                        let input_ptr = if input_frame_size > 0 {
+                            if let Some(buf) = &mut self.duplex_input {
+                                buf.as_ptr()
+                            } else {
+                                self.shm.get_slice(input_nbytes).unwrap().as_ptr()
+                            }
+                        } else {
+                            ptr::null()
+                        };
+                        let output_ptr = if output_frame_size > 0 {
+                            self.shm.get_mut_slice(output_nbytes).unwrap().as_mut_ptr()
+                        } else {
+                            ptr::null_mut()
+                        };
+
+                        self.data_cb.unwrap()(
+                            ptr::null_mut(), // https://github.com/kinetiknz/cubeb/issues/518
+                            self.user_ptr as *mut c_void,
+                            input_ptr as *const _,
+                            output_ptr as *mut _,
+                            nframes as _,
+                        )
+                    };
+
+                    CallbackResp::Data(nframes as isize)
+                })
+            }
+            CallbackReq::State(state) => {
+                trace!("stream_thread: State Callback: {:?}", state);
+                run_in_callback(|| unsafe {
+                    self.state_cb.unwrap()(ptr::null_mut(), self.user_ptr as *mut _, state);
+                });
+
+                CallbackResp::State
+            }
+            CallbackReq::DeviceChange => {
+                run_in_callback(|| {
+                    let cb = *self.device_change_cb.lock().unwrap();
+                    if let Some(cb) = cb {
+                        unsafe {
+                            cb(self.user_ptr as *mut _);
+                        }
+                    } else {
+                        warn!("DeviceChange received with null callback");
+                    }
+                });
+
+                CallbackResp::DeviceChange
+            }
+        }
+    }
+}
+
+impl<'ctx> ClientStream<'ctx> {
+    fn init(
+        ctx: &'ctx ClientContext,
+        init_params: messages::StreamInitParams,
+        data_callback: ffi::cubeb_data_callback,
+        state_callback: ffi::cubeb_state_callback,
+        user_ptr: *mut c_void,
+    ) -> Result<Stream> {
+        assert_not_in_callback();
+
+        let rpc = ctx.rpc();
+        let create_params = StreamCreateParams {
+            input_stream_params: init_params.input_stream_params,
+            output_stream_params: init_params.output_stream_params,
+        };
+        let mut data = send_recv!(rpc, StreamCreate(create_params) => StreamCreated())?;
+
+        debug!(
+            "token = {}, handle = {:?} area_size = {:?}",
+            data.token, data.shm_handle, data.shm_area_size
+        );
+
+        let shm =
+            match unsafe { SharedMem::from(data.shm_handle.take_handle(), data.shm_area_size) } {
+                Ok(shm) => shm,
+                Err(e) => {
+                    warn!(
+                        "SharedMem client mapping failed (size={}, err={:?})",
+                        data.shm_area_size, e
+                    );
+                    return Err(Error::default());
+                }
+            };
+
+        let duplex_input = if let (Some(_), Some(_)) = (
+            init_params.input_stream_params,
+            init_params.output_stream_params,
+        ) {
+            let mut duplex_input = Vec::new();
+            match duplex_input.try_reserve_exact(data.shm_area_size) {
+                Ok(()) => Some(duplex_input),
+                Err(e) => {
+                    warn!(
+                        "duplex_input allocation failed (size={}, err={:?})",
+                        data.shm_area_size, e
+                    );
+                    return Err(Error::default());
+                }
+            }
+        } else {
+            None
+        };
+
+        let mut stream =
+            send_recv!(rpc, StreamInit(data.token, init_params) => StreamInitialized())?;
+        let stream = unsafe { sys::Pipe::from_raw_handle(stream.take_handle()) };
+
+        let user_data = user_ptr as usize;
+
+        let null_cb: ffi::cubeb_device_changed_callback = None;
+        let device_change_cb = Arc::new(Mutex::new(null_cb));
+
+        let (_shutdown_tx, shutdown_rx) = mpsc::channel();
+
+        let server = CallbackServer {
+            shm,
+            duplex_input,
+            data_cb: data_callback,
+            state_cb: state_callback,
+            user_ptr: user_data,
+            device_change_cb: device_change_cb.clone(),
+            _shutdown_tx,
+        };
+
+        ctx.callback_handle()
+            .bind_server(server, stream)
+            .map_err(|_| Error::default())?;
+
+        let stream = Box::into_raw(Box::new(ClientStream {
+            context: ctx,
+            user_ptr,
+            token: data.token,
+            device_change_cb,
+            shutdown_rx,
+        }));
+        Ok(unsafe { Stream::from_ptr(stream as *mut _) })
+    }
+}
+
+impl Drop for ClientStream<'_> {
+    fn drop(&mut self) {
+        debug!("ClientStream drop");
+        let _ = send_recv!(self.context.rpc(), StreamDestroy(self.token) => StreamDestroyed);
+        debug!("ClientStream drop - stream destroyed");
+        // Wait for CallbackServer to shutdown.  The remote server drops the RPC
+        // connection during StreamDestroy, which will cause CallbackServer to drop
+        // once the connection close is detected.  Dropping CallbackServer will
+        // cause the shutdown channel to error on recv, which we rely on to
+        // synchronize with CallbackServer dropping.
+        let _ = self.shutdown_rx.recv();
+        debug!("ClientStream dropped");
+    }
+}
+
+impl StreamOps for ClientStream<'_> {
+    fn start(&mut self) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamStart(self.token) => StreamStarted)
+    }
+
+    fn stop(&mut self) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamStop(self.token) => StreamStopped)
+    }
+
+    fn position(&mut self) -> Result<u64> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetPosition(self.token) => StreamPosition())
+    }
+
+    fn latency(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetLatency(self.token) => StreamLatency())
+    }
+
+    fn input_latency(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetInputLatency(self.token) => StreamInputLatency())
+    }
+
+    fn set_volume(&mut self, volume: f32) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamSetVolume(self.token, volume) => StreamVolumeSet)
+    }
+
+    fn set_name(&mut self, name: &CStr) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamSetName(self.token, name.to_owned()) => StreamNameSet)
+    }
+
+    fn current_device(&mut self) -> Result<&DeviceRef> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        match send_recv!(rpc, StreamGetCurrentDevice(self.token) => StreamCurrentDevice()) {
+            Ok(d) => Ok(unsafe { DeviceRef::from_ptr(Box::into_raw(Box::new(d.into()))) }),
+            Err(e) => Err(e),
+        }
+    }
+
+    fn device_destroy(&mut self, device: &DeviceRef) -> Result<()> {
+        assert_not_in_callback();
+        if device.as_ptr().is_null() {
+            Err(Error::error())
+        } else {
+            unsafe {
+                let _: Box<Device> = Box::from_raw(device.as_ptr() as *mut _);
+            }
+            Ok(())
+        }
+    }
+
+    fn register_device_changed_callback(
+        &mut self,
+        device_changed_callback: ffi::cubeb_device_changed_callback,
+    ) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        let enable = device_changed_callback.is_some();
+        *self.device_change_cb.lock().unwrap() = device_changed_callback;
+        send_recv!(rpc, StreamRegisterDeviceChangeCallback(self.token, enable) => StreamRegisterDeviceChangeCallback)
+    }
+}
+
+pub fn init(
+    ctx: &ClientContext,
+    init_params: messages::StreamInitParams,
+    data_callback: ffi::cubeb_data_callback,
+    state_callback: ffi::cubeb_state_callback,
+    user_ptr: *mut c_void,
+) -> Result<Stream> {
+    let stm = ClientStream::init(ctx, init_params, data_callback, state_callback, user_ptr)?;
+    debug_assert_eq!(stm.user_ptr(), user_ptr);
+    Ok(stm)
+}
diff --git a/third_party/rust/audioipc2-server/.cargo-checksum.json b/third_party/rust/audioipc2-server/.cargo-checksum.json
new file mode 100644
index 0000000000..733e618bb0
--- /dev/null
+++ b/third_party/rust/audioipc2-server/.cargo-checksum.json
@@ -0,0 +1 @@
+{"files":{"Cargo.toml":"4b00cd5b0ef67640c29a5b4c27ff0eb2bda60ea9bfaa0321625e2eb48714efa8","cbindgen.toml":"fb6abe1671497f432a06e40b1db7ed7cd2cceecbd9a2382193ad7534e8855e34","src/lib.rs":"06aff4fd1326aeabb16b01f81a6f3c59c1717ebe96285a063724830cdf30303a","src/server.rs":"ea839fe4607ba6b70a1ef5dfb2305eb668b148820c5590b87192609fbe3c9edd"},"package":null}
\ No newline at end of file
diff --git a/third_party/rust/audioipc2-server/Cargo.toml b/third_party/rust/audioipc2-server/Cargo.toml
new file mode 100644
index 0000000000..dbe87893a6
--- /dev/null
+++ b/third_party/rust/audioipc2-server/Cargo.toml
@@ -0,0 +1,40 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2018"
+name = "audioipc2-server"
+version = "0.5.0"
+authors = [
+    "Matthew Gregan <kinetik@flim.org>",
+    "Dan Glastonbury <dan.glastonbury@gmail.com>",
+]
+description = "Remote cubeb server"
+license = "ISC"
+
+[dependencies]
+cubeb-core = "0.10.0"
+log = "0.4"
+once_cell = "1.2.0"
+slab = "0.4"
+
+[dependencies.audio_thread_priority]
+version = "0.26.1"
+features = ["winapi"]
+default-features = false
+
+[dependencies.audioipc]
+path = "../audioipc"
+package = "audioipc2"
+
+[dependencies.error-chain]
+version = "0.12.0"
+default-features = false
diff --git a/third_party/rust/audioipc2-server/cbindgen.toml b/third_party/rust/audioipc2-server/cbindgen.toml
new file mode 100644
index 0000000000..b3267889c6
--- /dev/null
+++ b/third_party/rust/audioipc2-server/cbindgen.toml
@@ -0,0 +1,28 @@
+header = """/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */"""
+autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */"""
+include_version = true
+braces = "SameLine"
+line_length = 100
+tab_width = 2
+language = "C++"
+namespaces = ["mozilla", "audioipc2"]
+
+[export]
+item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"]
+
+[parse]
+parse_deps = true
+include = ["audioipc2"]
+
+[fn]
+args = "Vertical"
+rename_args = "GeckoCase"
+
+[struct]
+rename_fields = "GeckoCase"
+
+[defines]
+"windows" = "XP_WIN"
+"unix" = "XP_UNIX"
diff --git a/third_party/rust/audioipc2-server/src/lib.rs b/third_party/rust/audioipc2-server/src/lib.rs
new file mode 100644
index 0000000000..c90de38c66
--- /dev/null
+++ b/third_party/rust/audioipc2-server/src/lib.rs
@@ -0,0 +1,221 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+#![warn(unused_extern_crates)]
+
+#[macro_use]
+extern crate error_chain;
+#[macro_use]
+extern crate log;
+
+use audio_thread_priority::promote_current_thread_to_real_time;
+use audioipc::ipccore;
+use audioipc::sys;
+use audioipc::PlatformHandleType;
+use once_cell::sync::Lazy;
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::ptr;
+use std::sync::Mutex;
+use std::thread;
+
+mod server;
+
+struct CubebContextParams {
+    context_name: CString,
+    backend_name: Option<CString>,
+}
+
+static G_CUBEB_CONTEXT_PARAMS: Lazy<Mutex<CubebContextParams>> = Lazy::new(|| {
+    Mutex::new(CubebContextParams {
+        context_name: CString::new("AudioIPC Server").unwrap(),
+        backend_name: None,
+    })
+});
+
+#[allow(deprecated)]
+pub mod errors {
+    #![allow(clippy::upper_case_acronyms)]
+    error_chain! {
+        links {
+            AudioIPC(::audioipc::errors::Error, ::audioipc::errors::ErrorKind);
+        }
+        foreign_links {
+            Cubeb(cubeb_core::Error);
+            Io(::std::io::Error);
+        }
+    }
+}
+
+use crate::errors::*;
+
+struct ServerWrapper {
+    rpc_thread: ipccore::EventLoopThread,
+    callback_thread: ipccore::EventLoopThread,
+    device_collection_thread: ipccore::EventLoopThread,
+}
+
+fn register_thread(callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>) {
+    if let Some(func) = callback {
+        let name = CString::new(thread::current().name().unwrap()).unwrap();
+        func(name.as_ptr());
+    }
+}
+
+fn unregister_thread(callback: Option<extern "C" fn()>) {
+    if let Some(func) = callback {
+        func();
+    }
+}
+
+fn init_threads(
+    thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+    thread_destroy_callback: Option<extern "C" fn()>,
+) -> Result<ServerWrapper> {
+    let rpc_name = "AudioIPC Server RPC";
+    let rpc_thread = ipccore::EventLoopThread::new(
+        rpc_name.to_string(),
+        None,
+        move || {
+            trace!("Starting {} thread", rpc_name);
+            register_thread(thread_create_callback);
+            audioipc::server_platform_init();
+        },
+        move || {
+            unregister_thread(thread_destroy_callback);
+            trace!("Stopping {} thread", rpc_name);
+        },
+    )
+    .map_err(|e| {
+        debug!("Failed to start {} thread: {:?}", rpc_name, e);
+        e
+    })?;
+
+    let callback_name = "AudioIPC Server Callback";
+    let callback_thread = ipccore::EventLoopThread::new(
+        callback_name.to_string(),
+        None,
+        move || {
+            trace!("Starting {} thread", callback_name);
+            if let Err(e) = promote_current_thread_to_real_time(0, 48000) {
+                debug!(
+                    "Failed to promote {} thread to real-time: {:?}",
+                    callback_name, e
+                );
+            }
+            register_thread(thread_create_callback);
+        },
+        move || {
+            unregister_thread(thread_destroy_callback);
+            trace!("Stopping {} thread", callback_name);
+        },
+    )
+    .map_err(|e| {
+        debug!("Failed to start {} thread: {:?}", callback_name, e);
+        e
+    })?;
+
+    let device_collection_name = "AudioIPC DeviceCollection RPC";
+    let device_collection_thread = ipccore::EventLoopThread::new(
+        device_collection_name.to_string(),
+        None,
+        move || {
+            trace!("Starting {} thread", device_collection_name);
+            register_thread(thread_create_callback);
+        },
+        move || {
+            unregister_thread(thread_destroy_callback);
+            trace!("Stopping {} thread", device_collection_name);
+        },
+    )
+    .map_err(|e| {
+        debug!("Failed to start {} thread: {:?}", device_collection_name, e);
+        e
+    })?;
+
+    Ok(ServerWrapper {
+        rpc_thread,
+        callback_thread,
+        device_collection_thread,
+    })
+}
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug)]
+pub struct AudioIpcServerInitParams {
+    // Fields only need to be public for ipctest.
+    pub thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+    pub thread_destroy_callback: Option<extern "C" fn()>,
+}
+
+#[allow(clippy::missing_safety_doc)]
+#[no_mangle]
+pub unsafe extern "C" fn audioipc2_server_start(
+    context_name: *const std::os::raw::c_char,
+    backend_name: *const std::os::raw::c_char,
+    init_params: *const AudioIpcServerInitParams,
+) -> *mut c_void {
+    assert!(!init_params.is_null());
+    let mut params = G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
+    if !context_name.is_null() {
+        params.context_name = CStr::from_ptr(context_name).to_owned();
+    }
+    if !backend_name.is_null() {
+        let backend_string = CStr::from_ptr(backend_name).to_owned();
+        params.backend_name = Some(backend_string);
+    }
+    match init_threads(
+        (*init_params).thread_create_callback,
+        (*init_params).thread_destroy_callback,
+    ) {
+        Ok(server) => Box::into_raw(Box::new(server)) as *mut _,
+        Err(_) => ptr::null_mut() as *mut _,
+    }
+}
+
+// A `shm_area_size` of 0 allows the server to calculate an appropriate shm size for each stream.
+// A non-zero `shm_area_size` forces all allocations to the specified size.
+#[no_mangle]
+pub extern "C" fn audioipc2_server_new_client(
+    p: *mut c_void,
+    shm_area_size: usize,
+) -> PlatformHandleType {
+    let wrapper: &ServerWrapper = unsafe { &*(p as *mut _) };
+
+    // We create a connected pair of anonymous IPC endpoints. One side
+    // is registered with the event loop core, the other side is returned
+    // to the caller to be remoted to the client to complete setup.
+    let (server_pipe, client_pipe) = match sys::make_pipe_pair() {
+        Ok((server_pipe, client_pipe)) => (server_pipe, client_pipe),
+        Err(e) => {
+            error!(
+                "audioipc_server_new_client - make_pipe_pair failed: {:?}",
+                e
+            );
+            return audioipc::INVALID_HANDLE_VALUE;
+        }
+    };
+
+    let rpc_thread = wrapper.rpc_thread.handle();
+    let callback_thread = wrapper.callback_thread.handle();
+    let device_collection_thread = wrapper.device_collection_thread.handle();
+
+    let server = server::CubebServer::new(
+        callback_thread.clone(),
+        device_collection_thread.clone(),
+        shm_area_size,
+    );
+    if let Err(e) = rpc_thread.bind_server(server, server_pipe) {
+        error!("audioipc_server_new_client - bind_server failed: {:?}", e);
+        return audioipc::INVALID_HANDLE_VALUE;
+    }
+
+    unsafe { client_pipe.into_raw() }
+}
+
+#[no_mangle]
+pub extern "C" fn audioipc2_server_stop(p: *mut c_void) {
+    let wrapper = unsafe { Box::<ServerWrapper>::from_raw(p as *mut _) };
+    drop(wrapper);
+}
diff --git a/third_party/rust/audioipc2-server/src/server.rs b/third_party/rust/audioipc2-server/src/server.rs
new file mode 100644
index 0000000000..0094a17681
--- /dev/null
+++ b/third_party/rust/audioipc2-server/src/server.rs
@@ -0,0 +1,921 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+#[cfg(target_os = "linux")]
+use audio_thread_priority::{promote_thread_to_real_time, RtPriorityThreadInfo};
+use audioipc::messages::SerializableHandle;
+use audioipc::messages::{
+    CallbackReq, CallbackResp, ClientMessage, Device, DeviceCollectionReq, DeviceCollectionResp,
+    DeviceInfo, RegisterDeviceCollectionChanged, ServerMessage, StreamCreate, StreamCreateParams,
+    StreamInitParams, StreamParams,
+};
+use audioipc::shm::SharedMem;
+use audioipc::{ipccore, rpccore, sys, PlatformHandle};
+use cubeb_core as cubeb;
+use cubeb_core::ffi;
+use std::convert::{From, TryInto};
+use std::ffi::CStr;
+use std::mem::size_of;
+use std::os::raw::{c_long, c_void};
+use std::rc::Rc;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::{cell::RefCell, sync::Mutex};
+use std::{panic, slice};
+
+use crate::errors::*;
+
+fn error(error: cubeb::Error) -> ClientMessage {
+    ClientMessage::Error(error.raw_code())
+}
+
+struct CubebDeviceCollectionManager {
+    servers: Mutex<Vec<(Rc<DeviceCollectionChangeCallback>, cubeb::DeviceType)>>,
+}
+
+impl CubebDeviceCollectionManager {
+    fn new() -> CubebDeviceCollectionManager {
+        CubebDeviceCollectionManager {
+            servers: Mutex::new(Vec::new()),
+        }
+    }
+
+    fn register(
+        &self,
+        context: &cubeb::Context,
+        server: &Rc<DeviceCollectionChangeCallback>,
+        devtype: cubeb::DeviceType,
+    ) -> cubeb::Result<()> {
+        let mut servers = self.servers.lock().unwrap();
+        if servers.is_empty() {
+            self.internal_register(context, true)?;
+        }
+        servers.push((server.clone(), devtype));
+        Ok(())
+    }
+
+    fn unregister(
+        &self,
+        context: &cubeb::Context,
+        server: &Rc<DeviceCollectionChangeCallback>,
+        devtype: cubeb::DeviceType,
+    ) -> cubeb::Result<()> {
+        let mut servers = self.servers.lock().unwrap();
+        servers.retain(|(s, d)| !Rc::ptr_eq(s, server) || d != &devtype);
+        if servers.is_empty() {
+            self.internal_register(context, false)?;
+        }
+        Ok(())
+    }
+
+    fn internal_register(&self, context: &cubeb::Context, enable: bool) -> cubeb::Result<()> {
+        for &(dir, cb) in &[
+            (
+                cubeb::DeviceType::INPUT,
+                device_collection_changed_input_cb_c as _,
+            ),
+            (
+                cubeb::DeviceType::OUTPUT,
+                device_collection_changed_output_cb_c as _,
+            ),
+        ] {
+            unsafe {
+                context.register_device_collection_changed(
+                    dir,
+                    if enable { Some(cb) } else { None },
+                    if enable {
+                        self as *const CubebDeviceCollectionManager as *mut c_void
+                    } else {
+                        std::ptr::null_mut()
+                    },
+                )?;
+            }
+        }
+        Ok(())
+    }
+
+    unsafe fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) {
+        let servers = self.servers.lock().unwrap();
+        servers.iter().for_each(|(s, d)| {
+            if d.contains(cubeb::DeviceType::from_bits_truncate(device_type)) {
+                s.device_collection_changed_callback(device_type)
+            }
+        });
+    }
+}
+
+impl Drop for CubebDeviceCollectionManager {
+    fn drop(&mut self) {
+        assert!(self.servers.lock().unwrap().is_empty());
+    }
+}
+
+struct DevIdMap {
+    devices: Vec<usize>,
+}
+
+// A cubeb_devid is an opaque type which may be implemented with a stable
+// pointer in a cubeb backend.  cubeb_devids received remotely must be
+// validated before use, so DevIdMap provides a simple 1:1 mapping between a
+// cubeb_devid and an IPC-transportable value suitable for use as a unique
+// handle.
+impl DevIdMap {
+    fn new() -> DevIdMap {
+        let mut d = DevIdMap {
+            devices: Vec::with_capacity(32),
+        };
+        // A null cubeb_devid is used for selecting the default device.
+        // Pre-populate the mapping with 0 -> 0 to handle nulls.
+        d.devices.push(0);
+        d
+    }
+
+    // Given a cubeb_devid, return a unique stable value suitable for use
+    // over IPC.
+    fn make_handle(&mut self, devid: usize) -> usize {
+        if let Some(i) = self.devices.iter().position(|&d| d == devid) {
+            return i;
+        }
+        self.devices.push(devid);
+        self.devices.len() - 1
+    }
+
+    // Given a handle produced by `make_handle`, return the associated
+    // cubeb_devid.  Invalid handles result in a panic.
+    fn handle_to_id(&self, handle: usize) -> usize {
+        self.devices[handle]
+    }
+}
+
+struct CubebContextState {
+    // `manager` must be dropped before the `context` is destroyed.
+    manager: CubebDeviceCollectionManager,
+    context: cubeb::Result<cubeb::Context>,
+}
+
+thread_local!(static CONTEXT_KEY: RefCell<Option<CubebContextState>> = RefCell::new(None));
+
+fn cubeb_init_from_context_params() -> cubeb::Result<cubeb::Context> {
+    let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
+    let context_name = Some(params.context_name.as_c_str());
+    let backend_name = params.backend_name.as_deref();
+    let r = cubeb::Context::init(context_name, backend_name);
+    r.map_err(|e| {
+        info!("cubeb::Context::init failed r={:?}", e);
+        e
+    })
+}
+
+fn with_local_context<T, F>(f: F) -> T
+where
+    F: FnOnce(&cubeb::Result<cubeb::Context>, &mut CubebDeviceCollectionManager) -> T,
+{
+    CONTEXT_KEY.with(|k| {
+        let mut state = k.borrow_mut();
+        if state.is_none() {
+            *state = Some(CubebContextState {
+                manager: CubebDeviceCollectionManager::new(),
+                context: cubeb_init_from_context_params(),
+            });
+        }
+        let CubebContextState { manager, context } = state.as_mut().unwrap();
+        // Always reattempt to initialize cubeb, OS config may have changed.
+        if context.is_err() {
+            *context = cubeb_init_from_context_params();
+        }
+        f(context, manager)
+    })
+}
+
+struct DeviceCollectionClient;
+
+impl rpccore::Client for DeviceCollectionClient {
+    type ServerMessage = DeviceCollectionReq;
+    type ClientMessage = DeviceCollectionResp;
+}
+
+struct CallbackClient;
+
+impl rpccore::Client for CallbackClient {
+    type ServerMessage = CallbackReq;
+    type ClientMessage = CallbackResp;
+}
+
+struct ServerStreamCallbacks {
+    /// Size of input frame in bytes
+    input_frame_size: u16,
+    /// Size of output frame in bytes
+    output_frame_size: u16,
+    /// Shared memory buffer for transporting audio data to/from client
+    shm: SharedMem,
+    /// RPC interface for data_callback (on OS audio thread) to server callback thread
+    data_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>,
+    /// RPC interface for state_callback (on any thread) to server callback thread
+    state_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>,
+    /// RPC interface for device_change_callback (on any thread) to server callback thread
+    device_change_callback_rpc: rpccore::Proxy<CallbackReq, CallbackResp>,
+    /// Indicates stream is connected to client side.  Callbacks received before
+    /// the stream is in the connected state cannot be sent to the client side, so
+    /// are logged and otherwise ignored.
+    connected: AtomicBool,
+}
+
+impl ServerStreamCallbacks {
+    fn data_callback(&mut self, input: &[u8], output: &mut [u8], nframes: isize) -> isize {
+        trace!(
+            "Stream data callback: {} {} {}",
+            nframes,
+            input.len(),
+            output.len()
+        );
+        if !self.connected.load(Ordering::Acquire) {
+            warn!("Stream data callback triggered before stream connected");
+            return cubeb::ffi::CUBEB_ERROR.try_into().unwrap();
+        }
+
+        if self.input_frame_size != 0 {
+            if input.len() > self.shm.get_size() {
+                debug!(
+                    "bad input size: input={} shm={}",
+                    input.len(),
+                    self.shm.get_size()
+                );
+                return cubeb::ffi::CUBEB_ERROR.try_into().unwrap();
+            }
+            unsafe {
+                self.shm
+                    .get_mut_slice(input.len())
+                    .unwrap()
+                    .copy_from_slice(input);
+            }
+        }
+
+        if self.output_frame_size != 0 && output.len() > self.shm.get_size() {
+            debug!(
+                "bad output size: output={} shm={}",
+                output.len(),
+                self.shm.get_size()
+            );
+            return cubeb::ffi::CUBEB_ERROR.try_into().unwrap();
+        }
+
+        let r = self.data_callback_rpc.call(CallbackReq::Data {
+            nframes,
+            input_frame_size: self.input_frame_size as usize,
+            output_frame_size: self.output_frame_size as usize,
+        });
+
+        match r {
+            Ok(CallbackResp::Data(frames)) => {
+                if frames >= 0 && self.output_frame_size != 0 {
+                    let nbytes = frames as usize * self.output_frame_size as usize;
+                    unsafe {
+                        output[..nbytes].copy_from_slice(self.shm.get_slice(nbytes).unwrap());
+                    }
+                }
+                frames
+            }
+            _ => {
+                debug!("Unexpected message {:?} during data_callback", r);
+                cubeb::ffi::CUBEB_ERROR.try_into().unwrap()
+            }
+        }
+    }
+
+    fn state_callback(&self, state: cubeb::State) {
+        trace!("Stream state callback: {:?}", state);
+        if !self.connected.load(Ordering::Acquire) {
+            warn!("Stream state callback triggered before stream connected");
+            return;
+        }
+
+        let r = self
+            .state_callback_rpc
+            .call(CallbackReq::State(state.into()));
+        match r {
+            Ok(CallbackResp::State) => {}
+            _ => {
+                debug!("Unexpected message {:?} during state callback", r);
+            }
+        }
+    }
+
+    fn device_change_callback(&self) {
+        trace!("Stream device change callback");
+        if !self.connected.load(Ordering::Acquire) {
+            warn!("Stream device_change callback triggered before stream connected");
+            return;
+        }
+        let r = self
+            .device_change_callback_rpc
+            .call(CallbackReq::DeviceChange);
+        match r {
+            Ok(CallbackResp::DeviceChange) => {}
+            _ => {
+                debug!("Unexpected message {:?} during device change callback", r);
+            }
+        }
+    }
+}
+
+static SHM_ID: AtomicUsize = AtomicUsize::new(0);
+
+// Generate a temporary shm_id fragment that is unique to the process.  This
+// path is used temporarily to create a shm segment, which is then
+// immediately deleted from the filesystem while retaining handles to the
+// shm to be shared between the server and client.
+fn get_shm_id() -> String {
+    format!(
+        "cubeb-shm-{}-{}",
+        std::process::id(),
+        SHM_ID.fetch_add(1, Ordering::SeqCst)
+    )
+}
+
+struct ServerStream {
+    stream: Option<cubeb::Stream>,
+    cbs: Box<ServerStreamCallbacks>,
+    client_pipe: Option<PlatformHandle>,
+}
+
+impl Drop for ServerStream {
+    fn drop(&mut self) {
+        // `stream` *must* be dropped before `cbs`.
+        drop(self.stream.take());
+    }
+}
+
+struct DeviceCollectionChangeCallback {
+    rpc: rpccore::Proxy<DeviceCollectionReq, DeviceCollectionResp>,
+}
+
+impl DeviceCollectionChangeCallback {
+    fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) {
+        // TODO: Assert device_type is in devtype.
+        debug!(
+            "Sending device collection ({:?}) changed event",
+            device_type
+        );
+        let _ = self
+            .rpc
+            .call(DeviceCollectionReq::DeviceChange(device_type));
+    }
+}
+
+pub struct CubebServer {
+    callback_thread: ipccore::EventLoopHandle,
+    device_collection_thread: ipccore::EventLoopHandle,
+    streams: slab::Slab<ServerStream>,
+    remote_pid: Option<u32>,
+    device_collection_change_callbacks: Option<Rc<DeviceCollectionChangeCallback>>,
+    devidmap: DevIdMap,
+    shm_area_size: usize,
+}
+
+impl Drop for CubebServer {
+    fn drop(&mut self) {
+        if let Some(device_collection_change_callbacks) = &self.device_collection_change_callbacks {
+            debug!("CubebServer: dropped with device_collection_change_callbacks registered");
+            CONTEXT_KEY.with(|k| {
+                let mut state = k.borrow_mut();
+                if let Some(CubebContextState {
+                    manager,
+                    context: Ok(context),
+                }) = state.as_mut()
+                {
+                    for devtype in [cubeb::DeviceType::INPUT, cubeb::DeviceType::OUTPUT] {
+                        let r = manager.unregister(
+                            context,
+                            device_collection_change_callbacks,
+                            devtype,
+                        );
+                        if r.is_err() {
+                            debug!("CubebServer: unregister failed: {:?}", r);
+                        }
+                    }
+                }
+            })
+        }
+    }
+}
+
+#[allow(unknown_lints)] // non_send_fields_in_send_ty is Nightly-only as of 2021-11-29.
+#[allow(clippy::non_send_fields_in_send_ty)]
+// XXX: required for server setup, verify this is safe.
+unsafe impl Send for CubebServer {}
+
+impl rpccore::Server for CubebServer {
+    type ServerMessage = ServerMessage;
+    type ClientMessage = ClientMessage;
+
+    fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage {
+        if let ServerMessage::ClientConnect(pid) = req {
+            self.remote_pid = Some(pid);
+        }
+        with_local_context(|context, manager| match *context {
+            Err(_) => error(cubeb::Error::error()),
+            Ok(ref context) => self.process_msg(context, manager, &req),
+        })
+    }
+}
+
+// Debugging for BMO 1594216/1612044.
+macro_rules! try_stream {
+    ($self:expr, $stm_tok:expr) => {
+        if $self.streams.contains($stm_tok) {
+            $self.streams[$stm_tok]
+                .stream
+                .as_mut()
+                .expect("uninitialized stream")
+        } else {
+            error!(
+                "{}:{}:{} - Stream({}): invalid token",
+                file!(),
+                line!(),
+                column!(),
+                $stm_tok
+            );
+            return error(cubeb::Error::invalid_parameter());
+        }
+    };
+}
+
+impl CubebServer {
+    pub fn new(
+        callback_thread: ipccore::EventLoopHandle,
+        device_collection_thread: ipccore::EventLoopHandle,
+        shm_area_size: usize,
+    ) -> Self {
+        CubebServer {
+            callback_thread,
+            device_collection_thread,
+            streams: slab::Slab::<ServerStream>::new(),
+            remote_pid: None,
+            device_collection_change_callbacks: None,
+            devidmap: DevIdMap::new(),
+            shm_area_size,
+        }
+    }
+
+    // Process a request coming from the client.
+    fn process_msg(
+        &mut self,
+        context: &cubeb::Context,
+        manager: &mut CubebDeviceCollectionManager,
+        msg: &ServerMessage,
+    ) -> ClientMessage {
+        let resp: ClientMessage = match *msg {
+            ServerMessage::ClientConnect(_) => {
+                // remote_pid is set before cubeb initialization, just verify here.
+                assert!(self.remote_pid.is_some());
+                ClientMessage::ClientConnected
+            }
+
+            ServerMessage::ClientDisconnect => {
+                // TODO:
+                //self.connection.client_disconnect();
+                ClientMessage::ClientDisconnected
+            }
+
+            ServerMessage::ContextGetBackendId => {
+                ClientMessage::ContextBackendId(context.backend_id().to_string())
+            }
+
+            ServerMessage::ContextGetMaxChannelCount => context
+                .max_channel_count()
+                .map(ClientMessage::ContextMaxChannelCount)
+                .unwrap_or_else(error),
+
+            ServerMessage::ContextGetMinLatency(ref params) => {
+                let format = cubeb::SampleFormat::from(params.format);
+                let layout = cubeb::ChannelLayout::from(params.layout);
+
+                let params = cubeb::StreamParamsBuilder::new()
+                    .format(format)
+                    .rate(params.rate)
+                    .channels(params.channels)
+                    .layout(layout)
+                    .take();
+
+                context
+                    .min_latency(&params)
+                    .map(ClientMessage::ContextMinLatency)
+                    .unwrap_or_else(error)
+            }
+
+            ServerMessage::ContextGetPreferredSampleRate => context
+                .preferred_sample_rate()
+                .map(ClientMessage::ContextPreferredSampleRate)
+                .unwrap_or_else(error),
+
+            ServerMessage::ContextGetDeviceEnumeration(device_type) => context
+                .enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type))
+                .map(|devices| {
+                    let v: Vec<DeviceInfo> = devices
+                        .iter()
+                        .map(|i| {
+                            let mut tmp: DeviceInfo = i.as_ref().into();
+                            // Replace each cubeb_devid with a unique handle suitable for IPC.
+                            tmp.devid = self.devidmap.make_handle(tmp.devid);
+                            tmp
+                        })
+                        .collect();
+                    ClientMessage::ContextEnumeratedDevices(v)
+                })
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamCreate(ref params) => self
+                .process_stream_create(params)
+                .unwrap_or_else(|_| error(cubeb::Error::error())),
+
+            ServerMessage::StreamInit(stm_tok, ref params) => self
+                .process_stream_init(context, stm_tok, params)
+                .unwrap_or_else(|_| error(cubeb::Error::error())),
+
+            ServerMessage::StreamDestroy(stm_tok) => {
+                if self.streams.contains(stm_tok) {
+                    debug!("Unregistering stream {:?}", stm_tok);
+                    self.streams.remove(stm_tok);
+                } else {
+                    // Debugging for BMO 1594216/1612044.
+                    error!("StreamDestroy({}): invalid token", stm_tok);
+                    return error(cubeb::Error::invalid_parameter());
+                }
+                ClientMessage::StreamDestroyed
+            }
+
+            ServerMessage::StreamStart(stm_tok) => try_stream!(self, stm_tok)
+                .start()
+                .map(|_| ClientMessage::StreamStarted)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamStop(stm_tok) => try_stream!(self, stm_tok)
+                .stop()
+                .map(|_| ClientMessage::StreamStopped)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetPosition(stm_tok) => try_stream!(self, stm_tok)
+                .position()
+                .map(ClientMessage::StreamPosition)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetLatency(stm_tok) => try_stream!(self, stm_tok)
+                .latency()
+                .map(ClientMessage::StreamLatency)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetInputLatency(stm_tok) => try_stream!(self, stm_tok)
+                .input_latency()
+                .map(ClientMessage::StreamInputLatency)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamSetVolume(stm_tok, volume) => try_stream!(self, stm_tok)
+                .set_volume(volume)
+                .map(|_| ClientMessage::StreamVolumeSet)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamSetName(stm_tok, ref name) => try_stream!(self, stm_tok)
+                .set_name(name)
+                .map(|_| ClientMessage::StreamNameSet)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetCurrentDevice(stm_tok) => try_stream!(self, stm_tok)
+                .current_device()
+                .map(|device| ClientMessage::StreamCurrentDevice(Device::from(device)))
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamRegisterDeviceChangeCallback(stm_tok, enable) => {
+                try_stream!(self, stm_tok)
+                    .register_device_changed_callback(if enable {
+                        Some(device_change_cb_c)
+                    } else {
+                        None
+                    })
+                    .map(|_| ClientMessage::StreamRegisterDeviceChangeCallback)
+                    .unwrap_or_else(error)
+            }
+
+            ServerMessage::ContextSetupDeviceCollectionCallback => {
+                let (server_pipe, client_pipe) = match sys::make_pipe_pair() {
+                    Ok((server_pipe, client_pipe)) => (server_pipe, client_pipe),
+                    Err(e) => {
+                        debug!(
+                            "ContextSetupDeviceCollectionCallback - make_pipe_pair failed: {:?}",
+                            e
+                        );
+                        return error(cubeb::Error::error());
+                    }
+                };
+
+                // TODO: this should bind the client_pipe and send the server_pipe to the remote, but
+                //       additional work is required as it's not possible to convert a Windows sys::Pipe into a raw handle.
+                // TODO: Use the rpc_thread instead of an extra device_collection_thread, but a reentrant bind_client
+                //       is required to support that.
+                let rpc = match self
+                    .device_collection_thread
+                    .bind_client::<DeviceCollectionClient>(server_pipe)
+                {
+                    Ok(rpc) => rpc,
+                    Err(e) => {
+                        debug!(
+                            "ContextSetupDeviceCollectionCallback - bind_client: {:?}",
+                            e
+                        );
+                        return error(cubeb::Error::error());
+                    }
+                };
+
+                self.device_collection_change_callbacks =
+                    Some(Rc::new(DeviceCollectionChangeCallback { rpc }));
+                let fd = RegisterDeviceCollectionChanged {
+                    platform_handle: SerializableHandle::new(client_pipe, self.remote_pid.unwrap()),
+                };
+
+                ClientMessage::ContextSetupDeviceCollectionCallback(fd)
+            }
+
+            ServerMessage::ContextRegisterDeviceCollectionChanged(device_type, enable) => self
+                .process_register_device_collection_changed(
+                    context,
+                    manager,
+                    cubeb::DeviceType::from_bits_truncate(device_type),
+                    enable,
+                )
+                .unwrap_or_else(error),
+
+            #[cfg(target_os = "linux")]
+            ServerMessage::PromoteThreadToRealTime(thread_info) => {
+                let info = RtPriorityThreadInfo::deserialize(thread_info);
+                match promote_thread_to_real_time(info, 0, 48000) {
+                    Ok(_) => {
+                        info!("Promotion of content process thread to real-time OK");
+                    }
+                    Err(_) => {
+                        warn!("Promotion of content process thread to real-time error");
+                    }
+                }
+                ClientMessage::ThreadPromoted
+            }
+        };
+
+        trace!("process_msg: req={:?}, resp={:?}", msg, resp);
+
+        resp
+    }
+
+    fn process_register_device_collection_changed(
+        &mut self,
+        context: &cubeb::Context,
+        manager: &mut CubebDeviceCollectionManager,
+        devtype: cubeb::DeviceType,
+        enable: bool,
+    ) -> cubeb::Result<ClientMessage> {
+        if devtype == cubeb::DeviceType::UNKNOWN {
+            return Err(cubeb::Error::invalid_parameter());
+        }
+
+        assert!(self.device_collection_change_callbacks.is_some());
+        let cbs = self.device_collection_change_callbacks.as_ref().unwrap();
+
+        if enable {
+            manager.register(context, cbs, devtype)
+        } else {
+            manager.unregister(context, cbs, devtype)
+        }
+        .map(|_| ClientMessage::ContextRegisteredDeviceCollectionChanged)
+    }
+
+    // Stream create is special, so it's been separated from process_msg.
+    fn process_stream_create(&mut self, params: &StreamCreateParams) -> Result<ClientMessage> {
+        fn frame_size_in_bytes(params: Option<&StreamParams>) -> u16 {
+            params
+                .map(|p| {
+                    let format = p.format.into();
+                    let sample_size = match format {
+                        cubeb::SampleFormat::S16LE
+                        | cubeb::SampleFormat::S16BE
+                        | cubeb::SampleFormat::S16NE => 2,
+                        cubeb::SampleFormat::Float32LE
+                        | cubeb::SampleFormat::Float32BE
+                        | cubeb::SampleFormat::Float32NE => 4,
+                    };
+                    let channel_count = p.channels as u16;
+                    sample_size * channel_count
+                })
+                .unwrap_or(0u16)
+        }
+
+        // Create the callback handling struct which is attached the cubeb stream.
+        let input_frame_size = frame_size_in_bytes(params.input_stream_params.as_ref());
+        let output_frame_size = frame_size_in_bytes(params.output_stream_params.as_ref());
+
+        // Estimate a safe shmem size for this stream configuration.  If the server was configured with a fixed
+        // shm_area_size override, use that instead.
+        // TODO: Add a new cubeb API to query the precise buffer size required for a given stream config.
+        // https://github.com/mozilla/audioipc-2/issues/124
+        let shm_area_size = if self.shm_area_size == 0 {
+            let frame_size = output_frame_size.max(input_frame_size) as u32;
+            let in_rate = params.input_stream_params.map(|p| p.rate).unwrap_or(0);
+            let out_rate = params.output_stream_params.map(|p| p.rate).unwrap_or(0);
+            let rate = out_rate.max(in_rate);
+            // 1s of audio, rounded up to the nearest 64kB.
+            // Stream latency is capped at 1s in process_stream_init.
+            (((rate * frame_size) + 0xffff) & !0xffff) as usize
+        } else {
+            self.shm_area_size
+        };
+        debug!("shm_area_size = {}", shm_area_size);
+
+        let shm = SharedMem::new(&get_shm_id(), shm_area_size)?;
+        let shm_handle = unsafe { shm.make_handle()? };
+
+        let (server_pipe, client_pipe) = sys::make_pipe_pair()?;
+        // TODO: this should bind the client_pipe and send the server_pipe to the remote, but
+        //       additional work is required as it's not possible to convert a Windows sys::Pipe into a raw handle.
+        let rpc = self
+            .callback_thread
+            .bind_client::<CallbackClient>(server_pipe)?;
+
+        let cbs = Box::new(ServerStreamCallbacks {
+            input_frame_size,
+            output_frame_size,
+            shm,
+            state_callback_rpc: rpc.clone(),
+            device_change_callback_rpc: rpc.clone(),
+            data_callback_rpc: rpc,
+            connected: AtomicBool::new(false),
+        });
+
+        let entry = self.streams.vacant_entry();
+        let key = entry.key();
+        debug!("Registering stream {:?}", key);
+
+        entry.insert(ServerStream {
+            stream: None,
+            cbs,
+            client_pipe: Some(client_pipe),
+        });
+
+        Ok(ClientMessage::StreamCreated(StreamCreate {
+            token: key,
+            shm_handle: SerializableHandle::new(shm_handle, self.remote_pid.unwrap()),
+            shm_area_size,
+        }))
+    }
+
+    // Stream init is special, so it's been separated from process_msg.
+    fn process_stream_init(
+        &mut self,
+        context: &cubeb::Context,
+        stm_tok: usize,
+        params: &StreamInitParams,
+    ) -> Result<ClientMessage> {
+        // Create cubeb stream from params
+        let stream_name = params
+            .stream_name
+            .as_ref()
+            .and_then(|name| CStr::from_bytes_with_nul(name).ok());
+
+        // Map IPC handle back to cubeb_devid.
+        let input_device = self.devidmap.handle_to_id(params.input_device) as *const _;
+        let input_stream_params = params.input_stream_params.as_ref().map(|isp| unsafe {
+            cubeb::StreamParamsRef::from_ptr(isp as *const StreamParams as *mut _)
+        });
+
+        // Map IPC handle back to cubeb_devid.
+        let output_device = self.devidmap.handle_to_id(params.output_device) as *const _;
+        let output_stream_params = params.output_stream_params.as_ref().map(|osp| unsafe {
+            cubeb::StreamParamsRef::from_ptr(osp as *const StreamParams as *mut _)
+        });
+
+        // TODO: Manage stream latency requests with respect to the RT deadlines the callback_thread was configured for.
+        fn round_up_pow2(v: u32) -> u32 {
+            debug_assert!(v >= 1);
+            1 << (32 - (v - 1).leading_zeros())
+        }
+        let rate = params
+            .output_stream_params
+            .map(|p| p.rate)
+            .unwrap_or_else(|| params.input_stream_params.map(|p| p.rate).unwrap());
+        // Note: minimum latency supported by AudioIPC is currently ~5ms.  This restriction may be reduced by later IPC improvements.
+        let min_latency = round_up_pow2(5 * rate / 1000);
+        // Note: maximum latency is restricted by the SharedMem size.
+        let max_latency = rate;
+        let latency = params.latency_frames.clamp(min_latency, max_latency);
+        trace!(
+            "stream rate={} latency requested={} calculated={}",
+            rate,
+            params.latency_frames,
+            latency
+        );
+
+        let server_stream = &mut self.streams[stm_tok];
+        assert!(size_of::<Box<ServerStreamCallbacks>>() == size_of::<usize>());
+        let user_ptr = server_stream.cbs.as_ref() as *const ServerStreamCallbacks as *mut c_void;
+
+        let stream = unsafe {
+            let stream = context.stream_init(
+                stream_name,
+                input_device,
+                input_stream_params,
+                output_device,
+                output_stream_params,
+                latency,
+                Some(data_cb_c),
+                Some(state_cb_c),
+                user_ptr,
+            );
+            match stream {
+                Ok(stream) => stream,
+                Err(e) => {
+                    debug!("Unregistering stream {:?} (stream error {:?})", stm_tok, e);
+                    self.streams.remove(stm_tok);
+                    return Err(e.into());
+                }
+            }
+        };
+
+        server_stream.stream = Some(stream);
+
+        let client_pipe = server_stream
+            .client_pipe
+            .take()
+            .expect("invalid state after StreamCreated");
+        server_stream.cbs.connected.store(true, Ordering::Release);
+        Ok(ClientMessage::StreamInitialized(SerializableHandle::new(
+            client_pipe,
+            self.remote_pid.unwrap(),
+        )))
+    }
+}
+
+// C callable callbacks
+unsafe extern "C" fn data_cb_c(
+    _: *mut ffi::cubeb_stream,
+    user_ptr: *mut c_void,
+    input_buffer: *const c_void,
+    output_buffer: *mut c_void,
+    nframes: c_long,
+) -> c_long {
+    let ok = panic::catch_unwind(|| {
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        let input = if input_buffer.is_null() {
+            &[]
+        } else {
+            let nbytes = nframes * c_long::from(cbs.input_frame_size);
+            slice::from_raw_parts(input_buffer as *const u8, nbytes as usize)
+        };
+        let output: &mut [u8] = if output_buffer.is_null() {
+            &mut []
+        } else {
+            let nbytes = nframes * c_long::from(cbs.output_frame_size);
+            slice::from_raw_parts_mut(output_buffer as *mut u8, nbytes as usize)
+        };
+        cbs.data_callback(input, output, nframes as isize) as c_long
+    });
+    ok.unwrap_or(cubeb::ffi::CUBEB_ERROR as c_long)
+}
+
+unsafe extern "C" fn state_cb_c(
+    _: *mut ffi::cubeb_stream,
+    user_ptr: *mut c_void,
+    state: ffi::cubeb_state,
+) {
+    let ok = panic::catch_unwind(|| {
+        let state = cubeb::State::from(state);
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        cbs.state_callback(state);
+    });
+    ok.expect("State callback panicked");
+}
+
+unsafe extern "C" fn device_change_cb_c(user_ptr: *mut c_void) {
+    let ok = panic::catch_unwind(|| {
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        cbs.device_change_callback();
+    });
+    ok.expect("Device change callback panicked");
+}
+
+unsafe extern "C" fn device_collection_changed_input_cb_c(
+    _: *mut ffi::cubeb,
+    user_ptr: *mut c_void,
+) {
+    let ok = panic::catch_unwind(|| {
+        let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager);
+        manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_INPUT);
+    });
+    ok.expect("Collection changed (input) callback panicked");
+}
+
+unsafe extern "C" fn device_collection_changed_output_cb_c(
+    _: *mut ffi::cubeb,
+    user_ptr: *mut c_void,
+) {
+    let ok = panic::catch_unwind(|| {
+        let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager);
+        manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_OUTPUT);
+    });
+    ok.expect("Collection changed (output) callback panicked");
+}
diff --git a/third_party/rust/audioipc2/.cargo-checksum.json b/third_party/rust/audioipc2/.cargo-checksum.json
new file mode 100644
index 0000000000..96d8f77888
--- /dev/null
+++ b/third_party/rust/audioipc2/.cargo-checksum.json
@@ -0,0 +1 @@
+{"files":{"Cargo.toml":"6d3b3004351e3313ef74472ab6b3f96d59a785b3d873698ca4caf67eb3d47aab","benches/serialization.rs":"d56855d868dab6aa22c8b03a61084535351b76c94b68d8b1d20764e352fe473f","build.rs":"65df9a97c6cdaa3faf72581f04ac289197b0b1797d69d22c1796e957ff1089e2","src/codec.rs":"38408b512d935cd7889a03b25dd14b36083ec4e6d2fcabd636182cf45e3d50bc","src/errors.rs":"67a4a994d0724397657581cde153bdfc05ce86e7efc467f23fafc8f64df80fa4","src/ipccore.rs":"ba339eebdc2d8a6d0cb9b294344809e63e404e220fda643b63a8a3ff63a755e8","src/lib.rs":"9b107cb52081eeea3fa742d30361db70f7138baa423dfe21d37dcf5087afc338","src/messages.rs":"452362da2cace9a0f2e3134c190ecb6a9997f8be4036cde06643e17c6c238240","src/rpccore.rs":"025b6614f1c42b96b0a8e74fd7881032d338c66e0d67ec0af70f910a9e30ebe1","src/shm.rs":"1d88f19606899e3e477865d6b84bbe3e272f51618a1c2d57b6dab03a4787cde3","src/sys/mod.rs":"e6fa1d260abf093e1f7b50185195e2d3aee0eb8c9774c6f253953b5896d838f3","src/sys/unix/cmsg.rs":"22349b3df39b51b9c414da363313c92d41b02a623753ffcca6f59613e8f79eb2","src/sys/unix/cmsghdr.c":"d7344b3dc15cdce410c68669b848bb81f7fe36362cd3699668cb613fa05180f8","src/sys/unix/mod.rs":"59835f0d5509940078b1820a54f49fc5514adeb3e45e7d21e3ab917431da2e74","src/sys/unix/msg.rs":"0e297d73bae9414184f85c2209cca0a3fde6d999a3f1d3f42faa3f56b6d57233","src/sys/windows/mod.rs":"3441a3212c6d44443a5975621d9594b0c841e5a7f113aa1b108a080330df2b77"},"package":null}
\ No newline at end of file
diff --git a/third_party/rust/audioipc2/Cargo.toml b/third_party/rust/audioipc2/Cargo.toml
new file mode 100644
index 0000000000..98de8c77a6
--- /dev/null
+++ b/third_party/rust/audioipc2/Cargo.toml
@@ -0,0 +1,82 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2018"
+name = "audioipc2"
+version = "0.5.0"
+authors = [
+    "Matthew Gregan <kinetik@flim.org>",
+    "Dan Glastonbury <dan.glastonbury@gmail.com>",
+]
+description = "Remote Cubeb IPC"
+license = "ISC"
+
+[[bench]]
+name = "serialization"
+harness = false
+
+[dependencies]
+bincode = "1.3"
+byteorder = "1"
+bytes = "1"
+crossbeam-queue = "0.3"
+cubeb = "0.10"
+log = "0.4"
+scopeguard = "1.1.0"
+serde = "1"
+serde_bytes = "0.11"
+serde_derive = "1"
+slab = "0.4"
+
+[dependencies.error-chain]
+version = "0.12.0"
+default-features = false
+
+[dependencies.mio]
+version = "0.8"
+features = [
+    "os-poll",
+    "net",
+    "os-ext",
+]
+
+[dev-dependencies]
+env_logger = "0.9"
+
+[dev-dependencies.criterion]
+version = "0.3"
+features = ["html_reports"]
+
+[build-dependencies]
+cc = "1.0"
+
+[target."cfg(target_os = \"android\")".dependencies]
+ashmem = "0.1.2"
+
+[target."cfg(target_os = \"linux\")".dependencies.audio_thread_priority]
+version = "0.26.1"
+default-features = false
+
+[target."cfg(unix)".dependencies]
+arrayvec = "0.7"
+iovec = "0.1"
+libc = "0.2"
+memmap2 = "0.5"
+
+[target."cfg(windows)".dependencies.winapi]
+version = "0.3"
+features = [
+    "combaseapi",
+    "handleapi",
+    "memoryapi",
+    "objbase",
+]
diff --git a/third_party/rust/audioipc2/benches/serialization.rs b/third_party/rust/audioipc2/benches/serialization.rs
new file mode 100644
index 0000000000..39f770a939
--- /dev/null
+++ b/third_party/rust/audioipc2/benches/serialization.rs
@@ -0,0 +1,93 @@
+use audioipc::codec::{Codec, LengthDelimitedCodec};
+use audioipc::messages::DeviceInfo;
+use audioipc::ClientMessage;
+use audioipc2 as audioipc;
+use bytes::BytesMut;
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+
+fn bench(c: &mut Criterion, name: &str, msg: impl Fn() -> ClientMessage) {
+    let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> =
+        LengthDelimitedCodec::default();
+    let mut buf = BytesMut::with_capacity(8192);
+    c.bench_function(&format!("encode/{}", name), |b| {
+        b.iter_batched(
+            || msg(),
+            |msg| {
+                codec.encode(msg, &mut buf).unwrap();
+                buf.clear();
+            },
+            BatchSize::SmallInput,
+        )
+    });
+
+    let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> =
+        LengthDelimitedCodec::default();
+    let mut buf = BytesMut::with_capacity(8192);
+    codec.encode(msg(), &mut buf).unwrap();
+    c.bench_function(&format!("decode/{}", name), |b| {
+        b.iter_batched_ref(
+            || buf.clone(),
+            |buf| {
+                codec.decode(buf).unwrap().unwrap();
+            },
+            BatchSize::SmallInput,
+        )
+    });
+
+    let mut codec: LengthDelimitedCodec<ClientMessage, ClientMessage> =
+        LengthDelimitedCodec::default();
+    let mut buf = BytesMut::with_capacity(8192);
+    c.bench_function(&format!("roundtrip/{}", name), |b| {
+        b.iter_batched(
+            || msg(),
+            |msg| {
+                codec.encode(msg, &mut buf).unwrap();
+                codec.decode(&mut buf).unwrap().unwrap();
+            },
+            BatchSize::SmallInput,
+        )
+    });
+}
+
+pub fn criterion_benchmark(c: &mut Criterion) {
+    bench(c, "tiny", || ClientMessage::ClientConnected);
+    bench(c, "small", || ClientMessage::StreamPosition(0));
+    bench(c, "medium", || {
+        ClientMessage::ContextEnumeratedDevices(make_device_vec(2))
+    });
+    bench(c, "large", || {
+        ClientMessage::ContextEnumeratedDevices(make_device_vec(20))
+    });
+    bench(c, "huge", || {
+        ClientMessage::ContextEnumeratedDevices(make_device_vec(128))
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
+
+fn make_device_vec(n: usize) -> Vec<DeviceInfo> {
+    let mut devices = Vec::with_capacity(n);
+    for i in 0..n {
+        let device = DeviceInfo {
+            devid: i,
+            device_id: Some(vec![0u8; 64]),
+            friendly_name: Some(vec![0u8; 64]),
+            group_id: Some(vec![0u8; 64]),
+            vendor_name: Some(vec![0u8; 64]),
+            device_type: 0,
+            state: 0,
+            preferred: 0,
+            format: 0,
+            default_format: 0,
+            max_channels: 0,
+            default_rate: 0,
+            max_rate: 0,
+            min_rate: 0,
+            latency_lo: 0,
+            latency_hi: 0,
+        };
+        devices.push(device);
+    }
+    devices
+}
diff --git a/third_party/rust/audioipc2/build.rs b/third_party/rust/audioipc2/build.rs
new file mode 100644
index 0000000000..453dafad96
--- /dev/null
+++ b/third_party/rust/audioipc2/build.rs
@@ -0,0 +1,7 @@
+fn main() {
+    if std::env::var_os("CARGO_CFG_UNIX").is_some() {
+        cc::Build::new()
+            .file("src/sys/unix/cmsghdr.c")
+            .compile("cmsghdr");
+    }
+}
diff --git a/third_party/rust/audioipc2/src/codec.rs b/third_party/rust/audioipc2/src/codec.rs
new file mode 100644
index 0000000000..2c61faa6ea
--- /dev/null
+++ b/third_party/rust/audioipc2/src/codec.rs
@@ -0,0 +1,203 @@
+// 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.
+
+// The assert in LengthDelimitedCodec::decode triggers this clippy warning but
+// requires upgrading the workspace to Rust 2021 to resolve.
+// This should be fixed in Rust 1.68, after which the following `allow` can be deleted.
+#![allow(clippy::uninlined_format_args)]
+
+use bincode::{self, Options};
+use byteorder::{ByteOrder, LittleEndian};
+use bytes::{Buf, BufMut, BytesMut};
+use serde::de::DeserializeOwned;
+use serde::ser::Serialize;
+use std::convert::TryInto;
+use std::fmt::Debug;
+use std::io;
+use std::marker::PhantomData;
+use std::mem::size_of;
+
+////////////////////////////////////////////////////////////////////////////////
+// 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 into 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,
+    encode_buf: Vec<u8>,
+    __in: PhantomData<In>,
+    __out: PhantomData<Out>,
+}
+
+enum State {
+    Length,
+    Data(u32),
+}
+
+const MAX_MESSAGE_LEN: u32 = 1024 * 1024;
+const MAGIC: u64 = 0xa4d1_019c_c910_1d4a;
+const HEADER_LEN: usize = size_of::<u32>() + size_of::<u64>();
+
+impl<In, Out> Default for LengthDelimitedCodec<In, Out> {
+    fn default() -> Self {
+        Self {
+            state: State::Length,
+            encode_buf: Vec::with_capacity(crate::ipccore::IPC_CLIENT_BUFFER_SIZE),
+            __in: PhantomData,
+            __out: PhantomData,
+        }
+    }
+}
+
+impl<In, Out> LengthDelimitedCodec<In, Out> {
+    // Lengths are encoded as little endian u32
+    fn decode_length(buf: &mut BytesMut) -> Option<u32> {
+        if buf.len() < HEADER_LEN {
+            // Not enough data
+            return None;
+        }
+
+        let magic = LittleEndian::read_u64(&buf[0..8]);
+        assert_eq!(magic, MAGIC);
+
+        // Consume the length field
+        let n = LittleEndian::read_u32(&buf[8..12]);
+        buf.advance(HEADER_LEN);
+        Some(n)
+    }
+
+    fn decode_data(buf: &mut BytesMut, n: u32) -> io::Result<Option<Out>>
+    where
+        Out: DeserializeOwned + Debug,
+    {
+        let n = n.try_into().unwrap();
+
+        // 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);
+        }
+
+        trace!("Attempting to decode");
+        let msg = bincode::options()
+            .with_limit(MAX_MESSAGE_LEN as u64)
+            .deserialize::<Out>(&buf[..n])
+            .map_err(|e| match *e {
+                bincode::ErrorKind::Io(e) => e,
+                _ => io::Error::new(io::ErrorKind::Other, *e),
+            })?;
+        buf.advance(n);
+
+        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) => {
+                        assert!(
+                            n <= MAX_MESSAGE_LEN,
+                            "assertion failed: {} <= {}",
+                            n,
+                            MAX_MESSAGE_LEN
+                        );
+                        self.state = State::Data(n);
+
+                        // Ensure that the buffer has enough space to read the
+                        // incoming payload
+                        buf.reserve(n.try_into().unwrap());
+
+                        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 length header.
+                buf.reserve(HEADER_LEN);
+
+                Ok(Some(data))
+            }
+            None => Ok(None),
+        }
+    }
+
+    fn encode(&mut self, item: Self::In, buf: &mut BytesMut) -> io::Result<()> {
+        trace!("Attempting to encode");
+
+        self.encode_buf.clear();
+        if let Err(e) = bincode::options()
+            .with_limit(MAX_MESSAGE_LEN as u64)
+            .serialize_into::<_, Self::In>(&mut self.encode_buf, &item)
+        {
+            trace!("message encode failed: {:?}", *e);
+            match *e {
+                bincode::ErrorKind::Io(e) => return Err(e),
+                _ => return Err(io::Error::new(io::ErrorKind::Other, *e)),
+            }
+        }
+
+        let encoded_len = self.encode_buf.len();
+        assert!(encoded_len <= MAX_MESSAGE_LEN as usize);
+        buf.reserve(encoded_len + HEADER_LEN);
+        buf.put_u64_le(MAGIC);
+        buf.put_u32_le(encoded_len.try_into().unwrap());
+        buf.extend_from_slice(&self.encode_buf);
+
+        Ok(())
+    }
+}
diff --git a/third_party/rust/audioipc2/src/errors.rs b/third_party/rust/audioipc2/src/errors.rs
new file mode 100644
index 0000000000..f4f3e32f5f
--- /dev/null
+++ b/third_party/rust/audioipc2/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/audioipc2/src/ipccore.rs b/third_party/rust/audioipc2/src/ipccore.rs
new file mode 100644
index 0000000000..16468b9015
--- /dev/null
+++ b/third_party/rust/audioipc2/src/ipccore.rs
@@ -0,0 +1,918 @@
+// Copyright © 2021 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use std::io::{self, Result};
+use std::sync::{mpsc, Arc};
+use std::thread;
+
+use mio::{event::Event, Events, Interest, Poll, Registry, Token, Waker};
+use slab::Slab;
+
+use crate::messages::AssociateHandleForMessage;
+use crate::rpccore::{
+    make_client, make_server, Client, Handler, Proxy, RequestQueue, RequestQueueSender, Server,
+};
+use crate::{
+    codec::Codec,
+    codec::LengthDelimitedCodec,
+    sys::{self, RecvMsg, SendMsg},
+};
+
+use serde::{de::DeserializeOwned, Serialize};
+use std::fmt::Debug;
+
+const WAKE_TOKEN: Token = Token(!0);
+
+thread_local!(static IN_EVENTLOOP: std::cell::RefCell<Option<thread::ThreadId>> = std::cell::RefCell::new(None));
+
+fn assert_not_in_event_loop_thread() {
+    IN_EVENTLOOP.with(|b| {
+        assert_ne!(*b.borrow(), Some(thread::current().id()));
+    });
+}
+
+// Requests sent by an EventLoopHandle to be serviced by
+// the handle's associated EventLoop.
+enum Request {
+    // See EventLoop::add_connection
+    AddConnection(
+        sys::Pipe,
+        Box<dyn Driver + Send>,
+        mpsc::Sender<Result<Token>>,
+    ),
+    // See EventLoop::shutdown
+    Shutdown,
+    // See EventLoop::wake_connection
+    WakeConnection(Token),
+}
+
+// EventLoopHandle is a cloneable external reference
+// to a running EventLoop, allowing registration of
+// new client and server connections, in addition to
+// requesting the EventLoop shut down cleanly.
+#[derive(Clone, Debug)]
+pub struct EventLoopHandle {
+    waker: Arc<Waker>,
+    requests: RequestQueueSender<Request>,
+}
+
+impl EventLoopHandle {
+    pub fn bind_client<C: Client + 'static>(
+        &self,
+        connection: sys::Pipe,
+    ) -> Result<Proxy<<C as Client>::ServerMessage, <C as Client>::ClientMessage>>
+    where
+        <C as Client>::ServerMessage: Serialize + Debug + AssociateHandleForMessage + Send,
+        <C as Client>::ClientMessage: DeserializeOwned + Debug + AssociateHandleForMessage + Send,
+    {
+        let (handler, mut proxy) = make_client::<C>()?;
+        let driver = Box::new(FramedDriver::new(handler));
+        let r = self.add_connection(connection, driver);
+        trace!("EventLoop::bind_client {:?}", r);
+        r.map(|token| {
+            proxy.connect_event_loop(self.clone(), token);
+            proxy
+        })
+    }
+
+    pub fn bind_server<S: Server + Send + 'static>(
+        &self,
+        server: S,
+        connection: sys::Pipe,
+    ) -> Result<()>
+    where
+        <S as Server>::ServerMessage: DeserializeOwned + Debug + AssociateHandleForMessage + Send,
+        <S as Server>::ClientMessage: Serialize + Debug + AssociateHandleForMessage + Send,
+    {
+        let handler = make_server::<S>(server);
+        let driver = Box::new(FramedDriver::new(handler));
+        let r = self.add_connection(connection, driver);
+        trace!("EventLoop::bind_server {:?}", r);
+        r.map(|_| ())
+    }
+
+    // Register a new connection with associated driver on the EventLoop.
+    // TODO: Since this is called from a Gecko main thread, make this non-blocking wrt. the EventLoop.
+    fn add_connection(
+        &self,
+        connection: sys::Pipe,
+        driver: Box<dyn Driver + Send>,
+    ) -> Result<Token> {
+        assert_not_in_event_loop_thread();
+        let (tx, rx) = mpsc::channel();
+        self.requests
+            .push(Request::AddConnection(connection, driver, tx))
+            .map_err(|_| {
+                debug!("EventLoopHandle::add_connection send failed");
+                io::ErrorKind::ConnectionAborted
+            })?;
+        self.waker.wake()?;
+        rx.recv().map_err(|_| {
+            debug!("EventLoopHandle::add_connection recv failed");
+            io::ErrorKind::ConnectionAborted
+        })?
+    }
+
+    // Signal EventLoop to shutdown.  Causes EventLoop::poll to return Ok(false).
+    fn shutdown(&self) -> Result<()> {
+        self.requests.push(Request::Shutdown).map_err(|_| {
+            debug!("EventLoopHandle::shutdown send failed");
+            io::ErrorKind::ConnectionAborted
+        })?;
+        self.waker.wake()
+    }
+
+    // Signal EventLoop to wake connection specified by `token` for processing.
+    pub(crate) fn wake_connection(&self, token: Token) {
+        if self.requests.push(Request::WakeConnection(token)).is_ok() {
+            self.waker.wake().expect("wake failed");
+        }
+    }
+}
+
+// EventLoop owns all registered connections, and is responsible for calling each connection's
+// `handle_event` or `handle_wake` any time a readiness or wake event associated with that connection is
+// produced.
+struct EventLoop {
+    poll: Poll,
+    events: Events,
+    waker: Arc<Waker>,
+    name: String,
+    connections: Slab<Connection>,
+    requests: Arc<RequestQueue<Request>>,
+}
+
+const EVENT_LOOP_INITIAL_CLIENTS: usize = 64; // Initial client allocation, exceeding this will cause the connection slab to grow.
+const EVENT_LOOP_EVENTS_PER_ITERATION: usize = 256; // Number of events per poll() step, arbitrary limit.
+
+impl EventLoop {
+    fn new(name: String) -> Result<EventLoop> {
+        let poll = Poll::new()?;
+        let waker = Arc::new(Waker::new(poll.registry(), WAKE_TOKEN)?);
+        let eventloop = EventLoop {
+            poll,
+            events: Events::with_capacity(EVENT_LOOP_EVENTS_PER_ITERATION),
+            waker,
+            name,
+            connections: Slab::with_capacity(EVENT_LOOP_INITIAL_CLIENTS),
+            requests: Arc::new(RequestQueue::new(EVENT_LOOP_INITIAL_CLIENTS)),
+        };
+
+        Ok(eventloop)
+    }
+
+    // Return a cloneable handle for controlling the EventLoop externally.
+    fn handle(&mut self) -> EventLoopHandle {
+        EventLoopHandle {
+            waker: self.waker.clone(),
+            requests: self.requests.new_sender(),
+        }
+    }
+
+    // Register a connection and driver.
+    fn add_connection(
+        &mut self,
+        connection: sys::Pipe,
+        driver: Box<dyn Driver + Send>,
+    ) -> Result<Token> {
+        if self.connections.len() == self.connections.capacity() {
+            trace!("{}: connection slab full, insert will allocate", self.name);
+        }
+        let entry = self.connections.vacant_entry();
+        let token = Token(entry.key());
+        assert_ne!(token, WAKE_TOKEN);
+        let connection = Connection::new(connection, token, driver, self.poll.registry())?;
+        debug!("{}: {:?}: new connection", self.name, token);
+        entry.insert(connection);
+        Ok(token)
+    }
+
+    // Step EventLoop once.  Call this in a loop from a dedicated thread.
+    // Returns false if EventLoop is shutting down.
+    // Each step may call `handle_event` on any registered connection that
+    // has received readiness events from the poll wakeup.
+    fn poll(&mut self) -> Result<bool> {
+        loop {
+            let r = self.poll.poll(&mut self.events, None);
+            match r {
+                Ok(()) => break,
+                Err(ref e) if interrupted(e) => continue,
+                Err(e) => return Err(e),
+            }
+        }
+
+        for event in self.events.iter() {
+            match event.token() {
+                WAKE_TOKEN => {
+                    debug!("{}: WAKE: wake event, will process requests", self.name);
+                }
+                token => {
+                    debug!("{}: {:?}: connection event: {:?}", self.name, token, event);
+                    let done = if let Some(connection) = self.connections.get_mut(token.0) {
+                        match connection.handle_event(event, self.poll.registry()) {
+                            Ok(done) => {
+                                debug!("{}: connection {:?} done={}", self.name, token, done);
+                                done
+                            }
+                            Err(e) => {
+                                debug!("{}: {:?}: connection error: {:?}", self.name, token, e);
+                                true
+                            }
+                        }
+                    } else {
+                        // Spurious event, log and ignore.
+                        debug!(
+                            "{}: {:?}: token not found in slab: {:?}",
+                            self.name, token, event
+                        );
+                        false
+                    };
+                    if done {
+                        debug!("{}: {:?}: done, removing", self.name, token);
+                        let mut connection = self.connections.remove(token.0);
+                        if let Err(e) = connection.shutdown(self.poll.registry()) {
+                            debug!(
+                                "{}: EventLoop drop - closing connection for {:?} failed: {:?}",
+                                self.name, token, e
+                            );
+                        }
+                    }
+                }
+            }
+        }
+
+        // If the waker was signalled there may be pending requests to process.
+        while let Some(req) = self.requests.pop() {
+            match req {
+                Request::AddConnection(pipe, driver, tx) => {
+                    debug!("{}: EventLoop: handling add_connection", self.name);
+                    let r = self.add_connection(pipe, driver);
+                    tx.send(r).expect("EventLoop::add_connection");
+                }
+                Request::Shutdown => {
+                    debug!("{}: EventLoop: handling shutdown", self.name);
+                    return Ok(false);
+                }
+                Request::WakeConnection(token) => {
+                    debug!(
+                        "{}: EventLoop: handling wake_connection {:?}",
+                        self.name, token
+                    );
+                    let done = if let Some(connection) = self.connections.get_mut(token.0) {
+                        match connection.handle_wake(self.poll.registry()) {
+                            Ok(done) => done,
+                            Err(e) => {
+                                debug!("{}: {:?}: connection error: {:?}", self.name, token, e);
+                                true
+                            }
+                        }
+                    } else {
+                        // Spurious wake, log and ignore.
+                        debug!(
+                            "{}: {:?}: token not found in slab: wake_connection",
+                            self.name, token
+                        );
+                        false
+                    };
+                    if done {
+                        debug!("{}: {:?}: done (wake), removing", self.name, token);
+                        let mut connection = self.connections.remove(token.0);
+                        if let Err(e) = connection.shutdown(self.poll.registry()) {
+                            debug!(
+                                "{}: EventLoop drop - closing connection for {:?} failed: {:?}",
+                                self.name, token, e
+                            );
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(true)
+    }
+}
+
+impl Drop for EventLoop {
+    fn drop(&mut self) {
+        debug!("{}: EventLoop drop", self.name);
+        for (token, connection) in &mut self.connections {
+            debug!(
+                "{}: EventLoop drop - closing connection for {:?}",
+                self.name, token
+            );
+            if let Err(e) = connection.shutdown(self.poll.registry()) {
+                debug!(
+                    "{}: EventLoop drop - closing connection for {:?} failed: {:?}",
+                    self.name, token, e
+                );
+            }
+        }
+        debug!("{}: EventLoop drop done", self.name);
+    }
+}
+
+// Connection wraps an interprocess connection (Pipe) and manages
+// receiving inbound and sending outbound buffers (and associated handles, if any).
+// The associated driver is responsible for message framing and serialization.
+struct Connection {
+    io: sys::Pipe,
+    token: Token,
+    interest: Option<Interest>,
+    inbound: sys::ConnectionBuffer,
+    outbound: sys::ConnectionBuffer,
+    driver: Box<dyn Driver + Send>,
+}
+
+pub(crate) const IPC_CLIENT_BUFFER_SIZE: usize = 16384;
+
+impl Connection {
+    fn new(
+        mut io: sys::Pipe,
+        token: Token,
+        driver: Box<dyn Driver + Send>,
+        registry: &Registry,
+    ) -> Result<Connection> {
+        let interest = Interest::READABLE;
+        registry.register(&mut io, token, interest)?;
+        Ok(Connection {
+            io,
+            token,
+            interest: Some(interest),
+            inbound: sys::ConnectionBuffer::with_capacity(IPC_CLIENT_BUFFER_SIZE),
+            outbound: sys::ConnectionBuffer::with_capacity(IPC_CLIENT_BUFFER_SIZE),
+            driver,
+        })
+    }
+
+    fn shutdown(&mut self, registry: &Registry) -> Result<()> {
+        trace!(
+            "{:?}: connection shutdown interest={:?}",
+            self.token,
+            self.interest
+        );
+        let r = self.io.shutdown();
+        trace!("{:?}: connection shutdown r={:?}", self.token, r);
+        self.interest = None;
+        registry.deregister(&mut self.io)
+    }
+
+    // Connections are always interested in READABLE.  clear_readable is only
+    // called when the connection is in the process of shutting down.
+    fn clear_readable(&mut self, registry: &Registry) -> Result<()> {
+        self.update_registration(
+            registry,
+            self.interest.and_then(|i| i.remove(Interest::READABLE)),
+        )
+    }
+
+    // Connections toggle WRITABLE based on the state of the `outbound` buffer.
+    fn set_writable(&mut self, registry: &Registry) -> Result<()> {
+        self.update_registration(
+            registry,
+            Some(
+                self.interest
+                    .map_or_else(|| Interest::WRITABLE, |i| i.add(Interest::WRITABLE)),
+            ),
+        )
+    }
+
+    fn clear_writable(&mut self, registry: &Registry) -> Result<()> {
+        self.update_registration(
+            registry,
+            self.interest.and_then(|i| i.remove(Interest::WRITABLE)),
+        )
+    }
+
+    // Update connection registration with the current readiness event interests.
+    fn update_registration(
+        &mut self,
+        registry: &Registry,
+        new_interest: Option<Interest>,
+    ) -> Result<()> {
+        // Note: Updating registration always triggers a writable event with NamedPipes, so
+        // it's important to skip updating registration when the set of interests hasn't changed.
+        if new_interest != self.interest {
+            trace!(
+                "{:?}: updating readiness registration old={:?} new={:?}",
+                self.token,
+                self.interest,
+                new_interest
+            );
+            self.interest = new_interest;
+            if let Some(interest) = self.interest {
+                registry.reregister(&mut self.io, self.token, interest)?;
+            } else {
+                registry.deregister(&mut self.io)?;
+            }
+        }
+        Ok(())
+    }
+
+    // Handle readiness event.  Errors returned are fatal for this connection, resulting in removal from the EventLoop connection list.
+    // The EventLoop will call this for any connection that has received an event.
+    fn handle_event(&mut self, event: &Event, registry: &Registry) -> Result<bool> {
+        debug!("{:?}: handling event {:?}", self.token, event);
+        assert_eq!(self.token, event.token());
+        let done = if event.is_readable() {
+            self.recv_inbound()?
+        } else {
+            trace!("{:?}: not readable", self.token);
+            false
+        };
+        self.flush_outbound()?;
+        if self.send_outbound(registry)? {
+            // Hit EOF during send
+            return Ok(true);
+        }
+        debug!(
+            "{:?}: handling event done (recv done={}, outbound={})",
+            self.token,
+            done,
+            self.outbound.is_empty()
+        );
+        let done = done && self.outbound.is_empty();
+        // If driver is done and outbound is clear, unregister connection.
+        if done {
+            trace!("{:?}: driver done, clearing read interest", self.token);
+            self.clear_readable(registry)?;
+        }
+        Ok(done)
+    }
+
+    // Handle wake event.  Errors returned are fatal for this connection, resulting in removal from the EventLoop connection list.
+    // The EventLoop will call this to clear the outbound buffer for any connection that has received a wake event.
+    fn handle_wake(&mut self, registry: &Registry) -> Result<bool> {
+        debug!("{:?}: handling wake", self.token);
+        self.flush_outbound()?;
+        if self.send_outbound(registry)? {
+            // Hit EOF during send
+            return Ok(true);
+        }
+        debug!("{:?}: handling wake done", self.token);
+        Ok(false)
+    }
+
+    fn recv_inbound(&mut self) -> Result<bool> {
+        // If the connection is readable, read into inbound and pass to driver for processing until all ready data
+        // has been consumed.
+        loop {
+            trace!("{:?}: pre-recv inbound: {:?}", self.token, self.inbound);
+            let r = self.io.recv_msg(&mut self.inbound);
+            match r {
+                Ok(0) => {
+                    trace!(
+                        "{:?}: recv EOF unprocessed inbound={}",
+                        self.token,
+                        self.inbound.is_empty()
+                    );
+                    return Ok(true);
+                }
+                Ok(n) => {
+                    trace!("{:?}: recv bytes: {}, process_inbound", self.token, n);
+                    let r = self.driver.process_inbound(&mut self.inbound);
+                    trace!("{:?}: process_inbound done: {:?}", self.token, r);
+                    match r {
+                        Ok(done) => {
+                            if done {
+                                return Ok(true);
+                            }
+                        }
+                        Err(e) => {
+                            debug!(
+                                "{:?}: process_inbound error: {:?} unprocessed inbound={}",
+                                self.token,
+                                e,
+                                self.inbound.is_empty()
+                            );
+                            return Err(e);
+                        }
+                    }
+                }
+                Err(ref e) if would_block(e) => {
+                    trace!("{:?}: recv would_block: {:?}", self.token, e);
+                    return Ok(false);
+                }
+                Err(ref e) if interrupted(e) => {
+                    trace!("{:?}: recv interrupted: {:?}", self.token, e);
+                    continue;
+                }
+                Err(e) => {
+                    debug!("{:?}: recv error: {:?}", self.token, e);
+                    return Err(e);
+                }
+            }
+        }
+    }
+
+    fn flush_outbound(&mut self) -> Result<()> {
+        // Enqueue outbound messages to the outbound buffer, then try to write out to connection.
+        // There may be outbound messages even if there was no inbound processing, so always attempt
+        // to enqueue and flush.
+        trace!("{:?}: flush_outbound", self.token);
+        let r = self.driver.flush_outbound(&mut self.outbound);
+        trace!("{:?}: flush_outbound done: {:?}", self.token, r);
+        if let Err(e) = r {
+            debug!("{:?}: flush_outbound error: {:?}", self.token, e);
+            return Err(e);
+        }
+        Ok(())
+    }
+
+    fn send_outbound(&mut self, registry: &Registry) -> Result<bool> {
+        // Attempt to flush outbound buffer.  If the connection's write buffer is full, register for WRITABLE
+        // and complete flushing when associated notitication arrives later.
+        while !self.outbound.is_empty() {
+            let r = self.io.send_msg(&mut self.outbound);
+            match r {
+                Ok(0) => {
+                    trace!("{:?}: send EOF", self.token);
+                    return Ok(true);
+                }
+                Ok(n) => {
+                    trace!("{:?}: send bytes: {}", self.token, n);
+                }
+                Err(ref e) if would_block(e) => {
+                    trace!(
+                        "{:?}: send would_block: {:?}, setting write interest",
+                        self.token,
+                        e
+                    );
+                    // Register for write events.
+                    self.set_writable(registry)?;
+                    break;
+                }
+                Err(ref e) if interrupted(e) => {
+                    trace!("{:?}: send interrupted: {:?}", self.token, e);
+                    continue;
+                }
+                Err(e) => {
+                    debug!("{:?}: send error: {:?}", self.token, e);
+                    return Err(e);
+                }
+            }
+            trace!("{:?}: post-send: outbound {:?}", self.token, self.outbound);
+        }
+        // Outbound buffer flushed, clear registration for WRITABLE.
+        if self.outbound.is_empty() {
+            trace!("{:?}: outbound empty, clearing write interest", self.token);
+            self.clear_writable(registry)?;
+        }
+        Ok(false)
+    }
+}
+
+impl Drop for Connection {
+    fn drop(&mut self) {
+        debug!("{:?}: Connection drop", self.token);
+    }
+}
+
+fn would_block(err: &std::io::Error) -> bool {
+    err.kind() == std::io::ErrorKind::WouldBlock
+}
+
+fn interrupted(err: &std::io::Error) -> bool {
+    err.kind() == std::io::ErrorKind::Interrupted
+}
+
+// Driver only has a single implementation, but must be hidden behind a Trait object to
+// hide the varying FramedDriver sizes (due to different `T` values).
+trait Driver {
+    // Handle inbound messages.  Returns true if Driver is done; this will trigger Connection removal and cleanup.
+    fn process_inbound(&mut self, inbound: &mut sys::ConnectionBuffer) -> Result<bool>;
+
+    // Write outbound messages to `outbound`.
+    fn flush_outbound(&mut self, outbound: &mut sys::ConnectionBuffer) -> Result<()>;
+}
+
+// Length-delimited connection framing and (de)serialization is handled by the inbound and outbound processing.
+// Handlers can then process message Requests and Responses without knowledge of serialization or
+// handle remoting.
+impl<T> Driver for FramedDriver<T>
+where
+    T: Handler,
+    T::In: DeserializeOwned + Debug + AssociateHandleForMessage,
+    T::Out: Serialize + Debug + AssociateHandleForMessage,
+{
+    // Caller passes `inbound` data, this function will trim any complete messages from `inbound` and pass them to the handler for processing.
+    fn process_inbound(&mut self, inbound: &mut sys::ConnectionBuffer) -> Result<bool> {
+        debug!("process_inbound: {:?}", inbound);
+
+        // Repeatedly call `decode` as long as it produces items, passing each produced item to the handler to action.
+        #[allow(unused_mut)]
+        while let Some(mut item) = self.codec.decode(&mut inbound.buf)? {
+            if item.has_associated_handle() {
+                // On Unix, dequeue a handle from the connection and update the item's handle.
+                #[cfg(unix)]
+                {
+                    let new = inbound
+                        .pop_handle()
+                        .expect("inbound handle expected for item");
+                    unsafe { item.set_local_handle(new.take()) };
+                }
+                // On Windows, the deserialized item contains the correct handle value, so
+                // convert it to an owned handle on the item.
+                #[cfg(windows)]
+                {
+                    assert!(inbound.pop_handle().is_none());
+                    unsafe { item.set_local_handle() };
+                }
+            }
+
+            self.handler.consume(item)?;
+        }
+
+        Ok(false)
+    }
+
+    // Caller will try to write `outbound` to associated connection, queuing any data that can't be transmitted immediately.
+    fn flush_outbound(&mut self, outbound: &mut sys::ConnectionBuffer) -> Result<()> {
+        debug!("flush_outbound: {:?}", outbound.buf);
+
+        // Repeatedly grab outgoing items from the handler, passing each to `encode` for serialization into `outbound`.
+        while let Some(mut item) = self.handler.produce()? {
+            let handle = if item.has_associated_handle() {
+                #[allow(unused_mut)]
+                let mut handle = item.take_handle();
+                // On Windows, the handle is transferred by duplicating it into the target remote process.
+                #[cfg(windows)]
+                unsafe {
+                    item.set_remote_handle(handle.send_to_target()?);
+                }
+                Some(handle)
+            } else {
+                None
+            };
+
+            self.codec.encode(item, &mut outbound.buf)?;
+            if let Some(handle) = handle {
+                // `outbound` retains ownership of the handle until the associated
+                // encoded item in `outbound.buf` is sent to the remote process.
+                outbound.push_handle(handle);
+            }
+        }
+        Ok(())
+    }
+}
+
+struct FramedDriver<T: Handler> {
+    codec: LengthDelimitedCodec<T::Out, T::In>,
+    handler: T,
+}
+
+impl<T: Handler> FramedDriver<T> {
+    fn new(handler: T) -> FramedDriver<T> {
+        FramedDriver {
+            codec: Default::default(),
+            handler,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct EventLoopThread {
+    thread: Option<thread::JoinHandle<Result<()>>>,
+    name: String,
+    handle: EventLoopHandle,
+}
+
+impl EventLoopThread {
+    pub fn new<F1, F2>(
+        name: String,
+        stack_size: Option<usize>,
+        after_start: F1,
+        before_stop: F2,
+    ) -> Result<Self>
+    where
+        F1: Fn() + Send + 'static,
+        F2: Fn() + Send + 'static,
+    {
+        let mut event_loop = EventLoop::new(name.clone())?;
+        let handle = event_loop.handle();
+
+        let builder = thread::Builder::new()
+            .name(name.clone())
+            .stack_size(stack_size.unwrap_or(64 * 4096));
+
+        let thread = builder.spawn(move || {
+            trace!("{}: event loop thread enter", event_loop.name);
+            after_start();
+            let _thread_exit_guard = scopeguard::guard((), |_| before_stop());
+
+            let r = loop {
+                let start = std::time::Instant::now();
+                let r = event_loop.poll();
+                trace!(
+                    "{}: event loop poll r={:?}, took={}μs",
+                    event_loop.name,
+                    r,
+                    start.elapsed().as_micros()
+                );
+                match r {
+                    Ok(true) => continue,
+                    _ => break r,
+                }
+            };
+
+            trace!("{}: event loop thread exit", event_loop.name);
+            r.map(|_| ())
+        })?;
+
+        Ok(EventLoopThread {
+            thread: Some(thread),
+            name,
+            handle,
+        })
+    }
+
+    pub fn handle(&self) -> &EventLoopHandle {
+        &self.handle
+    }
+}
+
+impl Drop for EventLoopThread {
+    // Shut down event loop and executor thread.  Blocks until complete.
+    fn drop(&mut self) {
+        trace!("{}: EventLoopThread shutdown", self.name);
+        if let Err(e) = self.handle.shutdown() {
+            debug!("{}: initiating shutdown failed: {:?}", self.name, e);
+        }
+        let thread = self.thread.take().expect("event loop thread");
+        if let Err(e) = thread.join() {
+            error!("{}: EventLoopThread failed: {:?}", self.name, e);
+        }
+        trace!("{}: EventLoopThread shutdown done", self.name);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use serde_derive::{Deserialize, Serialize};
+
+    use super::*;
+
+    #[derive(Debug, Serialize, Deserialize, PartialEq)]
+    enum TestServerMessage {
+        TestRequest,
+    }
+    impl AssociateHandleForMessage for TestServerMessage {}
+
+    struct TestServerImpl {}
+
+    impl Server for TestServerImpl {
+        type ServerMessage = TestServerMessage;
+        type ClientMessage = TestClientMessage;
+
+        fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage {
+            assert_eq!(req, TestServerMessage::TestRequest);
+            TestClientMessage::TestResponse
+        }
+    }
+
+    #[derive(Debug, Serialize, Deserialize, PartialEq)]
+    enum TestClientMessage {
+        TestResponse,
+    }
+
+    impl AssociateHandleForMessage for TestClientMessage {}
+
+    struct TestClientImpl {}
+
+    impl Client for TestClientImpl {
+        type ServerMessage = TestServerMessage;
+        type ClientMessage = TestClientMessage;
+    }
+
+    fn init() {
+        let _ = env_logger::builder().is_test(true).try_init();
+    }
+
+    fn setup() -> (
+        EventLoopThread,
+        EventLoopThread,
+        Proxy<TestServerMessage, TestClientMessage>,
+    ) {
+        // Server setup and registration.
+        let server = EventLoopThread::new("test-server".to_string(), None, || {}, || {})
+            .expect("server EventLoopThread");
+        let server_handle = server.handle();
+
+        let (server_pipe, client_pipe) = sys::make_pipe_pair().expect("server make_pipe_pair");
+        server_handle
+            .bind_server(TestServerImpl {}, server_pipe)
+            .expect("server bind_server");
+
+        // Client setup and registration.
+        let client = EventLoopThread::new("test-client".to_string(), None, || {}, || {})
+            .expect("client EventLoopThread");
+        let client_handle = client.handle();
+
+        let client_pipe = unsafe { sys::Pipe::from_raw_handle(client_pipe) };
+        let client_proxy = client_handle
+            .bind_client::<TestClientImpl>(client_pipe)
+            .expect("client bind_client");
+
+        (server, client, client_proxy)
+    }
+
+    // Verify basic EventLoopThread functionality works.  Create a server and client EventLoopThread, then send
+    // a single message from the client to the server and wait for the expected response.
+    #[test]
+    fn basic() {
+        init();
+        let (server, client, client_proxy) = setup();
+
+        // RPC message from client to server.
+        let response = client_proxy.call(TestServerMessage::TestRequest);
+        let response = response.expect("client response");
+        assert_eq!(response, TestClientMessage::TestResponse);
+
+        // Explicit shutdown.
+        drop(client);
+        drop(server);
+    }
+
+    // Same as `basic`, but shut down server before client.
+    #[test]
+    fn basic_reverse_drop_order() {
+        init();
+        let (server, client, client_proxy) = setup();
+
+        // RPC message from client to server.
+        let response = client_proxy.call(TestServerMessage::TestRequest);
+        let response = response.expect("client response");
+        assert_eq!(response, TestClientMessage::TestResponse);
+
+        // Explicit shutdown.
+        drop(server);
+        drop(client);
+    }
+
+    #[test]
+    fn dead_server() {
+        init();
+        let (server, _client, client_proxy) = setup();
+        drop(server);
+
+        let response = client_proxy.call(TestServerMessage::TestRequest);
+        response.expect_err("sending on closed channel");
+    }
+
+    #[test]
+    fn dead_client() {
+        init();
+        let (_server, client, client_proxy) = setup();
+        drop(client);
+
+        let response = client_proxy.call(TestServerMessage::TestRequest);
+        response.expect_err("sending on a closed channel");
+    }
+
+    #[test]
+    fn disconnected_handle() {
+        init();
+        let server = EventLoopThread::new("test-server".to_string(), None, || {}, || {})
+            .expect("server EventLoopThread");
+        let server_handle = server.handle().clone();
+        drop(server);
+
+        server_handle
+            .shutdown()
+            .expect_err("sending on closed channel");
+    }
+
+    #[test]
+    fn clone_after_drop() {
+        init();
+        let (server, client, client_proxy) = setup();
+        drop(server);
+        drop(client);
+
+        let clone = client_proxy.clone();
+        let response = clone.call(TestServerMessage::TestRequest);
+        response.expect_err("sending to a dropped ClientHandler");
+    }
+
+    #[test]
+    fn basic_event_loop_thread_callbacks() {
+        init();
+        let (start_tx, start_rx) = mpsc::channel();
+        let (stop_tx, stop_rx) = mpsc::channel();
+
+        let elt = EventLoopThread::new(
+            "test-thread-callbacks".to_string(),
+            None,
+            move || start_tx.send(()).unwrap(),
+            move || stop_tx.send(()).unwrap(),
+        )
+        .expect("server EventLoopThread");
+
+        start_rx.recv().expect("after_start callback done");
+
+        drop(elt);
+
+        stop_rx.recv().expect("before_stop callback done");
+    }
+}
diff --git a/third_party/rust/audioipc2/src/lib.rs b/third_party/rust/audioipc2/src/lib.rs
new file mode 100644
index 0000000000..eb6d843ba4
--- /dev/null
+++ b/third_party/rust/audioipc2/src/lib.rs
@@ -0,0 +1,214 @@
+// 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;
+
+pub mod codec;
+#[allow(deprecated)]
+pub mod errors;
+pub mod messages;
+pub mod shm;
+
+pub mod ipccore;
+pub mod rpccore;
+pub mod sys;
+
+pub use crate::messages::{ClientMessage, ServerMessage};
+
+#[cfg(unix)]
+use std::os::unix::io::IntoRawFd;
+#[cfg(windows)]
+use std::os::windows::io::IntoRawHandle;
+
+use std::io::Result;
+
+// 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> {
+        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> {
+        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, FALSE};
+#[cfg(windows)]
+use winapi::um::{handleapi, processthreadsapi, winnt};
+
+// Duplicate `source_handle` to `target_pid`.  Returns the value of the new handle inside the target process.
+// If `target_pid` is `None`, `source_handle` is duplicated in the current process.
+#[cfg(windows)]
+pub(crate) unsafe fn duplicate_platform_handle(
+    source_handle: PlatformHandleType,
+    target_pid: Option<DWORD>,
+) -> Result<PlatformHandleType> {
+    let source_process = processthreadsapi::GetCurrentProcess();
+    let target_process = 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",
+            ));
+        }
+        Some(target)
+    } else {
+        None
+    };
+
+    let mut target_handle = std::ptr::null_mut();
+    let ok = handleapi::DuplicateHandle(
+        source_process,
+        source_handle,
+        target_process.unwrap_or(source_process),
+        &mut target_handle,
+        0,
+        FALSE,
+        winnt::DUPLICATE_SAME_ACCESS,
+    );
+    if let Some(target) = target_process {
+        handleapi::CloseHandle(target);
+    };
+    if ok == FALSE {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "DuplicateHandle failed",
+        ));
+    }
+    Ok(target_handle)
+}
+
+// Close `target_handle_to_close` inside target process `target_pid` using a
+// special invocation of `DuplicateHandle`. See
+// https://docs.microsoft.com/en-us/windows/win32/api/handleapi/nf-handleapi-duplicatehandle#:~:text=Normally%20the%20target,dwOptions%20to%20DUPLICATE_CLOSE_SOURCE.
+#[cfg(windows)]
+pub(crate) unsafe fn close_target_handle(
+    target_handle_to_close: PlatformHandleType,
+    target_pid: DWORD,
+) -> Result<()> {
+    let target_process =
+        processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, target_pid);
+    if !valid_handle(target_process) {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "invalid target process",
+        ));
+    }
+
+    let ok = handleapi::DuplicateHandle(
+        target_process,
+        target_handle_to_close,
+        std::ptr::null_mut(),
+        std::ptr::null_mut(),
+        0,
+        FALSE,
+        winnt::DUPLICATE_CLOSE_SOURCE,
+    );
+    handleapi::CloseHandle(target_process);
+    if ok == FALSE {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "DuplicateHandle failed",
+        ));
+    }
+    Ok(())
+}
+
+#[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/audioipc2/src/messages.rs b/third_party/rust/audioipc2/src/messages.rs
new file mode 100644
index 0000000000..853f056bae
--- /dev/null
+++ b/third_party/rust/audioipc2/src/messages.rs
@@ -0,0 +1,632 @@
+// 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;
+use crate::INVALID_HANDLE_VALUE;
+#[cfg(target_os = "linux")]
+use audio_thread_priority::RtPriorityThreadInfo;
+use cubeb::{self, ffi};
+use serde_derive::Deserialize;
+use serde_derive::Serialize;
+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 {
+    #[serde(with = "serde_bytes")]
+    pub output_name: Option<Vec<u8>>,
+    #[serde(with = "serde_bytes")]
+    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,
+    #[serde(with = "serde_bytes")]
+    pub device_id: Option<Vec<u8>>,
+    #[serde(with = "serde_bytes")]
+    pub friendly_name: Option<Vec<u8>>,
+    #[serde(with = "serde_bytes")]
+    pub group_id: Option<Vec<u8>>,
+    #[serde(with = "serde_bytes")]
+    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 From<&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 {
+    #[serde(with = "serde_bytes")]
+    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 shm_handle: SerializableHandle,
+    pub shm_area_size: usize,
+}
+
+#[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(SerializableHandle),
+    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,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum CallbackResp {
+    Data(isize),
+    State,
+    DeviceChange,
+    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 platform 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 as follows:
+//
+// 1. Created in the initial state `Owned`, with a valid `target_pid`.
+// 2. Ownership is transferred out for processing during IPC send, becoming `Empty` temporarily.
+//    See `AssociateHandleForMessage::take_handle`.
+//    - Windows: DuplicateHandle transfers the handle to the remote process.
+//               This produces a new handle value in the local process representing the remote handle.
+//               This value must be sent to the remote, so `AssociateHandleForMessage::set_remote_handle`
+//               is used to transform the handle into a `SerializableValue`.
+//    - Unix: sendmsg transfers the handle to the remote process.  The handle is left `Empty`.
+//            (Note: this occurs later, when the serialized message buffer is sent)
+// 3. Message containing `SerializableValue` or `Empty` (depending on handle processing in step 2)
+//    is serialized and sent via IPC.
+// 4. Message received and deserialized in target process.
+//    - Windows: `AssociateHandleForMessage::set_local_handle converts the received `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 `AssociateHandleForMessage::set_local_handle`.
+#[derive(Debug)]
+pub enum SerializableHandle {
+    // Owned handle, with optional target_pid on sending side.
+    Owned(PlatformHandle, Option<u32>),
+    // Transitional IPC states:
+    SerializableValue(PlatformHandleType), // Windows
+    Empty,                                 // Unix
+}
+
+// PlatformHandle is non-Send and contains 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))
+    }
+
+    // Called on the receiving side to take ownership of the handle.
+    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"),
+        }
+    }
+
+    // Called on the sending side to take ownership of the handle for
+    // handling platform-specific remoting.
+    fn take_handle_for_send(&mut self) -> RemoteHandle {
+        match std::mem::replace(self, SerializableHandle::Empty) {
+            SerializableHandle::Owned(handle, target_pid) => unsafe {
+                RemoteHandle::new(
+                    handle.into_raw(),
+                    target_pid.expect("target process required"),
+                )
+            },
+            _ => panic!("take_handle_for_send called in invalid state"),
+        }
+    }
+
+    fn new_owned(handle: PlatformHandleType) -> SerializableHandle {
+        SerializableHandle::Owned(PlatformHandle::new(handle), None)
+    }
+
+    #[cfg(windows)]
+    fn make_owned(&mut self) {
+        if let SerializableHandle::SerializableValue(handle) = self {
+            *self = SerializableHandle::new_owned(*handle);
+        } else {
+            panic!("make_owned called in invalid state")
+        }
+    }
+
+    fn new_serializable_value(handle: PlatformHandleType) -> SerializableHandle {
+        SerializableHandle::SerializableValue(handle)
+    }
+
+    fn get_serializable_value(&self) -> PlatformHandleType {
+        match *self {
+            SerializableHandle::SerializableValue(handle) => handle,
+            SerializableHandle::Empty => INVALID_HANDLE_VALUE,
+            _ => 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 serde::de::Visitor<'_> 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,
+    {
+        Ok(SerializableHandle::new_serializable_value(
+            value as PlatformHandleType,
+        ))
+    }
+}
+
+// Represents a PlatformHandle in-flight between processes.
+// On Unix platforms, this is just a plain owned Handle, closed on drop.
+// On Windows, `RemoteHandle` also retains ownership of the `target_handle`
+// in the `target` process.  Once the handle has been successfully sent
+// to the remote, the sender should call `mark_sent()` to relinquish
+// ownership of `target_handle` in the remote.
+#[derive(Debug)]
+pub struct RemoteHandle {
+    pub(crate) handle: PlatformHandleType,
+    #[cfg(windows)]
+    pub(crate) target: u32,
+    #[cfg(windows)]
+    pub(crate) target_handle: Option<PlatformHandleType>,
+}
+
+impl RemoteHandle {
+    #[allow(clippy::missing_safety_doc)]
+    pub unsafe fn new(handle: PlatformHandleType, _target: u32) -> Self {
+        RemoteHandle {
+            handle,
+            #[cfg(windows)]
+            target: _target,
+            #[cfg(windows)]
+            target_handle: None,
+        }
+    }
+
+    #[cfg(windows)]
+    pub fn mark_sent(&mut self) {
+        self.target_handle.take();
+    }
+
+    #[cfg(windows)]
+    #[allow(clippy::missing_safety_doc)]
+    pub unsafe fn send_to_target(&mut self) -> std::io::Result<PlatformHandleType> {
+        let target_handle = crate::duplicate_platform_handle(self.handle, Some(self.target))?;
+        self.target_handle = Some(target_handle);
+        Ok(target_handle)
+    }
+
+    #[cfg(unix)]
+    #[allow(clippy::missing_safety_doc)]
+    pub unsafe fn take(self) -> PlatformHandleType {
+        let h = self.handle;
+        std::mem::forget(self);
+        h
+    }
+}
+
+impl Drop for RemoteHandle {
+    fn drop(&mut self) {
+        unsafe {
+            crate::close_platform_handle(self.handle);
+        }
+        #[cfg(windows)]
+        unsafe {
+            if let Some(target_handle) = self.target_handle {
+                if let Err(e) = crate::close_target_handle(target_handle, self.target) {
+                    trace!("RemoteHandle failed to close target handle: {:?}", e);
+                }
+            }
+        }
+    }
+}
+
+unsafe impl Send for RemoteHandle {}
+
+pub trait AssociateHandleForMessage {
+    // True if this item has an associated handle attached for remoting.
+    fn has_associated_handle(&self) -> bool {
+        false
+    }
+
+    // Take ownership of the associated handle, leaving the item's
+    // associated handle empty.
+    fn take_handle(&mut self) -> RemoteHandle {
+        panic!("take_handle called on item without associated handle");
+    }
+
+    #[allow(clippy::missing_safety_doc)]
+    // Replace an empty associated handle with a non-owning serializable value
+    // indicating the value of the handle in the remote process.
+    #[cfg(windows)]
+    unsafe fn set_remote_handle(&mut self, _: PlatformHandleType) {
+        panic!("set_remote_handle called on item without associated handle");
+    }
+
+    #[allow(clippy::missing_safety_doc)]
+    // Replace a serialized associated handle value with an owned local handle.
+    #[cfg(windows)]
+    unsafe fn set_local_handle(&mut self) {
+        panic!("set_local_handle called on item without associated handle");
+    }
+
+    #[allow(clippy::missing_safety_doc)]
+    // Replace an empty associated handle with an owned local handle.
+    #[cfg(unix)]
+    unsafe fn set_local_handle(&mut self, _: PlatformHandleType) {
+        panic!("set_local_handle called on item without associated handle");
+    }
+}
+
+impl AssociateHandleForMessage for ClientMessage {
+    fn has_associated_handle(&self) -> bool {
+        matches!(
+            *self,
+            ClientMessage::StreamCreated(_)
+                | ClientMessage::StreamInitialized(_)
+                | ClientMessage::ContextSetupDeviceCollectionCallback(_)
+        )
+    }
+
+    fn take_handle(&mut self) -> RemoteHandle {
+        match *self {
+            ClientMessage::StreamCreated(ref mut data) => data.shm_handle.take_handle_for_send(),
+            ClientMessage::StreamInitialized(ref mut data) => data.take_handle_for_send(),
+            ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => {
+                data.platform_handle.take_handle_for_send()
+            }
+            _ => panic!("take_handle called on item without associated handle"),
+        }
+    }
+
+    #[cfg(windows)]
+    unsafe fn set_remote_handle(&mut self, handle: PlatformHandleType) {
+        match *self {
+            ClientMessage::StreamCreated(ref mut data) => {
+                data.shm_handle = SerializableHandle::new_serializable_value(handle);
+            }
+            ClientMessage::StreamInitialized(ref mut data) => {
+                *data = SerializableHandle::new_serializable_value(handle);
+            }
+            ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => {
+                data.platform_handle = SerializableHandle::new_serializable_value(handle);
+            }
+            _ => panic!("set_remote_handle called on item without associated handle"),
+        }
+    }
+
+    #[cfg(windows)]
+    unsafe fn set_local_handle(&mut self) {
+        match *self {
+            ClientMessage::StreamCreated(ref mut data) => data.shm_handle.make_owned(),
+            ClientMessage::StreamInitialized(ref mut data) => data.make_owned(),
+            ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => {
+                data.platform_handle.make_owned()
+            }
+            _ => panic!("set_local_handle called on item without associated handle"),
+        }
+    }
+
+    #[cfg(unix)]
+    unsafe fn set_local_handle(&mut self, handle: PlatformHandleType) {
+        match *self {
+            ClientMessage::StreamCreated(ref mut data) => {
+                data.shm_handle = SerializableHandle::new_owned(handle);
+            }
+            ClientMessage::StreamInitialized(ref mut data) => {
+                *data = SerializableHandle::new_owned(handle);
+            }
+            ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => {
+                data.platform_handle = SerializableHandle::new_owned(handle);
+            }
+            _ => panic!("set_local_handle called on item without associated handle"),
+        }
+    }
+}
+
+impl AssociateHandleForMessage for ServerMessage {}
+
+impl AssociateHandleForMessage for DeviceCollectionReq {}
+impl AssociateHandleForMessage for DeviceCollectionResp {}
+
+impl AssociateHandleForMessage for CallbackReq {}
+impl AssociateHandleForMessage 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/audioipc2/src/rpccore.rs b/third_party/rust/audioipc2/src/rpccore.rs
new file mode 100644
index 0000000000..69ed2e8cf0
--- /dev/null
+++ b/third_party/rust/audioipc2/src/rpccore.rs
@@ -0,0 +1,470 @@
+// Copyright © 2021 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crossbeam_queue::ArrayQueue;
+use mio::Token;
+use std::cell::UnsafeCell;
+use std::collections::VecDeque;
+use std::io::{self, Error, ErrorKind, Result};
+use std::marker::PhantomPinned;
+use std::mem::ManuallyDrop;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Weak};
+
+use crate::ipccore::EventLoopHandle;
+
+// This provides a safe-ish method for a thread to allocate
+// stack storage space for a result, then pass a (wrapped)
+// pointer to that location to another thread via
+// a CompletionWriter to eventually store a result into.
+struct Completion<T> {
+    item: UnsafeCell<Option<T>>,
+    writer: AtomicBool,
+    _pin: PhantomPinned, // disable rustc's no-alias
+}
+
+impl<T> Completion<T> {
+    fn new() -> Self {
+        Completion {
+            item: UnsafeCell::new(None),
+            writer: AtomicBool::new(false),
+            _pin: PhantomPinned,
+        }
+    }
+
+    // Wait until the writer completes, then return the result.
+    // This is intended to be a single-use function, once the writer
+    // has completed any further attempts to wait will return None.
+    fn wait(&self) -> Option<T> {
+        // Wait for the writer to complete or be dropped.
+        while self.writer.load(Ordering::Acquire) {
+            std::thread::park();
+        }
+        unsafe { (*self.item.get()).take() }
+    }
+
+    // Create a writer for the other thread to store the
+    // expected result into.
+    fn writer(&self) -> CompletionWriter<T> {
+        assert!(!self.writer.load(Ordering::Relaxed));
+        self.writer.store(true, Ordering::Release);
+        CompletionWriter {
+            ptr: self as *const _ as *mut _,
+            waiter: std::thread::current(),
+        }
+    }
+}
+
+impl<T> Drop for Completion<T> {
+    fn drop(&mut self) {
+        // Wait for the outstanding writer to complete before
+        // dropping, since the CompletionWriter references
+        // memory owned by this object.
+        while self.writer.load(Ordering::Acquire) {
+            std::thread::park();
+        }
+    }
+}
+
+struct CompletionWriter<T> {
+    ptr: *mut Completion<T>, // Points to a Completion on another thread's stack
+    waiter: std::thread::Thread, // Identifies thread waiting for completion
+}
+
+impl<T> CompletionWriter<T> {
+    fn set(self, value: T) {
+        // Store the result into the Completion's memory.
+        // Since `set` consumes `self`, rely on `Drop` to
+        // mark the writer as done and wake the Completion's
+        // thread.
+        unsafe {
+            assert!((*self.ptr).writer.load(Ordering::Relaxed));
+            *(*self.ptr).item.get() = Some(value);
+        }
+    }
+}
+
+impl<T> Drop for CompletionWriter<T> {
+    fn drop(&mut self) {
+        // Mark writer as complete - if `set` was not called,
+        // the waiter will receive `None`.
+        unsafe {
+            (*self.ptr).writer.store(false, Ordering::Release);
+        }
+        // Wake the Completion's thread.
+        self.waiter.unpark();
+    }
+}
+
+// Safety: CompletionWriter holds a pointer to a Completion
+// residing on another thread's stack.  The Completion always
+// waits for an outstanding writer if present, and CompletionWriter
+// releases the waiter and wakes the Completion's thread on drop,
+// so this pointer will always be live for the duration of a
+// CompletionWriter.
+unsafe impl<T> Send for CompletionWriter<T> {}
+
+// RPC message handler.  Implemented by ClientHandler (for Client)
+// and ServerHandler (for Server).
+pub(crate) trait Handler {
+    type In;
+    type Out;
+
+    // Consume a request
+    fn consume(&mut self, request: Self::In) -> Result<()>;
+
+    // Produce a response
+    fn produce(&mut self) -> Result<Option<Self::Out>>;
+}
+
+// Client RPC definition.  This supplies the expected message
+// request and response types.
+pub trait Client {
+    type ServerMessage;
+    type ClientMessage;
+}
+
+// Server RPC definition.  This supplies the expected message
+// request and response types.  `process` is passed inbound RPC
+// requests by the ServerHandler to be responded to by the server.
+pub trait Server {
+    type ServerMessage;
+    type ClientMessage;
+
+    fn process(&mut self, req: Self::ServerMessage) -> Self::ClientMessage;
+}
+
+// RPC Client Proxy implementation.
+type ProxyRequest<Request, Response> = (Request, CompletionWriter<Response>);
+
+// RPC Proxy that may be `clone`d for use by multiple owners/threads.
+// A Proxy `call` arranges for the supplied request to be transmitted
+// to the associated Server via RPC and blocks awaiting the response
+// via the associated `Completion`.
+// A ClientHandler normally lives until the last Proxy is dropped, but if the ClientHandler
+// encounters an internal error, `requests` will fail to upgrade, allowing
+// the proxy to report an error.
+#[derive(Debug)]
+pub struct Proxy<Request, Response> {
+    handle: Option<(EventLoopHandle, Token)>,
+    requests: ManuallyDrop<RequestQueueSender<ProxyRequest<Request, Response>>>,
+}
+
+impl<Request, Response> Proxy<Request, Response> {
+    fn new(requests: RequestQueueSender<ProxyRequest<Request, Response>>) -> Self {
+        Self {
+            handle: None,
+            requests: ManuallyDrop::new(requests),
+        }
+    }
+
+    pub fn call(&self, request: Request) -> Result<Response> {
+        let response = Completion::new();
+        self.requests.push((request, response.writer()))?;
+        self.wake_connection();
+        match response.wait() {
+            Some(resp) => Ok(resp),
+            None => Err(Error::new(ErrorKind::Other, "proxy recv error")),
+        }
+    }
+
+    pub(crate) fn connect_event_loop(&mut self, handle: EventLoopHandle, token: Token) {
+        self.handle = Some((handle, token));
+    }
+
+    fn wake_connection(&self) {
+        let (handle, token) = self
+            .handle
+            .as_ref()
+            .expect("proxy not connected to event loop");
+        handle.wake_connection(*token);
+    }
+}
+
+impl<Request, Response> Clone for Proxy<Request, Response> {
+    fn clone(&self) -> Self {
+        let mut clone = Self::new((*self.requests).clone());
+        let (handle, token) = self
+            .handle
+            .as_ref()
+            .expect("proxy not connected to event loop");
+        clone.connect_event_loop(handle.clone(), *token);
+        clone
+    }
+}
+
+impl<Request, Response> Drop for Proxy<Request, Response> {
+    fn drop(&mut self) {
+        trace!("Proxy drop, waking EventLoop");
+        // Must drop `requests` before waking the connection, otherwise
+        // the wake may be processed before the (last) weak reference is
+        // dropped.
+        let last_proxy = self.requests.live_proxies();
+        unsafe {
+            ManuallyDrop::drop(&mut self.requests);
+        }
+        if last_proxy == 1 && self.handle.is_some() {
+            self.wake_connection()
+        }
+    }
+}
+
+const RPC_CLIENT_INITIAL_PROXIES: usize = 32; // Initial proxy pre-allocation per client.
+
+// Client-specific Handler implementation.
+// The IPC EventLoop Driver calls this to execute client-specific
+// RPC handling.  Serialized messages sent via a Proxy are queued
+// for transmission when `produce` is called.
+// Deserialized messages are passed via `consume` to
+// trigger response completion by sending the response via a channel
+// connected to a ProxyResponse.
+pub(crate) struct ClientHandler<C: Client> {
+    in_flight: VecDeque<CompletionWriter<C::ClientMessage>>,
+    requests: Arc<RequestQueue<ProxyRequest<C::ServerMessage, C::ClientMessage>>>,
+}
+
+impl<C: Client> ClientHandler<C> {
+    fn new(
+        requests: Arc<RequestQueue<ProxyRequest<C::ServerMessage, C::ClientMessage>>>,
+    ) -> ClientHandler<C> {
+        ClientHandler::<C> {
+            in_flight: VecDeque::with_capacity(RPC_CLIENT_INITIAL_PROXIES),
+            requests,
+        }
+    }
+}
+
+impl<C: Client> Handler for ClientHandler<C> {
+    type In = C::ClientMessage;
+    type Out = C::ServerMessage;
+
+    fn consume(&mut self, response: Self::In) -> Result<()> {
+        trace!("ClientHandler::consume");
+        if let Some(response_writer) = self.in_flight.pop_front() {
+            response_writer.set(response);
+        } else {
+            return Err(Error::new(ErrorKind::Other, "request/response mismatch"));
+        }
+
+        Ok(())
+    }
+
+    fn produce(&mut self) -> Result<Option<Self::Out>> {
+        trace!("ClientHandler::produce");
+
+        // If the weak count is zero, no proxies are attached and
+        // no further proxies can be attached since every proxy
+        // after the initial one is cloned from an existing instance.
+        self.requests.check_live_proxies()?;
+        // Try to get a new message
+        match self.requests.pop() {
+            Some((request, response_writer)) => {
+                trace!("  --> received request");
+                self.in_flight.push_back(response_writer);
+                Ok(Some(request))
+            }
+            None => {
+                trace!("  --> no request");
+                Ok(None)
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct RequestQueue<T> {
+    queue: ArrayQueue<T>,
+}
+
+impl<T> RequestQueue<T> {
+    pub(crate) fn new(size: usize) -> Self {
+        RequestQueue {
+            queue: ArrayQueue::new(size),
+        }
+    }
+
+    pub(crate) fn pop(&self) -> Option<T> {
+        self.queue.pop()
+    }
+
+    pub(crate) fn new_sender(self: &Arc<Self>) -> RequestQueueSender<T> {
+        RequestQueueSender {
+            inner: Arc::downgrade(self),
+        }
+    }
+
+    pub(crate) fn check_live_proxies(self: &Arc<Self>) -> Result<()> {
+        if Arc::weak_count(self) == 0 {
+            return Err(io::ErrorKind::ConnectionAborted.into());
+        }
+        Ok(())
+    }
+}
+
+pub(crate) struct RequestQueueSender<T> {
+    inner: Weak<RequestQueue<T>>,
+}
+
+impl<T> RequestQueueSender<T> {
+    pub(crate) fn push(&self, request: T) -> Result<()> {
+        if let Some(consumer) = self.inner.upgrade() {
+            if consumer.queue.push(request).is_err() {
+                debug!("Proxy[{:p}]: call failed - CH::requests full", self);
+                return Err(io::ErrorKind::ConnectionAborted.into());
+            }
+            return Ok(());
+        }
+        debug!("Proxy[{:p}]: call failed - CH::requests dropped", self);
+        Err(Error::new(ErrorKind::Other, "proxy send error"))
+    }
+
+    pub(crate) fn live_proxies(&self) -> usize {
+        Weak::weak_count(&self.inner)
+    }
+}
+
+impl<T> Clone for RequestQueueSender<T> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<T> std::fmt::Debug for RequestQueueSender<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RequestQueueProducer")
+            .field("inner", &self.inner.as_ptr())
+            .finish()
+    }
+}
+
+#[allow(clippy::type_complexity)]
+pub(crate) fn make_client<C: Client>(
+) -> Result<(ClientHandler<C>, Proxy<C::ServerMessage, C::ClientMessage>)> {
+    let requests = Arc::new(RequestQueue::new(RPC_CLIENT_INITIAL_PROXIES));
+    let proxy_req = requests.new_sender();
+    let handler = ClientHandler::new(requests);
+
+    Ok((handler, Proxy::new(proxy_req)))
+}
+
+// Server-specific Handler implementation.
+// The IPC EventLoop Driver calls this to execute server-specific
+// RPC handling.  Deserialized messages are passed via `consume` to the
+// associated `server` for processing.  Server responses are then queued
+// for RPC to the associated client when `produce` is called.
+pub(crate) struct ServerHandler<S: Server> {
+    server: S,
+    in_flight: VecDeque<S::ClientMessage>,
+}
+
+impl<S: Server> Handler for ServerHandler<S> {
+    type In = S::ServerMessage;
+    type Out = S::ClientMessage;
+
+    fn consume(&mut self, message: Self::In) -> Result<()> {
+        trace!("ServerHandler::consume");
+        let response = self.server.process(message);
+        self.in_flight.push_back(response);
+        Ok(())
+    }
+
+    fn produce(&mut self) -> Result<Option<Self::Out>> {
+        trace!("ServerHandler::produce");
+
+        // Return the ready response
+        match self.in_flight.pop_front() {
+            Some(res) => {
+                trace!("  --> received response");
+                Ok(Some(res))
+            }
+            None => {
+                trace!("  --> no response ready");
+                Ok(None)
+            }
+        }
+    }
+}
+
+const RPC_SERVER_INITIAL_CLIENTS: usize = 32; // Initial client allocation per server.
+
+pub(crate) fn make_server<S: Server>(server: S) -> ServerHandler<S> {
+    ServerHandler::<S> {
+        server,
+        in_flight: VecDeque::with_capacity(RPC_SERVER_INITIAL_CLIENTS),
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn basic() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        assert!(queue.pop().is_none());
+        producer.push(1).unwrap();
+        assert!(queue.pop().is_some());
+        assert!(queue.pop().is_none());
+    }
+
+    #[test]
+    fn queue_dropped() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        drop(queue);
+        assert!(producer.push(1).is_err());
+    }
+
+    #[test]
+    fn queue_full() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        producer.push(1).unwrap();
+        assert!(producer.push(2).is_err());
+    }
+
+    #[test]
+    fn queue_producer_clone() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        let producer2 = producer.clone();
+        producer.push(1).unwrap();
+        assert!(producer2.push(2).is_err());
+    }
+
+    #[test]
+    fn queue_producer_drop() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        let producer2 = producer.clone();
+        drop(producer);
+        assert!(producer2.push(2).is_ok());
+    }
+
+    #[test]
+    fn queue_producer_weak() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        let producer2 = producer.clone();
+        drop(queue);
+        assert!(producer2.push(2).is_err());
+    }
+
+    #[test]
+    fn queue_producer_shutdown() {
+        let queue = Arc::new(RequestQueue::new(1));
+        let producer = queue.new_sender();
+        let producer2 = producer.clone();
+        producer.push(1).unwrap();
+        assert!(Arc::weak_count(&queue) == 2);
+        drop(producer);
+        assert!(Arc::weak_count(&queue) == 1);
+        drop(producer2);
+        assert!(Arc::weak_count(&queue) == 0);
+    }
+}
diff --git a/third_party/rust/audioipc2/src/shm.rs b/third_party/rust/audioipc2/src/shm.rs
new file mode 100644
index 0000000000..2aaa92ef36
--- /dev/null
+++ b/third_party/rust/audioipc2/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)]
+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, &params) } == 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().len(size).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().len(size).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 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)
+        }
+
+        pub fn get_size(&self) -> usize {
+            self.view.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 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)
+        }
+
+        pub fn get_size(&self) -> usize {
+            self.view.size
+        }
+    }
+}
diff --git a/third_party/rust/audioipc2/src/sys/mod.rs b/third_party/rust/audioipc2/src/sys/mod.rs
new file mode 100644
index 0000000000..0bcfdaa15e
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/mod.rs
@@ -0,0 +1,77 @@
+// Copyright © 2021 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use std::{collections::VecDeque, io::Result};
+
+use bytes::BytesMut;
+use mio::{event::Source, Interest, Registry, Token};
+
+#[cfg(unix)]
+mod unix;
+use crate::messages::RemoteHandle;
+
+#[cfg(unix)]
+pub use self::unix::*;
+
+#[cfg(windows)]
+mod windows;
+#[cfg(windows)]
+pub use self::windows::*;
+
+impl Source for Pipe {
+    fn register(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> {
+        self.io.register(registry, token, interests)
+    }
+
+    fn reregister(&mut self, registry: &Registry, token: Token, interests: Interest) -> Result<()> {
+        self.io.reregister(registry, token, interests)
+    }
+
+    fn deregister(&mut self, registry: &Registry) -> Result<()> {
+        self.io.deregister(registry)
+    }
+}
+
+const HANDLE_QUEUE_LIMIT: usize = 16;
+
+#[derive(Debug)]
+pub struct ConnectionBuffer {
+    pub buf: BytesMut,
+    handles: VecDeque<RemoteHandle>,
+}
+
+impl ConnectionBuffer {
+    pub fn with_capacity(cap: usize) -> Self {
+        ConnectionBuffer {
+            buf: BytesMut::with_capacity(cap),
+            handles: VecDeque::with_capacity(HANDLE_QUEUE_LIMIT),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.buf.is_empty()
+    }
+
+    pub fn push_handle(&mut self, handle: RemoteHandle) {
+        assert!(self.handles.len() < self.handles.capacity());
+        self.handles.push_back(handle)
+    }
+
+    pub fn pop_handle(&mut self) -> Option<RemoteHandle> {
+        self.handles.pop_front()
+    }
+}
+
+pub trait RecvMsg {
+    // Receive data from the associated connection.  `recv_msg` expects the capacity of
+    // the `ConnectionBuffer` members have been adjusted appropriately by the caller.
+    fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize>;
+}
+
+pub trait SendMsg {
+    // Send data on the associated connection.  `send_msg` consumes and adjusts the length of the
+    // `ConnectionBuffer` members based on the size of the successful send operation.
+    fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize>;
+}
diff --git a/third_party/rust/audioipc2/src/sys/unix/cmsg.rs b/third_party/rust/audioipc2/src/sys/unix/cmsg.rs
new file mode 100644
index 0000000000..581df9a847
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/unix/cmsg.rs
@@ -0,0 +1,104 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::sys::HANDLE_QUEUE_LIMIT;
+use bytes::{BufMut, BytesMut};
+use libc::{self, cmsghdr};
+use std::convert::TryInto;
+use std::os::unix::io::RawFd;
+use std::{mem, slice};
+
+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) }
+    }
+}
+
+// Encode `handles` into a cmsghdr in `buf`.
+pub fn encode_handles(cmsg: &mut BytesMut, handles: &[RawFd]) {
+    assert!(handles.len() <= HANDLE_QUEUE_LIMIT);
+    let msg = handles.as_bytes();
+
+    let cmsg_space = space(msg.len());
+    assert!(cmsg.remaining_mut() >= cmsg_space);
+
+    // 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: libc::SOL_SOCKET,
+        cmsg_type: libc::SCM_RIGHTS,
+        ..zeroed
+    };
+
+    unsafe {
+        let cmsghdr_ptr = cmsg.chunk_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);
+    }
+}
+
+// Decode `buf` containing a cmsghdr with one or more handle(s).
+pub fn decode_handles(buf: &mut BytesMut) -> arrayvec::ArrayVec<RawFd, HANDLE_QUEUE_LIMIT> {
+    let mut fds = arrayvec::ArrayVec::<RawFd, HANDLE_QUEUE_LIMIT>::new();
+
+    let cmsghdr_len = len(0);
+
+    if buf.len() < cmsghdr_len {
+        // No more entries---not enough data in `buf` for a
+        // complete message.
+        return fds;
+    }
+
+    let cmsg: &cmsghdr = unsafe { &*(buf.as_ptr() as *const _) };
+    #[allow(clippy::unnecessary_cast)] // `cmsg_len` type is platform-dependent.
+    let cmsg_len = cmsg.cmsg_len as usize;
+
+    match (cmsg.cmsg_level, cmsg.cmsg_type) {
+        (libc::SOL_SOCKET, libc::SCM_RIGHTS) => {
+            trace!("Found SCM_RIGHTS...");
+            let slice = &buf[cmsghdr_len..cmsg_len];
+            let slice = unsafe {
+                slice::from_raw_parts(
+                    slice.as_ptr() as *const _,
+                    slice.len() / mem::size_of::<i32>(),
+                )
+            };
+            fds.try_extend_from_slice(slice).unwrap();
+        }
+        (level, kind) => {
+            trace!("Skipping cmsg level, {}, type={}...", level, kind);
+        }
+    }
+
+    assert!(fds.len() <= HANDLE_QUEUE_LIMIT);
+    fds
+}
+
+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/audioipc2/src/sys/unix/cmsghdr.c b/third_party/rust/audioipc2/src/sys/unix/cmsghdr.c
new file mode 100644
index 0000000000..82d7852867
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/unix/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/audioipc2/src/sys/unix/mod.rs b/third_party/rust/audioipc2/src/sys/unix/mod.rs
new file mode 100644
index 0000000000..84f3f1edf2
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/unix/mod.rs
@@ -0,0 +1,126 @@
+// Copyright © 2021 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use std::io::Result;
+use std::os::unix::prelude::{AsRawFd, FromRawFd};
+
+use bytes::{Buf, BufMut, BytesMut};
+use iovec::IoVec;
+use mio::net::UnixStream;
+
+use crate::PlatformHandle;
+
+use super::{ConnectionBuffer, RecvMsg, SendMsg, HANDLE_QUEUE_LIMIT};
+
+pub mod cmsg;
+mod msg;
+
+pub struct Pipe {
+    pub(crate) io: UnixStream,
+    cmsg: BytesMut,
+}
+
+impl Pipe {
+    fn new(io: UnixStream) -> Self {
+        Pipe {
+            io,
+            cmsg: BytesMut::with_capacity(cmsg::space(
+                std::mem::size_of::<i32>() * HANDLE_QUEUE_LIMIT,
+            )),
+        }
+    }
+}
+
+// Create a connected "pipe" pair.  The `Pipe` is the server end,
+// the `PlatformHandle` is the client end to be remoted.
+pub fn make_pipe_pair() -> Result<(Pipe, PlatformHandle)> {
+    let (server, client) = UnixStream::pair()?;
+    Ok((Pipe::new(server), PlatformHandle::from(client)))
+}
+
+impl Pipe {
+    #[allow(clippy::missing_safety_doc)]
+    pub unsafe fn from_raw_handle(handle: crate::PlatformHandle) -> Pipe {
+        Pipe::new(UnixStream::from_raw_fd(handle.into_raw()))
+    }
+
+    pub fn shutdown(&mut self) -> Result<()> {
+        self.io.shutdown(std::net::Shutdown::Both)
+    }
+}
+
+impl RecvMsg for Pipe {
+    // Receive data (and fds) from the associated connection.  `recv_msg` expects the capacity of
+    // the `ConnectionBuffer` members has been adjusted appropriate by the caller.
+    fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> {
+        assert!(buf.buf.remaining_mut() > 0);
+        // TODO: MSG_CMSG_CLOEXEC not portable.
+        // TODO: MSG_NOSIGNAL not portable; macOS can set socket option SO_NOSIGPIPE instead.
+        #[cfg(target_os = "linux")]
+        let flags = libc::MSG_CMSG_CLOEXEC | libc::MSG_NOSIGNAL;
+        #[cfg(not(target_os = "linux"))]
+        let flags = 0;
+        let r = unsafe {
+            let chunk = buf.buf.chunk_mut();
+            let slice = std::slice::from_raw_parts_mut(chunk.as_mut_ptr(), chunk.len());
+            let mut iovec = [<&mut IoVec>::from(slice)];
+            msg::recv_msg_with_flags(
+                self.io.as_raw_fd(),
+                &mut iovec,
+                self.cmsg.chunk_mut(),
+                flags,
+            )
+        };
+        match r {
+            Ok((n, cmsg_n, msg_flags)) => unsafe {
+                trace!("recv_msg_with_flags flags={}", msg_flags);
+                buf.buf.advance_mut(n);
+                self.cmsg.advance_mut(cmsg_n);
+                let handles = cmsg::decode_handles(&mut self.cmsg);
+                self.cmsg.clear();
+                let unused = 0;
+                for h in handles {
+                    buf.push_handle(super::RemoteHandle::new(h, unused));
+                }
+                Ok(n)
+            },
+            Err(e) => Err(e),
+        }
+    }
+}
+
+impl SendMsg for Pipe {
+    // Send data (and fds) on the associated connection.  `send_msg` adjusts the length of the
+    // `ConnectionBuffer` members based on the size of the successful send operation.
+    fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> {
+        assert!(!buf.buf.is_empty());
+        if !buf.handles.is_empty() {
+            let mut handles = [-1i32; HANDLE_QUEUE_LIMIT];
+            for (i, h) in buf.handles.iter().enumerate() {
+                handles[i] = h.handle;
+            }
+            cmsg::encode_handles(&mut self.cmsg, &handles[..buf.handles.len()]);
+        }
+        let r = {
+            // TODO: MSG_NOSIGNAL not portable; macOS can set socket option SO_NOSIGPIPE instead.
+            #[cfg(target_os = "linux")]
+            let flags = libc::MSG_NOSIGNAL;
+            #[cfg(not(target_os = "linux"))]
+            let flags = 0;
+            let iovec = [<&IoVec>::from(&buf.buf[..buf.buf.len()])];
+            msg::send_msg_with_flags(self.io.as_raw_fd(), &iovec, &self.cmsg, flags)
+        };
+        match r {
+            Ok(n) => {
+                buf.buf.advance(n);
+                // Discard sent handles.
+                while buf.handles.pop_front().is_some() {}
+                self.cmsg.clear();
+                Ok(n)
+            }
+            Err(e) => Err(e),
+        }
+    }
+}
diff --git a/third_party/rust/audioipc2/src/sys/unix/msg.rs b/third_party/rust/audioipc2/src/sys/unix/msg.rs
new file mode 100644
index 0000000000..c2cd353289
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/unix/msg.rs
@@ -0,0 +1,82 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use bytes::buf::UninitSlice;
+use iovec::unix;
+use iovec::IoVec;
+use std::os::unix::io::RawFd;
+use std::{cmp, io, mem, ptr};
+
+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(crate) fn recv_msg_with_flags(
+    socket: RawFd,
+    bufs: &mut [&mut IoVec],
+    cmsg: &mut UninitSlice,
+    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.len() == 0 {
+        (ptr::null_mut(), 0)
+    } else {
+        (cmsg.as_mut_ptr() as _, 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) })?;
+
+    #[allow(clippy::unnecessary_cast)] // `msg_controllen` type is platform-dependent.
+    let controllen = msghdr.msg_controllen as usize;
+    Ok((n, controllen, msghdr.msg_flags))
+}
+
+pub(crate) 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/audioipc2/src/sys/windows/mod.rs b/third_party/rust/audioipc2/src/sys/windows/mod.rs
new file mode 100644
index 0000000000..973e9608d1
--- /dev/null
+++ b/third_party/rust/audioipc2/src/sys/windows/mod.rs
@@ -0,0 +1,102 @@
+// Copyright © 2021 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use std::{
+    fs::OpenOptions,
+    io::{Read, Write},
+    os::windows::prelude::{FromRawHandle, OpenOptionsExt},
+    sync::atomic::{AtomicUsize, Ordering},
+};
+
+use std::io::Result;
+
+use bytes::{Buf, BufMut};
+use mio::windows::NamedPipe;
+use winapi::um::winbase::FILE_FLAG_OVERLAPPED;
+
+use crate::PlatformHandle;
+
+use super::{ConnectionBuffer, RecvMsg, SendMsg};
+
+pub struct Pipe {
+    pub(crate) io: NamedPipe,
+}
+
+// Create a connected "pipe" pair.  The `Pipe` is the server end,
+// the `PlatformHandle` is the client end to be remoted.
+pub fn make_pipe_pair() -> Result<(Pipe, PlatformHandle)> {
+    let pipe_name = get_pipe_name();
+    let server = NamedPipe::new(&pipe_name)?;
+
+    let client = {
+        let mut opts = OpenOptions::new();
+        opts.read(true)
+            .write(true)
+            .custom_flags(FILE_FLAG_OVERLAPPED);
+        let file = opts.open(&pipe_name)?;
+        PlatformHandle::from(file)
+    };
+
+    Ok((Pipe::new(server), client))
+}
+
+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::Relaxed);
+    format!("\\\\.\\pipe\\LOCAL\\cubeb-pipe-{pid}-{pipe_id}")
+}
+
+impl Pipe {
+    pub fn new(io: NamedPipe) -> Self {
+        Self { io }
+    }
+
+    #[allow(clippy::missing_safety_doc)]
+    pub unsafe fn from_raw_handle(handle: crate::PlatformHandle) -> Pipe {
+        Pipe::new(NamedPipe::from_raw_handle(handle.into_raw()))
+    }
+
+    pub fn shutdown(&mut self) -> Result<()> {
+        self.io.disconnect()
+    }
+}
+
+impl RecvMsg for Pipe {
+    // Receive data from the associated connection.  `recv_msg` expects the capacity of
+    // the `ConnectionBuffer` members has been adjusted appropriate by the caller.
+    fn recv_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> {
+        assert!(buf.buf.remaining_mut() > 0);
+        let r = unsafe {
+            let chunk = buf.buf.chunk_mut();
+            let slice = std::slice::from_raw_parts_mut(chunk.as_mut_ptr(), chunk.len());
+            self.io.read(slice)
+        };
+        match r {
+            Ok(n) => unsafe {
+                buf.buf.advance_mut(n);
+                Ok(n)
+            },
+            e => e,
+        }
+    }
+}
+
+impl SendMsg for Pipe {
+    // Send data on the associated connection.  `send_msg` adjusts the length of the
+    // `ConnectionBuffer` members based on the size of the successful send operation.
+    fn send_msg(&mut self, buf: &mut ConnectionBuffer) -> Result<usize> {
+        assert!(!buf.buf.is_empty());
+        let r = self.io.write(&buf.buf[..buf.buf.len()]);
+        if let Ok(n) = r {
+            buf.buf.advance(n);
+            while let Some(mut handle) = buf.pop_handle() {
+                handle.mark_sent()
+            }
+        }
+        r
+    }
+}