diff --git a/mycelium/src/lib.rs b/mycelium/src/lib.rs index 5a4c8c6..238b82a 100644 --- a/mycelium/src/lib.rs +++ b/mycelium/src/lib.rs @@ -35,6 +35,7 @@ pub mod peer_manager; pub mod router; mod router_id; mod routing_table; +mod rr_cache; mod seqno_cache; mod sequence_number; mod source_table; diff --git a/mycelium/src/rr_cache.rs b/mycelium/src/rr_cache.rs new file mode 100644 index 0000000..913c6af --- /dev/null +++ b/mycelium/src/rr_cache.rs @@ -0,0 +1,100 @@ +//! This module contains a cache implementation for route requests + +use std::{ + net::{IpAddr, Ipv6Addr}, + sync::Arc, +}; + +use dashmap::DashMap; +use tokio::time::{Duration, Instant}; +use tracing::trace; + +use crate::{babel::RouteRequest, peer::Peer, subnet::Subnet, task::AbortHandle}; + +/// Clean the route request cache every 5 seconds +const CACHE_CLEANING_INTERVAL: Duration = Duration::from_secs(5); + +/// IP used for the [`Subnet`] in the cache in case there is no prefix specified. +const GLOBAL_SUBNET_IP: IpAddr = IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)); + +/// Prefix size to use for the [`Subnet`] in case there is no prefix specified. +const GLOBAL_SUBNET_PREFIX_SIZE: u8 = 0; + +/// A self cleaning cache for route requests. +pub struct RouteRequestCache { + /// The actual cache, mapping an instance of a route request to the peers which we've sent this + /// to. + cache: Arc>, + + _cleanup_task: AbortHandle, +} + +struct RouteRequestInfo { + /// The lowest generation we've forwarded. + generation: u8, + /// Peers which we've sent this route request to already. + receivers: Vec, + /// The moment we've sent this route request + sent: Instant, +} + +impl RouteRequestCache { + /// Create a new cache which cleans entries which are older than the given expiration. + /// + /// The cache cleaning is done periodically, so entries might live slightly longer than the + /// allowed expiration. + pub fn new(expiration: Duration) -> Self { + let cache = Arc::new(DashMap::with_hasher(ahash::RandomState::new())); + + let _cleanup_task = tokio::spawn({ + let cache = cache.clone(); + async move { + loop { + tokio::time::sleep(CACHE_CLEANING_INTERVAL).await; + + trace!("Cleaning route request cache"); + + cache.retain(|subnet, info: &mut RouteRequestInfo| { + if info.sent.elapsed() < expiration { + false + } else { + trace!(%subnet, "Removing exired route request from cache"); + true + } + }); + } + } + }) + .abort_handle() + .into(); + + Self { + cache, + _cleanup_task, + } + } + + /// Record a route request which has been sent to peers. + pub fn sent_route_request(&self, rr: RouteRequest, receivers: Vec) { + let subnet = rr.prefix().unwrap_or( + Subnet::new(GLOBAL_SUBNET_IP, GLOBAL_SUBNET_PREFIX_SIZE) + .expect("Static global IPv6 subnet is valid; qed"), + ); + let generation = rr.generation(); + + let rri = RouteRequestInfo { + generation, + receivers, + sent: Instant::now(), + }; + + self.cache.insert(subnet, rri); + } + + /// Get cached info about a route request for a subnet, if it exists. + pub fn info(&self, subnet: Subnet) -> Option<(u8, Vec)> { + self.cache + .get(&subnet) + .map(|rri| (rri.generation, rri.receivers.clone())) + } +}