From 79e3fcc710de68b40fd05be5421048bab658ddf4 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Wed, 23 Jul 2025 14:25:35 +0200 Subject: [PATCH] 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 | 26 +++++++++++++++++++- quinn/src/connection.rs | 33 +++++++++++++++++++++---- quinn/src/path.rs | 41 ++++++++++++++++++++++--------- 3 files changed, 83 insertions(+), 17 deletions(-) diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs index 83f4a12cd..3a953a9ed 100644 --- a/quinn-proto/src/connection/mod.rs +++ b/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`]. diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index 6608b0e70..b8a732a84 100644 --- a/quinn/src/connection.rs +++ b/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, /// Tracks paths being opened - open_path: FxHashMap>>, + open_path: FxHashMap>>>, /// Tracks paths being closed pub(crate) close_path: FxHashMap>, path_events: tokio::sync::broadcast::Sender, @@ -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 } diff --git a/quinn/src/path.rs b/quinn/src/path.rs index 168282c9a..6a8ac4227 100644 --- a/quinn/src/path.rs +++ b/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>, + opened: watch::Receiver>>, 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>, + opened: watch::Receiver>>, 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)), } }