From 06946a2bf4358a91129392d9b385297e3fe81fe1 Mon Sep 17 00:00:00 2001 From: cloudhead Date: Thu, 16 Nov 2023 20:52:19 +0100 Subject: [PATCH] node: New fetch queueing system The previous system allowed the same RID to be fetched from multiple nodes concurrently, which caused issues, especially when cloning. Worker code assumes that there is only one fetch per repository going on at any given time. This change ensures that. We keep a global fetch queue, instead of having a queue per session. We also enqueue fetches if a session is at capacity. Finally, we allow multiple fetch requests to receive results for the same fetch. --- radicle-node/src/service.rs | 212 +++++++++++++++++++--------- radicle-node/src/service/io.rs | 2 + radicle-node/src/service/session.rs | 78 +++++----- radicle-node/src/tests.rs | 95 ++++++++++--- 4 files changed, 262 insertions(+), 125 deletions(-) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 89feb277..86c57100 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -10,7 +10,7 @@ pub mod message; pub mod session; use std::collections::hash_map::Entry; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::ops::{Deref, DerefMut}; use std::sync::Arc; use std::{fmt, time}; @@ -195,6 +195,29 @@ pub enum CommandError { Tracking(#[from] tracking::Error), } +/// Error returned by [`Service::try_fetch`]. +#[derive(thiserror::Error, Debug)] +enum TryFetchError { + #[error("session {0} does not exist; cannot initiate fetch")] + SessionNotFound(NodeId), + #[error("session {0} is not connected; cannot initiate fetch")] + SessionNotConnected(NodeId), + #[error("session {0} fetch capacity reached; cannot initiate fetch")] + SessionCapacityReached(NodeId), + #[error(transparent)] + Namespaces(#[from] NamespacesError), +} + +/// Fetch state for an ongoing fetch. +#[derive(Debug)] +struct FetchState { + /// Node we're fetching from. + from: NodeId, + /// Channels waiting for fetch results. + subscribers: Vec>, +} + +/// The node service. #[derive(Debug)] pub struct Service { /// Service configuration. @@ -221,8 +244,10 @@ pub struct Service { node: NodeAnnouncement, /// Source of entropy. rng: Rng, - /// Fetch requests initiated by user, which are waiting for results. - fetch_reqs: HashMap<(Id, NodeId), chan::Sender>, + /// Ongoing fetches. + fetching: HashMap, + /// Fetch queue. + queue: VecDeque<(Id, NodeId)>, /// Request/connection rate limitter. limiter: RateLimiter, /// Current tracked repository bloom filter. @@ -292,7 +317,8 @@ where outbox: Outbox::default(), limiter: RateLimiter::default(), sessions, - fetch_reqs: HashMap::new(), + fetching: HashMap::new(), + queue: VecDeque::new(), filter: Filter::empty(), last_idle: LocalTime::default(), last_sync: LocalTime::default(), @@ -550,9 +576,7 @@ where } }, Command::Fetch(rid, seed, timeout, resp) => { - // TODO: Establish connections to unconnected seeds, and retry. - self.fetch_reqs.insert((rid, seed), resp); - self.fetch(rid, &seed, timeout); + self.fetch(rid, &seed, timeout, Some(resp)); } Command::TrackRepo(rid, scope, resp) => { // Update our tracking policy. @@ -623,57 +647,94 @@ where refs: Vec, timeout: time::Duration, ) { - self._fetch(rid, from, refs, timeout) + self._fetch(rid, from, refs, timeout, None) } /// Initiate an outgoing fetch for some repository. - fn fetch(&mut self, rid: Id, from: &NodeId, timeout: time::Duration) { - self._fetch(rid, from, vec![], timeout) + fn fetch( + &mut self, + rid: Id, + from: &NodeId, + timeout: time::Duration, + channel: Option>, + ) { + self._fetch(rid, from, vec![], timeout, channel) } - fn _fetch(&mut self, rid: Id, from: &NodeId, refs_at: Vec, timeout: time::Duration) { - let Some(session) = self.sessions.get_mut(from) else { - error!(target: "service", "Session {from} does not exist; cannot initiate fetch"); - return; + fn _fetch( + &mut self, + rid: Id, + from: &NodeId, + refs_at: Vec, + timeout: time::Duration, + channel: Option>, + ) { + match self.try_fetch(rid, from, refs_at, timeout) { + Ok(FetchState { subscribers, .. }) => { + if let Some(c) = channel { + if !subscribers.iter().any(|s| s.same_channel(&c)) { + subscribers.push(c); + } + } + } + Err(e) => { + if let Some(c) = channel { + c.send(FetchResult::Failed { + reason: e.to_string(), + }) + .ok(); + } + } + } + } + + fn try_fetch( + &mut self, + rid: Id, + from: &NodeId, + refs_at: Vec, + timeout: time::Duration, + ) -> Result<&mut FetchState, TryFetchError> { + let from = *from; + let Some(session) = self.sessions.get_mut(&from) else { + return Err(TryFetchError::SessionNotFound(from)); }; + let fetching = self.fetching.entry(rid); + + if let Entry::Occupied(fetching) = fetching { + if fetching.get().from == from { + debug!(target: "service", "Ignoring redundant fetch of {rid} from {from}"); + debug_assert!(session.is_fetching(&rid)); + } else { + debug!(target: "service", "Queueing fetch for {rid} with {from}.."); + self.queue.push_back((rid, from)); + } + return Ok(fetching.into_mut()); + } if !session.is_connected() { // This can happen if a session disconnects in the time between asking for seeds to // fetch from, and initiating the fetch from one of those seeds. - error!(target: "service", "Session {from} is not connected; cannot initiate fetch"); - return; + return Err(TryFetchError::SessionNotConnected(session.id)); } - let seed = session.id; + if session.is_at_capacity() { + debug!(target: "service", "Fetch capacity reached for {}, queueing {rid}..", session.id); + self.queue.push_back((rid, session.id)); - match session.fetch(rid) { - session::FetchResult::Queued => { - debug!(target: "service", "Fetch queued for {rid} with {seed}.."); - } - session::FetchResult::Ready => { - debug!(target: "service", "Fetch initiated for {rid} with {seed}.."); - match self.tracking.namespaces_for(&self.storage, &rid) { - Ok(namespaces) => { - self.outbox - .fetch(session, rid, namespaces, refs_at, timeout); - } - Err(err) => { - error!(target: "service", "Error getting namespaces for {rid}: {err}"); - - if let Some(resp) = self.fetch_reqs.remove(&(rid, seed)) { - resp.send(FetchResult::Failed { - reason: err.to_string(), - }) - .ok(); - } - } - }; - } - session::FetchResult::AlreadyFetching => { - debug!(target: "service", "Ignoring redundant attempt to fetch {rid} from {from}"); - } - session::FetchResult::NotConnected => { - error!(target: "service", "Unable to fetch {rid} from peer {seed}: peer is not connected"); - } + return Err(TryFetchError::SessionCapacityReached(session.id)); } + + let fetching = fetching.or_insert(FetchState { + from, + subscribers: vec![], + }); + let namespaces = self.tracking.namespaces_for(&self.storage, &rid)?; + + self.outbox + .fetch(session, rid, namespaces, refs_at, timeout); + + debug!(target: "service", "Fetch initiated for {rid} with {}..", session.id); + + Ok(fetching) } pub fn fetched( @@ -716,26 +777,39 @@ where } }; - if let Some(results) = self.fetch_reqs.remove(&(rid, remote)) { + let Some(fetching) = self.fetching.remove(&rid) else { + warn!(target: "service", "Received unexpected fetch result for {rid}, from {remote}"); + return; + }; + debug_assert_eq!(fetching.from, remote); + + if let Some(s) = self.sessions.get_mut(&remote) { + // Mark this RID as fetched for this session. + s.fetched(rid); + } + + for sub in &fetching.subscribers { debug!(target: "service", "Found existing fetch request, sending result.."); - if results.send(result).is_err() { + if sub.send(result.clone()).is_err() { error!(target: "service", "Error sending fetch result for {rid}.."); } else { debug!(target: "service", "Sent fetch result for {rid}.."); } - } else { - debug!(target: "service", "No fetch requests found for {rid}.."); + } + + if fetching.subscribers.is_empty() { + trace!(target: "service", "No fetch requests found for {rid}.."); // We only announce refs here when the fetch wasn't user-requested. This is // because the user might want to announce his fork, once he has created one, // or may choose to not announce anything. - match result { + match &result { FetchResult::Success { updated, namespaces, } if !updated.is_empty() => { - if let Err(e) = self.announce_refs(rid, namespaces) { + if let Err(e) = self.announce_refs(rid, namespaces.iter().cloned()) { error!(target: "service", "Failed to announce new refs: {e}"); } } @@ -751,12 +825,18 @@ where // network, if necessary. self.sync_and_announce(); - if let Some(s) = self.sessions.get_mut(&remote) { - if let Some(dequeued) = s.fetched(rid) { - debug!(target: "service", "Dequeued fetch {dequeued} from session {remote}.."); + // We can now try to dequeue another fetch. + self.dequeue_fetch(); + } - self.fetch(dequeued, &remote, FETCH_TIMEOUT); - } + /// Fetches are queued for two reasons: + /// 1. The RID was already being fetched. + /// 2. The session was already at fetch capacity. + pub fn dequeue_fetch(&mut self) { + if let Some((rid, nid)) = self.queue.pop_front() { + debug!(target: "service", "Dequeued fetch for {rid} from session {nid}.."); + + self.fetch(rid, &nid, FETCH_TIMEOUT, None); } } @@ -846,16 +926,19 @@ where }; let link = session.link; - // If the peer disconnected while we were fetching, return a failure to any - // potential fetcher. - for rid in session.fetching() { - if let Some(resp) = self.fetch_reqs.remove(&(rid, remote)) { + self.fetching.retain(|_, fetching| { + if fetching.from != remote { + return true; + } + // Remove and fail any pending fetches from this remote node. + for resp in &fetching.subscribers { resp.send(FetchResult::Failed { reason: format!("disconnected: {reason}"), }) .ok(); } - } + false + }); // Attempt to re-connect to persistent peers. if self.config.peer(&remote).is_some() { @@ -885,6 +968,7 @@ where self.maintain_connections(); } } + self.dequeue_fetch(); } pub fn received_message(&mut self, remote: NodeId, message: Message) { @@ -1012,7 +1096,7 @@ where } Ok(false) => { debug!(target: "service", "Missing tracked inventory {id}; initiating fetch.."); - self.fetch(*id, announcer, FETCH_TIMEOUT); + self.fetch(*id, announcer, FETCH_TIMEOUT, None); } Err(e) => { error!(target: "service", "Error checking local inventory: {e}"); @@ -1648,7 +1732,7 @@ where Ok(seeds) => { if let Some(connected) = NonEmpty::from_vec(seeds.connected().collect()) { for seed in connected { - self.fetch(rid, &seed.nid, FETCH_TIMEOUT); + self.fetch(rid, &seed.nid, FETCH_TIMEOUT, None); } } else { // TODO: We should make sure that this fetch is retried later, either diff --git a/radicle-node/src/service/io.rs b/radicle-node/src/service/io.rs index c46f1ab9..df7a1576 100644 --- a/radicle-node/src/service/io.rs +++ b/radicle-node/src/service/io.rs @@ -91,6 +91,8 @@ impl Outbox { refs_at: Vec, timeout: time::Duration, ) { + remote.fetching(rid); + let refs_at = (!refs_at.is_empty()).then_some(refs_at); self.io.push_back(Io::Fetch { rid, diff --git a/radicle-node/src/service/session.rs b/radicle-node/src/service/session.rs index 92965541..1e960853 100644 --- a/radicle-node/src/service/session.rs +++ b/radicle-node/src/service/session.rs @@ -1,4 +1,4 @@ -use std::collections::{HashSet, VecDeque}; +use std::collections::HashSet; use std::fmt; use crate::node::config::Limits; @@ -9,19 +9,6 @@ use crate::Link; pub use crate::node::{PingState, State}; -/// Return value of [`Session::fetch`]. -#[derive(Debug)] -pub enum FetchResult { - /// Maximum concurrent fetches reached. - Queued, - /// We are already fetching the given repo from this peer. - AlreadyFetching, - /// Ok, ready to fetch. - Ready, - /// This peer is not ready to fetch. - NotConnected, -} - #[derive(thiserror::Error, Debug)] pub enum Error { /// The remote peer sent an invalid announcement timestamp, @@ -70,8 +57,6 @@ pub struct Session { pub subscribe: Option, /// Last time a message was received from the peer. pub last_active: LocalTime, - /// Fetch queue. - pub queue: VecDeque, /// Connection attempts. For persistent peers, Tracks /// how many times we've attempted to connect. We reset this to zero @@ -112,7 +97,6 @@ impl Session { subscribe: None, persistent, last_active: LocalTime::default(), - queue: VecDeque::default(), attempts: 1, rng, limits, @@ -139,7 +123,6 @@ impl Session { subscribe: None, persistent, last_active: LocalTime::default(), - queue: VecDeque::default(), attempts: 0, rng, limits, @@ -162,38 +145,51 @@ impl Session { matches!(self.state, State::Initial) } + pub fn is_at_capacity(&self) -> bool { + if let State::Connected { fetching, .. } = &self.state { + if fetching.len() >= self.limits.fetch_concurrency { + return true; + } + } + false + } + + pub fn is_fetching(&self, rid: &Id) -> bool { + if let State::Connected { fetching, .. } = &self.state { + return fetching.contains(rid); + } + false + } + pub fn attempts(&self) -> usize { self.attempts } - pub fn fetch(&mut self, rid: Id) -> FetchResult { + /// Mark this session as fetching the given RID. + /// + /// # Panics + /// + /// If it is already fetching that RID, or the session is disconnected. + pub fn fetching(&mut self, rid: Id) { if let State::Connected { fetching, .. } = &mut self.state { - if fetching.contains(&rid) || self.queue.contains(&rid) { - return FetchResult::AlreadyFetching; - } - if fetching.len() >= self.limits.fetch_concurrency { - self.queue.push_back(rid); - return FetchResult::Queued; - } - fetching.insert(rid); - - FetchResult::Ready + assert!( + fetching.insert(rid), + "Session must not already be fetching {rid}" + ); } else { - FetchResult::NotConnected + panic!( + "Attempting to fetch {rid} from disconnected session {}", + self.id + ); } } - pub fn fetched(&mut self, rid: Id) -> Option { + pub fn fetched(&mut self, rid: Id) { if let State::Connected { fetching, .. } = &mut self.state { if !fetching.remove(&rid) { - log::error!(target: "service", "Fetched unknown repository {rid}"); - } - // Dequeue the next fetch, if any. - if let Some(rid) = self.queue.pop_front() { - return Some(rid); + log::warn!(target: "service", "Fetched unknown repository {rid}"); } } - None } pub fn to_attempted(&mut self) { @@ -234,14 +230,6 @@ impl Session { self.state = State::Initial; } - pub fn fetching(&self) -> HashSet { - if let State::Connected { fetching, .. } = &self.state { - fetching.clone() - } else { - HashSet::default() - } - } - pub fn ping(&mut self, reactor: &mut Outbox) -> Result<(), Error> { if let State::Connected { ping, .. } = &mut self.state { let msg = message::Ping::new(&mut self.rng); diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index ac2ca281..b6b406ee 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -41,6 +41,7 @@ use crate::test::storage as mock_storage; use crate::test::storage::MockStorage; use crate::wire::Decode; use crate::wire::Encode; +use crate::worker; use crate::worker::fetch; use crate::LocalTime; use crate::{git, identity, rad, runtime, service, test}; @@ -1175,16 +1176,14 @@ fn test_track_repo_subscribe() { } #[test] -fn test_fetch_missing_inventory() { +fn test_fetch_missing_inventory_on_gossip() { let rid = arbitrary::gen::(1); let mut alice = Peer::new("alice", [7, 7, 7, 7]); let bob = Peer::new("bob", [8, 8, 8, 8]); - let eve = Peer::new("eve", [9, 9, 9, 9]); - let (send, recv) = chan::bounded::(1); let now = LocalTime::now(); + alice.track_repo(&rid, node::tracking::Scope::All).unwrap(); alice.connect_to(&bob); - alice.connect_to(&eve); alice.receive( bob.id(), Message::inventory( @@ -1195,33 +1194,48 @@ fn test_fetch_missing_inventory() { bob.signer(), ), ); + alice + .outbox() + .find(|m| matches!(m, Io::Fetch { rid: other, .. } if other == &rid)) + .unwrap(); +} + +#[test] +fn test_fetch_missing_inventory_on_schedule() { + let rid = arbitrary::gen::(1); + let mut alice = Peer::new("alice", [7, 7, 7, 7]); + let bob = Peer::new("bob", [8, 8, 8, 8]); + let now = LocalTime::now(); + + alice.track_repo(&rid, node::tracking::Scope::All).unwrap(); + alice.connect_to(&bob); alice.receive( - eve.id(), + bob.id(), Message::inventory( InventoryAnnouncement { inventory: vec![rid].try_into().unwrap(), timestamp: now.as_millis(), }, - eve.signer(), + bob.signer(), ), ); - alice.command(Command::TrackRepo(rid, node::tracking::Scope::All, send)); + alice.fetched( + rid, + bob.id, + Err(worker::FetchError::Io( + io::ErrorKind::ConnectionReset.into(), + )), + ); alice.outbox().for_each(drop); - - assert!(recv.recv().unwrap()); - alice.elapse(service::SYNC_INTERVAL); alice .outbox() - .find(|m| matches!(m, Io::Fetch { .. })) - .unwrap(); - alice - .outbox() - .find(|m| matches!(m, Io::Fetch { .. })) + .find(|m| matches!(m, Io::Fetch { rid: other, .. } if other == &rid)) .unwrap(); } + #[test] -fn test_queued_fetch() { +fn test_queued_fetch_max_capacity() { let storage = arbitrary::nonempty_storage(3); let mut repo_keys = storage.inventory.keys(); let rid1 = *repo_keys.next().unwrap(); @@ -1267,6 +1281,55 @@ fn test_queued_fetch() { assert_matches!(alice.fetches().next(), Some((rid, _, _)) if rid == rid3); } +#[test] +fn test_queued_fetch_same_rid() { + let storage = arbitrary::nonempty_storage(3); + let mut repo_keys = storage.inventory.keys(); + let rid1 = *repo_keys.next().unwrap(); + let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage); + let bob = Peer::new("bob", [8, 8, 8, 8]); + let eve = Peer::new("eve", [9, 9, 9, 9]); + let carol = Peer::new("carol", [10, 10, 10, 10]); + + logger::init(log::Level::Debug); + + alice.connect_to(&bob); + alice.connect_to(&eve); + alice.connect_to(&carol); + + // Send the first fetch. + let (send, _recv1) = chan::bounded::(1); + alice.command(Command::Fetch(rid1, bob.id, DEFAULT_TIMEOUT, send)); + + // Send the 2nd fetch that will be queued. + let (send2, _recv2) = chan::bounded::(1); + alice.command(Command::Fetch(rid1, eve.id, DEFAULT_TIMEOUT, send2)); + + // Send the 3rd fetch that will be queued. + let (send3, _recv3) = chan::bounded::(1); + alice.command(Command::Fetch(rid1, carol.id, DEFAULT_TIMEOUT, send3)); + + // The first fetch is initiated. + assert_matches!(alice.fetches().next(), Some((rid, nid, _)) if rid == rid1 && nid == bob.id); + // We shouldn't send out the 2nd, 3rd fetch while we're doing the 1st fetch. + assert_matches!(alice.outbox().next(), None); + + // Have enough time pass that Alice sends a "ping" to Bob. + alice.elapse(KEEP_ALIVE_DELTA); + + // Finish the 1st fetch. + alice.fetched(rid1, bob.id, Ok(fetch::FetchResult::default())); + // Now the 1st fetch is done, the 2nd fetch is dequeued. + assert_matches!(alice.fetches().next(), Some((rid, nid, _)) if rid == rid1 && nid == eve.id); + // ... but not the third. + assert_matches!(alice.fetches().next(), None); + + // Finish the 2nd fetch. + alice.fetched(rid1, eve.id, Ok(fetch::FetchResult::default())); + // Now the 2nd fetch is done, the 3rd fetch is dequeued. + assert_matches!(alice.fetches().next(), Some((rid, nid, _)) if rid == rid1 && nid == carol.id); +} + #[test] fn test_refs_synced_event() { let temp = tempfile::tempdir().unwrap();