Bug 1646576 - Update audioipc to c04a660a. r=chunmin

Differential Revision: https://phabricator.services.mozilla.com/D80123

media/audioipc/README_MOZILLA

@@ -5,4 +5,4 @@ Makefile.in build files for the Mozilla build system.
 
 The audioipc-2 git repository is: https://github.com/djg/audioipc-2.git
 
-The git commit ID used was 1b2747695088ee5f8216bda9a7a4c4bd7655608e (2020-05-14 09:40:35 +1200)
+The git commit ID used was c04a660a1fbb728de0833c87f1c92e97021359d9 (2020-06-18 11:24:15 +1200)

media/audioipc/audioipc/Cargo.toml

@@ -12,7 +12,7 @@ edition = "2018"
 bincode = "1"
 bytes = "0.4"
 cubeb = "0.7"
-futures = "0.1.18"
+futures = "0.1.29"
 log = "0.4"
 memmap = "0.7"
 serde = "1"

media/audioipc/audioipc/src/fd_passing.rs

@@ -8,7 +8,7 @@ use crate::cmsg;
 use crate::codec::Codec;
 use crate::messages::AssocRawPlatformHandle;
 use bytes::{Bytes, BytesMut, IntoBuf};
-use futures::{AsyncSink, Poll, Sink, StartSend, Stream};
+use futures::{AsyncSink, Poll, Sink, StartSend, Stream, task};
 use std::collections::VecDeque;
 use std::os::unix::io::RawFd;
 use std::{fmt, io, mem};
@@ -268,7 +268,9 @@ where
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll_complete());
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
         self.io.shutdown()
     }
 }

media/audioipc/audioipc/src/frame.rs

@@ -5,7 +5,7 @@
 
 use crate::codec::Codec;
 use bytes::{Buf, Bytes, BytesMut, IntoBuf};
-use futures::{AsyncSink, Poll, Sink, StartSend, Stream};
+use futures::{AsyncSink, Poll, Sink, StartSend, Stream, task};
 use std::io;
 use tokio_io::{AsyncRead, AsyncWrite};
 
@@ -145,7 +145,9 @@ where
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll_complete());
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
         self.io.shutdown()
     }
 }

media/audioipc/audioipc/src/handle_passing.rs

@@ -6,7 +6,7 @@
 use crate::codec::Codec;
 use crate::messages::AssocRawPlatformHandle;
 use bytes::{Bytes, BytesMut, IntoBuf};
-use futures::{AsyncSink, Poll, Sink, StartSend, Stream};
+use futures::{AsyncSink, Poll, Sink, StartSend, Stream, task};
 use std::collections::VecDeque;
 use std::{fmt, io};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -179,11 +179,13 @@ where
         if let Some((handles, target_pid)) = item.platform_handles() {
             got_handles = true;
             let remote_handles = unsafe {
-                [
-                    duplicate_platformhandle(handles[0], target_pid)?,
-                    duplicate_platformhandle(handles[1], target_pid)?,
-                    duplicate_platformhandle(handles[2], target_pid)?,
-                ]
+                // Attempt to duplicate all 3 handles before checking
+                // result, since we rely on duplicate_platformhandle closing
+                // our source handles.
+                let r1 = duplicate_platformhandle(handles[0], target_pid);
+                let r2 = duplicate_platformhandle(handles[1], target_pid);
+                let r3 = duplicate_platformhandle(handles[2], target_pid);
+                [r1?, r2?, r3?]
             };
             trace!(
                 "item handles: {:?} remote_handles: {:?}",
@@ -216,7 +218,9 @@ where
     }
 
     fn close(&mut self) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll_complete());
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
         self.io.shutdown()
     }
 }
@@ -263,6 +267,7 @@ unsafe fn duplicate_platformhandle(
         FALSE,
         winnt::DUPLICATE_CLOSE_SOURCE | winnt::DUPLICATE_SAME_ACCESS,
     );
