Refactor UdpMuxNewAddr to be lock-less

transports/webrtc/src/lib.rs

@@ -86,6 +86,7 @@ pub mod transport;
 
 mod fingerprint;
 mod in_addr;
+mod req_res_chan;
 mod sdp;
 mod udp_mux;
 mod webrtc_connection;

transports/webrtc/src/req_res_chan.rs

@@ -0,0 +1,49 @@
+use futures::channel::mpsc;
+use futures::channel::oneshot;
+use futures::SinkExt;
+use futures_lite::StreamExt;
+use std::io;
+use std::task::{Context, Poll};
+
+pub fn new<Req, Res>(capacity: usize) -> (Sender<Req, Res>, Receiver<Req, Res>) {
+    let (sender, receiver) = mpsc::channel(capacity);
+
+    (
+        Sender {
+            inner: futures::lock::Mutex::new(sender),
+        },
+        Receiver { inner: receiver },
+    )
+}
+
+pub struct Sender<Req, Res> {
+    inner: futures::lock::Mutex<mpsc::Sender<(Req, oneshot::Sender<Res>)>>,
+}
+
+impl<Req, Res> Sender<Req, Res> {
+    pub async fn send(&self, req: Req) -> io::Result<Res> {
+        let (sender, receiver) = oneshot::channel();
+
+        self.inner
+            .lock()
+            .await
+            .send((req, sender))
+            .await
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+        let res = receiver
+            .await
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+
+        Ok(res)
+    }
+}
+
+pub struct Receiver<Req, Res> {
+    inner: mpsc::Receiver<(Req, oneshot::Sender<Res>)>,
+}
+
+impl<Req, Res> Receiver<Req, Res> {
+    pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<(Req, oneshot::Sender<Res>)>> {
+        self.inner.poll_next(cx)
+    }
+}

transports/webrtc/src/transport.rs

@@ -170,7 +170,7 @@ impl Transport for WebRTCTransport {
             .iter()
             .next()
             .ok_or(TransportError::Other(Error::NoListeners))?;
-        let udp_mux = first_listener.udp_mux.clone();
+        let udp_mux = first_listener.udp_mux.udp_mux_handle();
 
         // [`Transport::dial`] should do no work unless the returned [`Future`] is polled. Thus
         // do the `set_remote_description` call within the [`Future`].
@@ -259,7 +259,7 @@ pub struct WebRTCListenStream {
     config: WebRTCConfiguration,
 
     /// The UDP muxer that manages all ICE connections.
-    udp_mux: Arc<UDPMuxNewAddr>,
+    udp_mux: UDPMuxNewAddr,
 
     /// `Keypair` identifying this peer
     id_keys: identity::Keypair,
@@ -277,7 +277,7 @@ impl WebRTCListenStream {
         listener_id: ListenerId,
         listen_addr: SocketAddr,
         config: WebRTCConfiguration,
-        udp_mux: Arc<UDPMuxNewAddr>,
+        udp_mux: UDPMuxNewAddr,
         id_keys: identity::Keypair,
     ) -> Self {
         let in_addr = InAddr::new(listen_addr.ip());
@@ -382,13 +382,13 @@ impl Stream for WebRTCListenStream {
             }
 
             // Poll UDP muxer for new addresses or incoming data for streams.
-            match ready!(self.udp_mux.as_ref().poll(cx)) {
+            match ready!(self.udp_mux.poll(cx)) {
                 UDPMuxEvent::NewAddr(new_addr) => {
                     let local_addr = socketaddr_to_multiaddr(&self.listen_addr);
                     let send_back_addr = socketaddr_to_multiaddr(&new_addr.addr);
                     let event = TransportEvent::Incoming {
                         upgrade: Box::pin(upgrade(
-                            self.udp_mux.clone(),
+                            self.udp_mux.udp_mux_handle(),
                             self.config.clone(),
                             new_addr.addr,
                             new_addr.ufrag,