diff --git a/Cargo.lock b/Cargo.lock index b94f4d5ea..2e3f8cca1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1525,6 +1525,17 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2 0.2.21", + "equivalent", + "foldhash 0.2.0", +] + [[package]] name = "hdrhistogram" version = "7.5.4" @@ -2033,6 +2044,15 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lru" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b6180140927ee907000b0aa540091f6ea512ead4447c92b8fc35bc72788a5a6" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -4307,6 +4327,7 @@ dependencies = [ "hashbrown 0.16.1", "humantime", "libc", + "lru", "metrics", "moka", "object_store", diff --git a/Cargo.toml b/Cargo.toml index c593783db..9d8eade75 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,6 +79,7 @@ indexmap = "2.13.1" itertools = "0.14" libc = "0.2" libfuzzer-sys = "0.4" +lru = "0.18.1" metrics = "0.24" metrics-exporter-prometheus = "0.18" moka = { version = "0.12", features = ["sync"] } diff --git a/core/Cargo.toml b/core/Cargo.toml index 9332ff04d..4070139ca 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -51,14 +51,15 @@ governor = { workspace = true, optional = true } hashbrown = { workspace = true } humantime = { workspace = true } libc = { workspace = true } +lru = { workspace = true } metrics = { workspace = true } moka = { workspace = true } object_store = { workspace = true, optional = true } parking_lot = { workspace = true } parking_lot_core = { workspace = true } pin-project-lite = { workspace = true } -quick_cache = { workspace = true } qfilter = { workspace = true } +quick_cache = { workspace = true } rand = { workspace = true } rayon = { workspace = true } scopeguard = { workspace = true } diff --git a/core/src/block_strider/starter/cold_boot.rs b/core/src/block_strider/starter/cold_boot.rs index 867e4f8b4..51bb3c7eb 100644 --- a/core/src/block_strider/starter/cold_boot.rs +++ b/core/src/block_strider/starter/cold_boot.rs @@ -944,6 +944,8 @@ impl StarterInner { }); let ZerostateProof::Found { proof } = guard.data() else { + // NOTE: All nodes should serve zerostate proof so we `reject` + // the response in case the node returned `NotFound`. anyhow::bail!("zerostate proof not found"); }; @@ -1138,7 +1140,7 @@ async fn download_block_proof_task( let (handle, data) = res.split(); let KeyBlockProof::Found { proof: data } = data else { tracing::debug!(%block_id, "block proof not found"); - handle.accept(); + handle.ignore(); break 'validate; }; diff --git a/core/src/blockchain_rpc/client.rs b/core/src/blockchain_rpc/client.rs index e76040e14..ecf186346 100644 --- a/core/src/blockchain_rpc/client.rs +++ b/core/src/blockchain_rpc/client.rs @@ -20,6 +20,7 @@ use tycho_util::serde_helpers; use crate::overlay_client::{ Error, Neighbour, NeighbourType, PublicOverlayClient, QueryResponse, QueryResponseHandle, + RequestHistory, }; use crate::proto::blockchain::*; use crate::proto::overlay::BroadcastPrefix; @@ -54,6 +55,9 @@ pub struct BlockchainRpcClientConfig { /// /// Default: 10. pub download_retries: usize, + + /// Capacity of per-request peer history. + pub request_history_size: BlockchainRpcRequestHistorySize, } impl Default for BlockchainRpcClientConfig { @@ -62,10 +66,58 @@ impl Default for BlockchainRpcClientConfig { min_broadcast_timeout: Duration::from_millis(100), too_new_archive_threshold: 4, download_retries: 10, + request_history_size: Default::default(), } } } +macro_rules! define_request_history { + ( + config: $size:ident, + state: $state:ident, + default: { $($name:ident = $default:expr,)*$(,)? }$(,)? + ) => { + #[derive(Debug, Clone, Serialize, Deserialize)] + #[serde(default)] + pub struct $size { + $(pub $name: Option,)* + } + + impl $size { + pub fn build_history(&self) -> $state { + $state { + $($name: self.$name.map(RequestHistory::with_capacity),)* + } + } + } + + impl Default for $size { + fn default() -> Self { + Self { + $($name: $default,)* + } + } + } + + pub struct $state { + $(pub $name: Option,)* + } + }; +} + +define_request_history! { + config: BlockchainRpcRequestHistorySize, + state: BlockchainRpcRequestHistory, + default: { + get_block_full = Some(20), + get_next_block_full = Some(20), + get_key_block_proof = Some(20), + get_zerostate_proof = Some(40), + get_persistent_state_info = Some(40), + get_archive_info = Some(20), + }, +} + pub struct BlockchainRpcClientBuilder { config: BlockchainRpcClientConfig, mandatory_fields: MandatoryFields, @@ -74,6 +126,8 @@ pub struct BlockchainRpcClientBuilder { impl BlockchainRpcClientBuilder { pub fn build(self) -> BlockchainRpcClient { + let history = self.config.request_history_size.build_history(); + BlockchainRpcClient { inner: Arc::new(Inner { config: self.config, @@ -89,6 +143,7 @@ impl BlockchainRpcClientBuilder { ) .expect("correct quantile"), ), + history, }), } } @@ -252,10 +307,13 @@ impl BlockchainRpcClient { ) -> Result, Error> { let client = &self.inner.overlay_client; let data = client - .query::<_, KeyBlockIds>(&rpc::GetNextKeyBlockIds { - block_id: *block, - max_size, - }) + .query::<_, KeyBlockIds>( + &rpc::GetNextKeyBlockIds { + block_id: *block, + max_size, + }, + None, + ) .await?; Ok(data) } @@ -271,7 +329,8 @@ impl BlockchainRpcClient { ) -> Result { let overlay_client = self.inner.overlay_client.clone(); - let Some(neighbour) = overlay_client.neighbours().choose() else { + let history = self.inner.history.get_block_full.as_ref(); + let Some(neighbour) = overlay_client.neighbours().choose(history) else { return Err(Error::NoNeighbours); }; @@ -282,6 +341,7 @@ impl BlockchainRpcClient { overlay_client, neighbour, requirement, + history, retries, ) .await @@ -294,7 +354,8 @@ impl BlockchainRpcClient { ) -> Result { let overlay_client = self.inner.overlay_client.clone(); - let Some(neighbour) = overlay_client.neighbours().choose() else { + let history = self.inner.history.get_next_block_full.as_ref(); + let Some(neighbour) = overlay_client.neighbours().choose(history) else { return Err(Error::NoNeighbours); }; @@ -307,6 +368,7 @@ impl BlockchainRpcClient { overlay_client, neighbour, requirement, + history, retries, ) .await @@ -317,20 +379,30 @@ impl BlockchainRpcClient { block_id: &BlockId, ) -> Result, Error> { let client = &self.inner.overlay_client; - let data = client - .query::<_, KeyBlockProof>(&rpc::GetKeyBlockProof { - block_id: *block_id, - }) + + let res = client + .query::<_, KeyBlockProof>( + &rpc::GetKeyBlockProof { + block_id: *block_id, + }, + self.inner.history.get_key_block_proof.as_ref(), + ) .await?; - Ok(data) + + Ok(res) } pub async fn get_zerostate_proof(&self) -> Result, Error> { let client = &self.inner.overlay_client; - let data = client - .query::<_, ZerostateProof>(&rpc::GetZerostateProof) + + let res = client + .query::<_, ZerostateProof>( + &rpc::GetZerostateProof, + self.inner.history.get_zerostate_proof.as_ref(), + ) .await?; - Ok(data) + + Ok(res) } pub async fn get_persistent_state_info( @@ -338,12 +410,17 @@ impl BlockchainRpcClient { block_id: &BlockId, ) -> Result, Error> { let client = &self.inner.overlay_client; - let data = client - .query::<_, PersistentStateInfo>(&rpc::GetPersistentShardStateInfo { - block_id: *block_id, - }) + + let res = client + .query::<_, PersistentStateInfo>( + &rpc::GetPersistentShardStateInfo { + block_id: *block_id, + }, + self.inner.history.get_persistent_state_info.as_ref(), + ) .await?; - Ok(data) + + Ok(res) } pub async fn get_persistent_state_part( @@ -360,6 +437,7 @@ impl BlockchainRpcClient { block_id: *block_id, offset, }), + None, ) .await?; Ok(data) @@ -372,11 +450,14 @@ impl BlockchainRpcClient { ) -> Result { const NEIGHBOUR_COUNT: usize = 10; + let history = self.inner.history.get_persistent_state_info.as_ref(); + // Get reliable neighbours with higher weight - let neighbours = self - .overlay_client() - .neighbours() - .choose_multiple(NEIGHBOUR_COUNT, NeighbourType::Reliable); + let neighbours = self.overlay_client().neighbours().choose_multiple( + NEIGHBOUR_COUNT, + NeighbourType::Reliable, + history, + ); let req = match kind { PersistentStateKind::Shard => Request::from_tl(rpc::GetPersistentShardStateInfo { @@ -389,10 +470,11 @@ impl BlockchainRpcClient { let mut futures = FuturesUnordered::new(); for neighbour in neighbours { - futures.push( - self.overlay_client() - .query_raw::(neighbour.clone(), req.clone()), - ); + futures.push(self.overlay_client().query_raw::( + neighbour.clone(), + req.clone(), + history, + )); } let mut err = None; @@ -424,7 +506,9 @@ impl BlockchainRpcClient { neighbour, }); } - PersistentStateInfo::NotFound => {} + PersistentStateInfo::NotFound => { + handle.ignore(); + } } } @@ -496,10 +580,12 @@ impl BlockchainRpcClient { const NEIGHBOUR_COUNT: usize = 10; // Get reliable neighbours with higher weight - let neighbours = self - .overlay_client() - .neighbours() - .choose_multiple(NEIGHBOUR_COUNT, NeighbourType::Reliable); + let history = self.inner.history.get_archive_info.as_ref(); + let neighbours = self.overlay_client().neighbours().choose_multiple( + NEIGHBOUR_COUNT, + NeighbourType::Reliable, + history, + ); // Find a neighbour which has the requested archive let pending_archive = 'info: { @@ -510,7 +596,10 @@ impl BlockchainRpcClient { let mut futures = FuturesUnordered::new(); for neighbour in neighbours { - futures.push(self.overlay_client().query_raw(neighbour, req.clone())); + futures.push( + self.overlay_client() + .query_raw(neighbour, req.clone(), history), + ); } let mut err = None; @@ -539,10 +628,10 @@ impl BlockchainRpcClient { ArchiveInfo::TooNew => { new_archive_count += 1; - handle.accept(); + handle.ignore(); } ArchiveInfo::NotFound => { - handle.accept(); + handle.ignore(); } } } @@ -636,6 +725,7 @@ struct Inner { overlay_client: PublicOverlayClient, broadcast_listener: Option>, response_tracker: Mutex, + history: BlockchainRpcRequestHistory, } pub enum PendingArchiveResponse { @@ -693,13 +783,14 @@ async fn download_block_inner( overlay_client: PublicOverlayClient, neighbour: Neighbour, requirement: DataRequirement, + history: Option<&RequestHistory>, retries: usize, ) -> Result { let response = overlay_client - .query_raw::(neighbour.clone(), req) + .query_raw::(neighbour.clone(), req, history) .await?; - let (handle, block_full) = response.split(); + let (mut handle, block_full) = response.split(); let BlockFull::Found { block_id, @@ -708,9 +799,10 @@ async fn download_block_inner( queue_diff: queue_diff_data, } = block_full else { + handle.attach_history(history.cloned()); match requirement { DataRequirement::Optional => { - handle.accept(); + handle.ignore(); } DataRequirement::Expected => { handle.reject(); @@ -726,6 +818,10 @@ async fn download_block_inner( }); }; + if let Some(history) = history { + history.got_response(handle.neighbour().peer_id(), true); + } + const PARALLEL_REQUESTS: usize = 10; let target_size = block_data.size.get(); @@ -813,6 +909,8 @@ async fn download_block_inner( .map(Bytes::from) .map_err(Error::Internal)?; + handle.accept(); + Ok(BlockDataFullWithNeighbour { data: Some(BlockDataFull { block_id, @@ -834,7 +932,7 @@ async fn download_with_retries( let mut retries = 0; loop { match overlay_client - .query_raw::(neighbour.clone(), req.clone()) + .query_raw::(neighbour.clone(), req.clone(), None) .await { Ok(r) => { @@ -912,7 +1010,7 @@ mod tests { let from = offset as usize; let to = std::cmp::min(from + CHUNK_SIZE, compressed_data.len()); let chunk = compressed_data.slice(from..to); - let handle = QueryResponseHandle::with_roundtrip_ms(neighbour.clone(), 100); + let handle = QueryResponseHandle::with_roundtrip_ms(neighbour.clone(), 100, None); futures_util::future::ready(Ok::<_, StubError>((handle, chunk))) }, |result, chunk| { diff --git a/core/src/overlay_client/mod.rs b/core/src/overlay_client/mod.rs index a25954c56..5c837b04f 100644 --- a/core/src/overlay_client/mod.rs +++ b/core/src/overlay_client/mod.rs @@ -9,7 +9,7 @@ use tycho_network::{ConnectionError, Network, PublicOverlay, Request, UnknownPee pub use self::config::{NeighborsConfig, PublicOverlayClientConfig, ValidatorsConfig}; pub use self::neighbour::{Neighbour, NeighbourStats, PunishReason}; -pub use self::neighbours::{NeighbourType, Neighbours}; +pub use self::neighbours::{NeighbourType, Neighbours, RequestHistory}; pub use self::validators::{Validator, ValidatorSetPeers, ValidatorsResolver}; use crate::proto::overlay; @@ -131,12 +131,16 @@ impl PublicOverlayClient { self.inner.send_impl(neighbour, req).await } - pub async fn query(&self, data: R) -> Result, Error> + pub async fn query( + &self, + data: R, + history: Option<&RequestHistory>, + ) -> Result, Error> where R: tl_proto::TlWrite, for<'a> A: tl_proto::TlRead<'a, Repr = tl_proto::Boxed>, { - self.inner.query(data).await + self.inner.query(data, history).await } #[inline] @@ -144,11 +148,15 @@ impl PublicOverlayClient { &self, neighbour: Neighbour, req: Request, + history: Option<&RequestHistory>, ) -> Result, Error> where for<'a> A: tl_proto::TlRead<'a, Repr = tl_proto::Boxed>, { - self.inner.query_impl(neighbour, req).await?.parse() + self.inner + .query_impl(neighbour, req, history) + .await? + .parse() } } @@ -222,7 +230,7 @@ impl Inner { }; let peer_id = *neighbour.peer_id(); - match self.query_impl(neighbour.clone(), req.clone()).await { + match self.query_impl(neighbour.clone(), req.clone(), None).await { Ok(res) => match tl_proto::deserialize::(&res.data) { Ok(_) => { res.accept(); @@ -337,7 +345,7 @@ impl Inner { where R: tl_proto::TlWrite, { - let Some(neighbour) = self.neighbours.choose() else { + let Some(neighbour) = self.neighbours.choose(None) else { return Err(Error::NoNeighbours); }; @@ -352,16 +360,20 @@ impl Inner { res.map_err(Error::NetworkError) } - async fn query(&self, data: R) -> Result, Error> + async fn query( + &self, + data: R, + history: Option<&RequestHistory>, + ) -> Result, Error> where R: tl_proto::TlWrite, for<'a> A: tl_proto::TlRead<'a, Repr = tl_proto::Boxed>, { - let Some(neighbour) = self.neighbours.choose() else { + let Some(neighbour) = self.neighbours.choose(history) else { return Err(Error::NoNeighbours); }; - self.query_impl(neighbour, Request::from_tl(data)) + self.query_impl(neighbour, Request::from_tl(data), history) .await? .parse() } @@ -379,7 +391,7 @@ impl Inner { match res { Ok(response) => { - neighbour.track_request(&roundtrip, response.is_ok()); + neighbour.track_request(&roundtrip, Some(response.is_ok())); if let Err(e) = &response { apply_network_error(e, &neighbour); @@ -388,7 +400,7 @@ impl Inner { response.map_err(Error::NetworkError) } Err(_) => { - neighbour.track_request(&roundtrip, false); + neighbour.track_request(&roundtrip, Some(false)); neighbour.punish(PunishReason::Slow); Err(Error::Timeout) } @@ -399,6 +411,7 @@ impl Inner { &self, neighbour: Neighbour, req: Request, + history: Option<&RequestHistory>, ) -> Result, Error> { let started_at = Instant::now(); @@ -415,14 +428,15 @@ impl Inner { data: response.body, roundtrip_ms: roundtrip.as_millis() as u64, neighbour, + history: history.cloned(), }), Ok(Err(e)) => { - neighbour.track_request(&roundtrip, false); + neighbour.track_request(&roundtrip, Some(false)); apply_network_error(&e, &neighbour); Err(Error::NetworkError(e)) } Err(_) => { - neighbour.track_request(&roundtrip, false); + neighbour.track_request(&roundtrip, Some(false)); neighbour.punish(PunishReason::Slow); Err(Error::Timeout) } @@ -454,6 +468,7 @@ pub struct QueryResponse { data: A, neighbour: Neighbour, roundtrip_ms: u64, + history: Option, } impl QueryResponse { @@ -461,22 +476,36 @@ impl QueryResponse { &self.data } + pub fn neighbour(&self) -> &Neighbour { + &self.neighbour + } + pub fn split(self) -> (QueryResponseHandle, A) { - let handle = QueryResponseHandle::with_roundtrip_ms(self.neighbour, self.roundtrip_ms); + let handle = + QueryResponseHandle::with_roundtrip_ms(self.neighbour, self.roundtrip_ms, self.history); (handle, self.data) } - pub fn accept(self) -> (Neighbour, A) { - self.track_request(true); + pub fn accept(mut self) -> (Neighbour, A) { + self.track_request(Some(true)); + (self.neighbour, self.data) + } + + pub fn reject(mut self) -> (Neighbour, A) { + self.track_request(Some(false)); (self.neighbour, self.data) } - pub fn reject(self) -> (Neighbour, A) { - self.track_request(false); + pub fn ignore(mut self) -> (Neighbour, A) { + self.track_request(None); (self.neighbour, self.data) } - fn track_request(&self, success: bool) { + fn track_request(&mut self, success: Option) { + if let Some(history) = self.history.take() { + history.got_response(self.neighbour.peer_id(), matches!(success, Some(true))); + } + self.neighbour .track_request(&Duration::from_millis(self.roundtrip_ms), success); } @@ -500,6 +529,7 @@ impl QueryResponse { data, roundtrip_ms: self.roundtrip_ms, neighbour: self.neighbour, + history: self.history, }), overlay::Response::Err(code) => { self.reject(); @@ -512,27 +542,50 @@ impl QueryResponse { pub struct QueryResponseHandle { neighbour: Neighbour, roundtrip_ms: u64, + history: Option, } impl QueryResponseHandle { - pub fn with_roundtrip_ms(neighbour: Neighbour, roundtrip_ms: u64) -> Self { + pub fn with_roundtrip_ms( + neighbour: Neighbour, + roundtrip_ms: u64, + history: Option, + ) -> Self { Self { neighbour, roundtrip_ms, + history, } } - pub fn accept(self) -> Neighbour { - self.track_request(true); + pub fn attach_history(&mut self, history: Option) { + self.history = history; + } + + pub fn neighbour(&self) -> &Neighbour { + &self.neighbour + } + + pub fn accept(mut self) -> Neighbour { + self.track_request(Some(true)); + self.neighbour + } + + pub fn reject(mut self) -> Neighbour { + self.track_request(Some(false)); self.neighbour } - pub fn reject(self) -> Neighbour { - self.track_request(false); + pub fn ignore(mut self) -> Neighbour { + self.track_request(None); self.neighbour } - fn track_request(&self, success: bool) { + fn track_request(&mut self, success: Option) { + if let Some(history) = self.history.take() { + history.got_response(self.neighbour.peer_id(), matches!(success, Some(true))); + } + self.neighbour .track_request(&Duration::from_millis(self.roundtrip_ms), success); } diff --git a/core/src/overlay_client/neighbour.rs b/core/src/overlay_client/neighbour.rs index de92fe8b3..67aeb767d 100644 --- a/core/src/overlay_client/neighbour.rs +++ b/core/src/overlay_client/neighbour.rs @@ -65,7 +65,7 @@ impl Neighbour { Some(Duration::from_millis(roundtrip as u64)) } - pub fn track_request(&self, roundtrip: &Duration, success: bool) { + pub fn track_request(&self, roundtrip: &Duration, success: Option) { let roundtrip = truncate_time(roundtrip); self.inner.stats.write().update(roundtrip, success); } @@ -172,21 +172,24 @@ impl TrackedStats { (score >= Self::SCORE_THRESHOLD).then_some(score) } - fn update(&mut self, roundtrip: u16, success: bool) { + fn update(&mut self, roundtrip: u16, success: Option) { const SUCCESS_REQUEST_SCORE: u8 = 8; const FAILED_REQUEST_PENALTY: u8 = 8; - self.failed_requests_history <<= 1; - if success { - self.score = std::cmp::min( - self.score.saturating_add(SUCCESS_REQUEST_SCORE), - Self::MAX_SCORE, - ); - } else { - self.score = self.score.saturating_sub(FAILED_REQUEST_PENALTY); - self.failed += 1; - self.failed_requests_history |= 1; + if let Some(success) = success { + self.failed_requests_history <<= 1; + if success { + self.score = std::cmp::min( + self.score.saturating_add(SUCCESS_REQUEST_SCORE), + Self::MAX_SCORE, + ); + } else { + self.score = self.score.saturating_sub(FAILED_REQUEST_PENALTY); + self.failed += 1; + self.failed_requests_history |= 1; + } } + self.total += 1; let roundtrip_buffer = &mut self.roundtrip; diff --git a/core/src/overlay_client/neighbours.rs b/core/src/overlay_client/neighbours.rs index 86ca880fb..53aadbbc1 100644 --- a/core/src/overlay_client/neighbours.rs +++ b/core/src/overlay_client/neighbours.rs @@ -1,12 +1,13 @@ +use std::num::NonZeroUsize; use std::sync::Arc; use arc_swap::ArcSwap; -use parking_lot::Mutex; +use lru::LruCache; +use parking_lot::{Mutex, RwLock}; use rand::Rng; -use rand::distr::uniform::{UniformInt, UniformSampler}; use tokio::sync::Notify; use tycho_network::PeerId; -use tycho_util::FastHashSet; +use tycho_util::{FastHashSet, FastHasherState}; use crate::overlay_client::neighbour::Neighbour; #[derive(Clone)] @@ -46,14 +47,19 @@ impl Neighbours { &self.inner.changed } - pub fn choose(&self) -> Option { + pub fn choose(&self, history: Option<&RequestHistory>) -> Option { let selection_index = self.inner.selection_index.lock(); - selection_index.choose(&mut rand::rng()) + selection_index.choose(&mut rand::rng(), history) } - pub fn choose_multiple(&self, n: usize, neighbour_type: NeighbourType) -> Vec { + pub fn choose_multiple( + &self, + n: usize, + neighbour_type: NeighbourType, + history: Option<&RequestHistory>, + ) -> Vec { let selection_index = self.inner.selection_index.lock(); - selection_index.choose_multiple(&mut rand::rng(), n, neighbour_type) + selection_index.choose_multiple(&mut rand::rng(), n, neighbour_type, history) } /// Tries to apply neighbours score to selection index. @@ -209,15 +215,14 @@ struct Inner { struct SelectionIndex { /// Neighbour indices with cumulative weight. indices_with_weights: Vec<(Neighbour, u32)>, - /// Optional uniform distribution `[0; total_weight)`. - distribution: Option>, + total_weight: u32, } impl SelectionIndex { fn new(capacity: usize) -> Self { Self { indices_with_weights: Vec::with_capacity(capacity), - distribution: None, + total_weight: 0, } } @@ -232,18 +237,53 @@ impl SelectionIndex { } } - self.distribution = UniformInt::new(0, total_weight).ok(); + self.total_weight = total_weight; // TODO: fallback to uniform sample from any neighbour } - fn choose(&self, rng: &mut R) -> Option { - let chosen_weight = self.distribution.as_ref()?.sample(rng); + fn choose( + &self, + rng: &mut R, + history: Option<&RequestHistory>, + ) -> Option { + if self.total_weight == 0 || self.indices_with_weights.is_empty() { + return None; + } - // Find the first item which has a weight higher than the chosen weight. - let i = self - .indices_with_weights - .partition_point(|(_, w)| *w <= chosen_weight); + let i = if let Some(history) = history { + let mut adjusted_total_weight = 0; + let mut distrubution = vec![0; self.indices_with_weights.len()]; + + let history = history.state.read(); + let mut prev_total_weight = 0; + for ((neighbour, total_weight), distrubution) in + std::iter::zip(&self.indices_with_weights, &mut distrubution) + { + let fine = history + .fines + .peek(neighbour.peer_id()) + .copied() + .unwrap_or_default(); + + let weight = total_weight - prev_total_weight; + prev_total_weight = *total_weight; + + debug_assert!(weight > 0); + + let weight = std::cmp::max(weight.saturating_sub(fine as u32), 1); + adjusted_total_weight += weight; + *distrubution = adjusted_total_weight; + } + drop(history); + + let chosen_weight = rng.random_range(0..adjusted_total_weight); + distrubution.partition_point(|&w| w <= chosen_weight) + } else { + let chosen_weight = rng.random_range(0..self.total_weight); + self.indices_with_weights + .partition_point(|(_, w)| *w <= chosen_weight) + }; self.indices_with_weights .get(i) @@ -256,6 +296,7 @@ impl SelectionIndex { rng: &mut R, mut n: usize, neighbour_type: NeighbourType, + history: Option<&RequestHistory>, ) -> Vec { struct Element<'a> { key: f64, @@ -292,18 +333,46 @@ impl SelectionIndex { } let mut candidates = Vec::with_capacity(self.indices_with_weights.len()); - let mut prev_total_weight = None; - for (neighbour, total_weight) in &self.indices_with_weights { - let weight = match prev_total_weight { - Some(prev) => total_weight - prev, - None => *total_weight, - }; - prev_total_weight = Some(*total_weight); - - debug_assert!(weight > 0); - - let key = rng.random::().powf(1.0 / weight as f64); - candidates.push(Element { key, neighbour }); + let mut prev_total_weight = 0; + match history { + Some(history) => { + let history = history.state.read(); + for (neighbour, total_weight) in &self.indices_with_weights { + let fine = history + .fines + .peek(neighbour.peer_id()) + .copied() + .unwrap_or_default(); + + let weight = total_weight - prev_total_weight; + prev_total_weight = *total_weight; + + debug_assert!(weight > 0); + + let weight = std::cmp::max(weight.saturating_sub(fine as u32), 1); + + candidates.push(Element { + key: weight as f64, + neighbour, + }); + } + drop(history); + + for item in candidates.iter_mut() { + item.key = rng.random::().powf(1.0 / item.key); + } + } + None => { + for (neighbour, total_weight) in &self.indices_with_weights { + let weight = total_weight - prev_total_weight; + prev_total_weight = *total_weight; + + debug_assert!(weight > 0); + + let key = rng.random::().powf(1.0 / weight as f64); + candidates.push(Element { key, neighbour }); + } + } } // Partially sort the array to find the `n` elements with the greatest @@ -336,3 +405,51 @@ pub enum NeighbourType { All, Reliable, } + +#[derive(Clone)] +pub struct RequestHistory { + state: Arc>, +} + +impl RequestHistory { + // NOTE: This value is chosen to just slightly outweight the "fast node" factor. + const MAX_FINE: u8 = 16; + const INITIAL_FINE: u8 = 1; + + pub fn with_capacity(capacity: usize) -> Self { + Self { + state: Arc::new(RwLock::new(RequestHistoryState { + fines: LruCache::with_hasher( + NonZeroUsize::new(capacity).unwrap_or(NonZeroUsize::MIN), + Default::default(), + ), + })), + } + } + + pub fn got_response(&self, peer_id: &PeerId, useful: bool) { + let mut state = self.state.write(); + + if useful { + if let Some(fine) = state.fines.get_mut(peer_id) { + *fine >>= 1; + if *fine == 0 { + state.fines.pop(peer_id); + } + } + } else { + let mut apply_fine = true; + let fine = state.fines.get_or_insert_mut(*peer_id, || { + apply_fine = false; + Self::INITIAL_FINE + }); + if apply_fine && *fine < Self::MAX_FINE { + *fine <<= 1; + } + } + } +} + +struct RequestHistoryState { + fines: LruCache, +} diff --git a/core/tests/overlay_client.rs b/core/tests/overlay_client.rs index 73f1a5db6..9ebef21d4 100644 --- a/core/tests/overlay_client.rs +++ b/core/tests/overlay_client.rs @@ -45,7 +45,7 @@ pub async fn test() { let slice = initial_peers.as_slice(); while i < 1000 { // let start = Instant::now(); - let n_opt = neighbours.choose(); + let n_opt = neighbours.choose(None); // let end = Instant::now(); if let Some(n) = n_opt { @@ -56,10 +56,10 @@ pub async fn test() { let answer = indices[index].sample(&mut rng); if answer == 0 { println!("Success request to peer: {}", n.peer_id()); - n.track_request(&Duration::from_millis(200), true); + n.track_request(&Duration::from_millis(200), Some(true)); } else { println!("Failed request to peer: {}", n.peer_id()); - n.track_request(&Duration::from_millis(200), false); + n.track_request(&Duration::from_millis(200), Some(false)); } neighbours.try_apply_score(0);