Allow opening path only if it does not yet exists (#117)

* Allow opening path only if it does not yet exists

* Hook this up to futures

All this is terrible.

* clippy

quinn-proto/src/connection/mod.rs

@@ -517,7 +517,31 @@ impl Connection {
         }
     }
 
-    /// Open a new path
+    /// Opens a new path only if no path to the remote address exists so far
+    ///
+    /// See [`open_path`]. Returns `(path_id, true)` if the path already existed. `(path_id,
+    /// false)` if was opened.
+    ///
+    /// [`open_path`]: Connection::open_path
+    pub fn open_path_ensure(
+        &mut self,
+        remote: SocketAddr,
+        initial_status: PathStatus,
+        now: Instant,
+    ) -> Result<(PathId, bool), PathError> {
+        match self
+            .paths
+            .iter()
+            .find(|(_id, path)| path.data.remote == remote)
+        {
+            Some((path_id, _state)) => Ok((*path_id, true)),
+            None => self
+                .open_path(remote, initial_status, now)
+                .map(|id| (id, false)),
+        }
+    }
+
+    /// Opens a new path
     ///
     /// Further errors might occur and they will be emitted in [`PathEvent::LocallyClosed`] events.
     /// When the path is opened it will be reported as an [`PathEvent::Opened`].

quinn/src/connection.rs

@@ -359,10 +359,33 @@ impl Connection {
         }
     }
 
-    /// Open a (Multi)Path
+    /// Opens a new path if no path exists yet for the remote address
+    pub fn open_path_ensure(&self, addr: SocketAddr, initial_status: PathStatus) -> OpenPath {
+        let mut state = self.0.state.lock("open_path");
+        let now = state.runtime.now();
+        let open_res = state.inner.open_path_ensure(addr, initial_status, now);
+        state.wake();
+        match open_res {
+            Ok((path_id, existed)) if existed => {
+                match state.open_path.get(&path_id).map(|tx| tx.subscribe()) {
+                    Some(recv) => OpenPath::new(path_id, recv, self.0.clone()),
+                    None => OpenPath::ready(path_id, self.0.clone()),
+                }
+            }
+            Ok((path_id, _)) => {
+                let (tx, rx) = watch::channel(None);
+                state.open_path.insert(path_id, tx);
+                drop(state);
+                OpenPath::new(path_id, rx, self.0.clone())
+            }
+            Err(err) => OpenPath::rejected(err),
+        }
+    }
+
+    /// Opens a (Multi)Path
     pub fn open_path(&self, addr: SocketAddr, initial_status: PathStatus) -> OpenPath {
         let mut state = self.0.state.lock("open_path");
-        let (on_open_path_send, on_open_path_recv) = oneshot::channel();
+        let (on_open_path_send, on_open_path_recv) = watch::channel(None);
         let now = state.runtime.now();
         let open_res = state.inner.open_path(addr, initial_status, now);
         state.wake();
@@ -1065,7 +1088,7 @@ pub(crate) struct State {
     /// Always set to Some before the connection becomes drained
     pub(crate) error: Option<ConnectionError>,
     /// Tracks paths being opened
-    open_path: FxHashMap<PathId, oneshot::Sender<Result<(), PathError>>>,
+    open_path: FxHashMap<PathId, watch::Sender<Option<Result<(), PathError>>>>,
     /// Tracks paths being closed
     pub(crate) close_path: FxHashMap<PathId, oneshot::Sender<VarInt>>,
     path_events: tokio::sync::broadcast::Sender<PathEvent>,
@@ -1233,7 +1256,7 @@ impl State {
                 Path(ref evt @ PathEvent::Opened { id }) => {
                     self.path_events.send(evt.clone()).ok();
                     if let Some(sender) = self.open_path.remove(&id) {
-                        let _ = sender.send(Ok(()));
+                        let _ = sender.send(Some(Ok(())));
                     }
                 }
                 Path(ref evt @ PathEvent::Closed { id, error_code }) => {
@@ -1245,7 +1268,7 @@ impl State {
                 Path(ref evt @ PathEvent::LocallyClosed { id, error }) => {
                     self.path_events.send(evt.clone()).ok();
                     if let Some(sender) = self.open_path.remove(&id) {
-                        let _ = sender.send(Err(error));
+                        let _ = sender.send(Some(Err(error)));
                     }
                     // this will happen also for already opened paths
                 }

quinn/src/path.rs

@@ -4,7 +4,7 @@ use std::task::{Context, Poll, ready};
 use std::time::Duration;
 
 use proto::{ClosePathError, ClosedPath, ConnectionError, PathError, PathId, PathStatus, VarInt};
-use tokio::sync::oneshot;
+use tokio::sync::{oneshot, watch};
 
 use crate::connection::ConnectionRef;
 
@@ -12,25 +12,30 @@ use crate::connection::ConnectionRef;
 pub struct OpenPath(OpenPathInner);
 
 enum OpenPathInner {
-    /// Opening a path in underway.
+    /// Opening a path in underway
     ///
     /// This migth fail later on.
     Ongoing {
-        opened: oneshot::Receiver<Result<(), PathError>>,
+        opened: watch::Receiver<Option<Result<(), PathError>>>,
         path_id: PathId,
         conn: ConnectionRef,
     },
-    /// Opening a path failed immediately.
+    /// Opening a path failed immediately
     Rejected {
-        /// The error that occurred.
+        /// The error that occurred
         err: PathError,
     },
+    /// The path is already open
+    Ready {
+        path_id: PathId,
+        conn: ConnectionRef,
+    },
 }
 
 impl OpenPath {
     pub(crate) fn new(
         path_id: PathId,
-        opened: oneshot::Receiver<Result<(), PathError>>,
+        opened: watch::Receiver<Option<Result<(), PathError>>>,
         conn: ConnectionRef,
     ) -> Self {
         Self(OpenPathInner::Ongoing {
@@ -40,6 +45,10 @@ impl OpenPath {
         })
     }
 
+    pub(crate) fn ready(path_id: PathId, conn: ConnectionRef) -> Self {
+        Self(OpenPathInner::Ready { path_id, conn })
+    }
+
     pub(crate) fn rejected(err: PathError) -> Self {
         Self(OpenPathInner::Rejected { err })
     }
@@ -53,12 +62,22 @@ impl Future for OpenPath {
                 ref mut opened,
                 path_id,
                 ref mut conn,
-            } => Pin::new(opened).poll(ctx).map(|_| {
-                Ok(Path {
-                    id: path_id,
-                    conn: conn.clone(),
+            } => {
+                let mut fut = std::pin::pin!(opened.wait_for(|v| v.is_some()));
+                fut.as_mut().poll(ctx).map(|_| {
+                    Ok(Path {
+                        id: path_id,
+                        conn: conn.clone(),
+                    })
                 })
-            }),
+            }
+            OpenPathInner::Ready {
+                path_id,
+                ref mut conn,
+            } => Poll::Ready(Ok(Path {
+                id: path_id,
+                conn: conn.clone(),
+            })),
             OpenPathInner::Rejected { err } => Poll::Ready(Err(err)),
         }
     }