From f4257eb9537b3a7077562d9907f3f2432fdf4ca8 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Thu, 13 Apr 2023 11:32:04 +0200 Subject: [PATCH] node: Add setting for limiting fetch concurrency Can be passed via `--limit-fetch-concurrency `. --- radicle-node/src/main.rs | 19 ++++++----- radicle-node/src/service.rs | 32 +++++++++++------- radicle-node/src/service/config.rs | 3 ++ radicle-node/src/service/session.rs | 44 ++++++++++++++++++++----- radicle-node/src/test/peer.rs | 18 +++++++++- radicle-node/src/tests.rs | 51 +++++++++++++++++++++++++++++ radicle-node/src/tests/e2e.rs | 16 +++++++-- 7 files changed, 151 insertions(+), 32 deletions(-) diff --git a/radicle-node/src/main.rs b/radicle-node/src/main.rs index 7bfb55d2..18d7f857 100644 --- a/radicle-node/src/main.rs +++ b/radicle-node/src/main.rs @@ -21,14 +21,14 @@ Usage Options - --connect Connect to the given peer address on start - --external-address
Publicly accessible address (default 0.0.0.0:8776) - --git-daemon
Address to bind git-daemon to (default 0.0.0.0:9418) - --tracking-policy (track|block) Default tracking policy - --tracking-scope (trusted|all) Default scope for tracking policies - --force Force start even if an existing control socket is found - --help Print help - --listen
Address to listen on + --connect Connect to the given peer address on start + --external-address
Publicly accessible address (default 0.0.0.0:8776) + --git-daemon
Address to bind git-daemon to (default 0.0.0.0:9418) + --tracking-policy (track|block) Default tracking policy + --tracking-scope (trusted|all) Default scope for tracking policies + --force Force start even if an existing control socket is found + --help Print help + --listen
Address to listen on "#; @@ -96,6 +96,9 @@ impl Options { Long("limit-routing-max-size") => { limits.routing_max_size = parser.value()?.parse()?; } + Long("limit-fetch-concurrency") => { + limits.fetch_concurrency = parser.value()?.parse()?; + } Long("listen") => { let addr = parser.value()?.parse()?; listen.push(addr); diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index efa75147..2d595208 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -567,6 +567,9 @@ where let seed = 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}.."); @@ -654,11 +657,6 @@ where _ => log::debug!(target: "service", "Nothing to announce, no refs were updated.."), } } - - // Go back to "idle". - if let Some(s) = self.sessions.get_mut(&remote) { - s.fetched(rid); - } // TODO: Since this fetch could be either a full clone // or simply a ref update, we need to either announce // new inventory, or new refs. Right now, we announce @@ -666,11 +664,15 @@ where // // Announce the newly fetched repository to the // network, if necessary. - // - // Nb. This needs to be run after we've switched back - // to the gossip protocol, otherwise the messages will - // be queued. self.sync_and_announce(); + + if let Some(s) = self.sessions.get_mut(&remote) { + if let Some(dequeued) = s.fetched(rid) { + log::debug!(target: "service", "Dequeued fetch {dequeued} from session {remote}.."); + + self.fetch(dequeued, &remote); + } + } } pub fn accepted(&mut self, _addr: net::SocketAddr) { @@ -713,6 +715,7 @@ where self.config.is_persistent(&remote), self.rng.clone(), self.clock, + self.config.limits.clone(), )); self.reactor.write_all(peer, msgs); } @@ -1275,8 +1278,15 @@ where } let persistent = self.config.is_persistent(&nid); - self.sessions - .insert(nid, Session::outbound(nid, persistent, self.rng.clone())); + self.sessions.insert( + nid, + Session::outbound( + nid, + persistent, + self.rng.clone(), + self.config.limits.clone(), + ), + ); self.reactor.connect(nid, addr); true diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index 33747697..e2552911 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -20,6 +20,8 @@ pub struct Limits { pub routing_max_size: usize, /// How long to keep a routing table entry before being pruned. pub routing_max_age: LocalDuration, + /// Maximum number of concurrent fetches per per connection. + pub fetch_concurrency: usize, } impl Default for Limits { @@ -27,6 +29,7 @@ impl Default for Limits { Self { routing_max_size: 1000, routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + fetch_concurrency: 1, } } } diff --git a/radicle-node/src/service/session.rs b/radicle-node/src/service/session.rs index 3eeae834..4fdc5d7c 100644 --- a/radicle-node/src/service/session.rs +++ b/radicle-node/src/service/session.rs @@ -1,6 +1,7 @@ -use std::collections::HashSet; +use std::collections::{HashSet, VecDeque}; use std::fmt; +use crate::service::config::Limits; use crate::service::message; use crate::service::message::Message; use crate::service::{Id, LocalTime, NodeId, Reactor, Rng}; @@ -64,6 +65,8 @@ impl fmt::Display for 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. @@ -118,14 +121,17 @@ 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 /// upon successful connection. attempts: usize, - /// Source of entropy. rng: Rng, + /// Protocol limits. + limits: Limits, } impl fmt::Display for Session { @@ -148,7 +154,7 @@ impl fmt::Display for Session { } impl Session { - pub fn outbound(id: NodeId, persistent: bool, rng: Rng) -> Self { + pub fn outbound(id: NodeId, persistent: bool, rng: Rng, limits: Limits) -> Self { Self { id, state: State::Initial, @@ -156,12 +162,20 @@ impl Session { subscribe: None, persistent, last_active: LocalTime::default(), + queue: VecDeque::default(), attempts: 1, rng, + limits, } } - pub fn inbound(id: NodeId, persistent: bool, rng: Rng, time: LocalTime) -> Self { + pub fn inbound( + id: NodeId, + persistent: bool, + rng: Rng, + time: LocalTime, + limits: Limits, + ) -> Self { Self { id, state: State::Connected { @@ -173,8 +187,10 @@ impl Session { subscribe: None, persistent, last_active: LocalTime::default(), + queue: VecDeque::default(), attempts: 0, rng, + limits, } } @@ -200,22 +216,32 @@ impl Session { pub fn fetch(&mut self, rid: Id) -> FetchResult { if let State::Connected { fetching, .. } = &mut self.state { - if fetching.insert(rid) { - FetchResult::Ready - } else { - FetchResult::AlreadyFetching + 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 } else { FetchResult::NotConnected } } - pub fn fetched(&mut self, rid: Id) { + pub fn fetched(&mut self, rid: Id) -> Option { 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); + } } + None } pub fn to_attempted(&mut self) { diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 17d40a29..76de2a22 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -23,7 +23,7 @@ use crate::service::tracking::{Policy, Scope}; use crate::service::*; use crate::storage::git::transport::remote; use crate::storage::Inventory; -use crate::storage::{RemoteId, WriteStorage}; +use crate::storage::{Namespaces, RemoteId, WriteStorage}; use crate::test::arbitrary; use crate::test::simulator; use crate::test::storage::MockStorage; @@ -351,4 +351,20 @@ where pub fn outbox(&mut self) -> impl Iterator + '_ { iter::from_fn(|| self.service.reactor().next()) } + + /// Get a draining iterator over the peer's I/O outbox, which only returns fetches. + pub fn fetches(&mut self) -> impl Iterator + '_ { + iter::from_fn(|| self.service.reactor().next()).filter_map(|io| { + if let Io::Fetch { + rid, + remote, + namespaces, + } = io + { + Some((rid, remote, namespaces)) + } else { + None + } + }) + } } diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index 12ca1d88..084d28c5 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -22,6 +22,7 @@ use crate::service::ServiceState as _; use crate::service::*; use crate::storage::git::transport::{local, remote}; use crate::storage::git::Storage; +use crate::storage::Namespaces; use crate::storage::ReadStorage; use crate::test::arbitrary; use crate::test::assert_matches; @@ -269,6 +270,7 @@ fn test_inventory_pruning() { limits: Limits { routing_max_size: 0, routing_max_age: LocalDuration::from_secs(0), + ..Limits::default() }, peer_projects: vec![10; 5], wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), @@ -279,6 +281,7 @@ fn test_inventory_pruning() { limits: Limits { routing_max_size: 0, routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + ..Limits::default() }, peer_projects: vec![10; 5], wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), @@ -289,6 +292,7 @@ fn test_inventory_pruning() { limits: Limits { routing_max_size: 50, routing_max_age: LocalDuration::from_mins(0), + ..Limits::default() }, peer_projects: vec![10; 5], wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), @@ -299,6 +303,7 @@ fn test_inventory_pruning() { limits: Limits { routing_max_size: 25, routing_max_age: LocalDuration::from_mins(7 * 24 * 60), + ..Limits::default() }, peer_projects: vec![10; 5], wait_time: LocalDuration::from_mins(7 * 24 * 60) + LocalDuration::from_secs(1), @@ -1079,6 +1084,52 @@ fn test_fetch_missing_inventory() { .find(|m| matches!(m, Io::Fetch { .. })) .unwrap(); } +#[test] +fn test_queued_fetch() { + let storage = arbitrary::nonempty_storage(3); + let mut repo_keys = storage.inventory.keys(); + let rid1 = *repo_keys.next().unwrap(); + let rid2 = *repo_keys.next().unwrap(); + let rid3 = *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]); + + logger::init(log::Level::Debug); + + alice.connect_to(&bob); + + // Send the first fetch. + let (send, _recv1) = chan::bounded::(1); + alice.command(Command::Fetch(rid1, bob.id, send)); + + // Send the 2nd fetch that will be queued. + let (send2, _recv2) = chan::bounded::(1); + alice.command(Command::Fetch(rid2, bob.id, send2)); + + // Send the 3rd fetch that will be queued. + let (send3, _recv3) = chan::bounded::(1); + alice.command(Command::Fetch(rid3, bob.id, send3)); + + // The first fetch is initiated. + assert_matches!(alice.fetches().next(), Some((rid, _, _)) if rid == rid1); + // 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, Namespaces::All, bob.id, Ok(vec![])); + // Now the 1st fetch is done, the 2nd fetch is dequeued. + assert_matches!(alice.fetches().next(), Some((rid, _, _)) if rid == rid2); + // ... but not the third. + assert_matches!(alice.fetches().next(), None); + + // Finish the 2nd fetch. + alice.fetched(rid2, Namespaces::All, bob.id, Ok(vec![])); + // Now the 2nd fetch is done, the 3rd fetch is dequeued. + assert_matches!(alice.fetches().next(), Some((rid, _, _)) if rid == rid3); +} #[test] fn test_push_and_pull() { diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs index 668acae2..3c1f09d9 100644 --- a/radicle-node/src/tests/e2e.rs +++ b/radicle-node/src/tests/e2e.rs @@ -7,6 +7,7 @@ use radicle::test::fixtures; use radicle::{assert_matches, rad}; use crate::service; +use crate::service::config::Limits; use crate::service::tracking::Scope; use crate::storage::git::transport; use crate::test::environment::{converge, Environment, Node}; @@ -468,8 +469,9 @@ fn test_concurrent_fetches() { let mut alice_repos = HashSet::new(); let mut alice = Node::init(&env.tmp()); let mut bob = Node::init(&env.tmp()); + let repos = scale.max(4); - for i in 0..scale.max(3) { + for i in 0..repos { // Create a repo for Alice. let tmp = tempfile::tempdir().unwrap(); let (repo, _) = fixtures::repository(tmp.path()); @@ -487,8 +489,16 @@ fn test_concurrent_fetches() { bob_repos.insert(rid); } - let mut alice = alice.spawn(service::Config::default()); - let mut bob = bob.spawn(service::Config::default()); + let config = service::Config { + limits: Limits { + // Have one fetch be queued. + fetch_concurrency: repos - 1, + ..Limits::default() + }, + ..service::Config::default() + }; + let mut alice = alice.spawn(config.clone()); + let mut bob = bob.spawn(config); let alice_events = alice.handle.events(); let bob_events = bob.handle.events();