+    handleapi::CloseHandle(target);
     if ok == FALSE {
         return Err(std::io::Error::new(
             std::io::ErrorKind::Other,

media/audioipc/server/Cargo.toml

@@ -12,7 +12,7 @@ edition = "2018"
 audio_thread_priority = "0.22"
 audioipc = { path = "../audioipc" }
 cubeb-core = "0.7.0"
-futures = "0.1.18"
+futures = "0.1.29"
 once_cell = "1.2.0"
 log = "0.4"
 slab = "0.4"

media/audioipc/server/src/lib.rs

@@ -83,11 +83,18 @@ fn run() -> Result<ServerWrapper> {
         Err(e)
     })?;
 
-    let core_thread =
-        core::spawn_thread("AudioIPC Server RPC", move || Ok(()), || {}).or_else(|e| {
-            debug!("Failed to cubeb audio core event loop thread: {:?}", e);
-            Err(e)
-        })?;
+    let core_thread = core::spawn_thread(
+        "AudioIPC Server RPC",
+        move || {
+            audioipc::server_platform_init();
+            Ok(())
+        },
+        || {},
+    )
+    .or_else(|e| {
+        debug!("Failed to cubeb audio core event loop thread: {:?}", e);
+        Err(e)
+    })?;
 
     Ok(ServerWrapper {
         core_thread,

media/audioipc/server/src/server.rs

@@ -192,6 +192,21 @@ struct CubebContextState {
 
 thread_local!(static CONTEXT_KEY: RefCell<Option<CubebContextState>> = RefCell::new(None));
 
+fn cubeb_init_from_context_params() -> cubeb::Result<cubeb::Context> {
+    let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
+    let context_name = Some(params.context_name.as_c_str());
+    let backend_name = if let Some(ref name) = params.backend_name {
+        Some(name.as_c_str())
+    } else {
+        None
+    };
+    let r = cubeb::Context::init(context_name, backend_name);
+    r.map_err(|e| {
+        info!("cubeb::Context::init failed r={:?}", e);
+        e
+    })
+}
+
 fn with_local_context<T, F>(f: F) -> T
 where
     F: FnOnce(&cubeb::Result<cubeb::Context>, &mut CubebDeviceCollectionManager) -> T,
@@ -199,20 +214,17 @@ where
     CONTEXT_KEY.with(|k| {
         let mut state = k.borrow_mut();
         if state.is_none() {
-            let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
-            let context_name = Some(params.context_name.as_c_str());
-            let backend_name = if let Some(ref name) = params.backend_name {
-                Some(name.as_c_str())
-            } else {
-                None
-            };
-            audioipc::server_platform_init();
-            let context = cubeb::Context::init(context_name, backend_name);
-            let manager = CubebDeviceCollectionManager::new();
-            *state = Some(CubebContextState { context, manager });
+            *state = Some(CubebContextState {
+                context: cubeb_init_from_context_params(),
+                manager: CubebDeviceCollectionManager::new(),
+            });
         }
-        let state = state.as_mut().unwrap();
-        f(&state.context, &mut state.manager)
+        let CubebContextState { context, manager } = state.as_mut().unwrap();
+        // Always reattempt to initialize cubeb, OS config may have changed.
+        if context.is_err() {
+            *context = cubeb_init_from_context_params();
+        }
+        f(context, manager)
     })
 }
 
@@ -379,6 +391,9 @@ impl rpc::Server for CubebServer {
     >;
 
     fn process(&mut self, req: Self::Request) -> Self::Future {
+        if let ServerMessage::ClientConnect(pid) = req {
+            self.remote_pid = Some(pid);
+        }
         let resp = with_local_context(|context, manager| match *context {
             Err(_) => error(cubeb::Error::error()),
             Ok(ref context) => self.process_msg(context, manager, &req),
@@ -424,8 +439,9 @@ impl CubebServer {
         msg: &ServerMessage,
     ) -> ClientMessage {
         let resp: ClientMessage = match *msg {
-            ServerMessage::ClientConnect(pid) => {
-                self.remote_pid = Some(pid);
+            ServerMessage::ClientConnect(_) => {
+                // remote_pid is set before cubeb initialization, just verify here.
+                assert!(self.remote_pid.is_some());
                 ClientMessage::ClientConnected
             }