diff --git a/src/endpoint.rs b/src/endpoint.rs index 0e1f5c8..adb3532 100644 --- a/src/endpoint.rs +++ b/src/endpoint.rs @@ -12,13 +12,15 @@ use crate::proto::*; use crate::util::*; /// This trait should be implemented by all messages your application -/// wants to handle (click to read more). +/// wants to handle pub trait Message: Serialize + for<'de> Deserialize<'de> + Send + Sync { type Response: Serialize + for<'de> Deserialize<'de> + Send + Sync; } -pub(crate) type DynEndpoint = Box; - +/// This trait should be implemented by an object of your application +/// that can handle a message of type `M`. +/// +/// The handler object should be in an Arc, see `Endpoint::set_handler` #[async_trait] pub trait EndpointHandler: Send + Sync where @@ -27,6 +29,27 @@ where async fn handle(self: &Arc, m: M, from: NodeID) -> M::Response; } +/// If one simply wants to use an endpoint in a client fashion, +/// without locally serving requests to that endpoint, +/// use the unit type `()` as the handler type: +/// it will panic if it is ever made to handle request. +#[async_trait] +impl EndpointHandler for () { + async fn handle(self: &Arc<()>, _m: M, _from: NodeID) -> M::Response { + panic!("This endpoint should not have a local handler."); + } +} + +/// This struct represents an endpoint for message of type `M`. +/// +/// Creating a new endpoint is done by calling `NetApp::endpoint`. +/// An endpoint is identified primarily by its path, which is specified +/// at creation time. +/// +/// An `Endpoint` is used both to send requests to remote nodes, +/// and to specify the handler for such requests on the local node. +/// The type `H` represents the type of the handler object for +/// endpoint messages (see `EndpointHandler`). pub struct Endpoint where M: Message, @@ -51,9 +74,15 @@ where handler: ArcSwapOption::from(None), } } + + /// Set the object that is responsible of handling requests to + /// this endpoint on the local node. pub fn set_handler(&self, h: Arc) { self.handler.swap(Some(h)); } + + /// Call this endpoint on a remote node (or on the local node, + /// for that matter) pub async fn call( &self, target: &NodeID, @@ -84,6 +113,10 @@ where } } +// ---- Internal stuff ---- + +pub(crate) type DynEndpoint = Box; + #[async_trait] pub(crate) trait GenericEndpoint { async fn handle(&self, buf: &[u8], from: NodeID) -> Result, Error>; diff --git a/src/netapp.rs b/src/netapp.rs index bffa0e1..7fe1e71 100644 --- a/src/netapp.rs +++ b/src/netapp.rs @@ -125,21 +125,30 @@ impl NetApp { .store(Some(Arc::new(Box::new(handler)))); } - pub fn endpoint(self: &Arc, name: String) -> Arc> + /// Create a new endpoint with path `path`, + /// that handles messages of type `M`. + /// `H` is the type of the object that should handle requests + /// to this endpoint on the local node. If you don't want + /// to handle request on the local node (e.g. if this node + /// is only a client in the network), define the type `H` + /// to be `()`. + /// This function will panic if the endpoint has already been + /// created. + pub fn endpoint(self: &Arc, path: String) -> Arc> where M: Message + 'static, H: EndpointHandler + 'static, { - let endpoint = Arc::new(Endpoint::::new(self.clone(), name.clone())); + let endpoint = Arc::new(Endpoint::::new(self.clone(), path.clone())); let endpoint_arc = EndpointArc(endpoint.clone()); if self .endpoints .write() .unwrap() - .insert(name.clone(), Box::new(endpoint_arc)) + .insert(path.clone(), Box::new(endpoint_arc)) .is_some() { - panic!("Redefining endpoint: {}", name); + panic!("Redefining endpoint: {}", path); }; endpoint } diff --git a/src/peering/fullmesh.rs b/src/peering/fullmesh.rs index 1162048..c521590 100644 --- a/src/peering/fullmesh.rs +++ b/src/peering/fullmesh.rs @@ -4,6 +4,7 @@ use std::sync::atomic::{self, AtomicU64}; use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; +use arc_swap::ArcSwap; use async_trait::async_trait; use log::{debug, info, trace, warn}; use serde::{Deserialize, Serialize}; @@ -46,7 +47,7 @@ impl Message for PeerListMessage { // -- Algorithm data structures -- #[derive(Debug)] -struct PeerInfo { +struct PeerInfoInternal { addr: SocketAddr, state: PeerConnState, last_seen: Option, @@ -54,40 +55,49 @@ struct PeerInfo { } #[derive(Copy, Clone, Debug)] -pub struct PeerInfoPub { +pub struct PeerInfo { + /// The node's identifier (its public key) pub id: NodeID, + /// The node's network address pub addr: SocketAddr, + /// The current status of our connection to this node pub state: PeerConnState, + /// The last time at which the node was seen pub last_seen: Option, + /// The average ping to this node on recent observations (if at least one ping value is known) pub avg_ping: Option, + /// The maximum observed ping to this node on recent observations (if at least one + /// ping value is known) pub max_ping: Option, + /// The median ping to this node on recent observations (if at least one ping value + /// is known) pub med_ping: Option, } -// PeerConnState: possible states for our tentative connections to given peer -// This module is only interested in recording connection info for outgoing -// TCP connections +/// PeerConnState: possible states for our tentative connections to given peer +/// This structure is only interested in recording connection info for outgoing +/// TCP connections #[derive(Copy, Clone, Debug, PartialEq)] pub enum PeerConnState { - // This entry represents ourself + /// This entry represents ourself (the local node) Ourself, - // We currently have a connection to this peer + /// We currently have a connection to this peer Connected, - // Our next connection tentative (the nth, where n is the first value) - // will be at given Instant + /// Our next connection tentative (the nth, where n is the first value of the tuple) + /// will be at given Instant Waiting(usize, Instant), - // A connection tentative is in progress + /// A connection tentative is in progress (the nth, where n is the value stored) Trying(usize), - // We abandonned trying to connect to this peer (too many failed attempts) + /// We abandonned trying to connect to this peer (too many failed attempts) Abandonned, } struct KnownHosts { - list: HashMap, + list: HashMap, hash: hash::Digest, } @@ -100,7 +110,7 @@ impl KnownHosts { fn update_hash(&mut self) { self.hash = Self::calculate_hash(&self.list); } - fn map_into_vec(input: &HashMap) -> Vec<(NodeID, SocketAddr)> { + fn map_into_vec(input: &HashMap) -> Vec<(NodeID, SocketAddr)> { let mut list = Vec::with_capacity(input.len()); for (id, peer) in input.iter() { if peer.state == PeerConnState::Connected || peer.state == PeerConnState::Ourself { @@ -109,35 +119,43 @@ impl KnownHosts { } list } - fn calculate_hash(input: &HashMap) -> hash::Digest { + fn calculate_hash(input: &HashMap) -> hash::Digest { let mut list = Self::map_into_vec(input); list.sort(); let mut hash_state = hash::State::new(); for (id, addr) in list { hash_state.update(&id[..]); - hash_state.update(&format!("{}", addr).into_bytes()[..]); + hash_state.update(&format!("{}\n", addr).into_bytes()[..]); } hash_state.finalize() } } +/// A "Full Mesh" peering strategy is a peering strategy that tries +/// to establish and maintain a direct connection with all of the +/// known nodes in the network. pub struct FullMeshPeeringStrategy { netapp: Arc, known_hosts: RwLock, - next_ping_id: AtomicU64, + public_peer_list: ArcSwap>, + next_ping_id: AtomicU64, ping_endpoint: Arc>, peer_list_endpoint: Arc>, } impl FullMeshPeeringStrategy { + /// Create a new Full Mesh peering strategy. + /// The strategy will not be run until `.run()` is called and awaited. + /// Once that happens, the peering strategy will try to connect + /// to all of the nodes specified in the bootstrap list. pub fn new(netapp: Arc, bootstrap_list: Vec<(NodeID, SocketAddr)>) -> Arc { let mut known_hosts = KnownHosts::new(); for (id, addr) in bootstrap_list { if id != netapp.id { known_hosts.list.insert( id, - PeerInfo { + PeerInfoInternal { addr, state: PeerConnState::Waiting(0, Instant::now()), last_seen: None, @@ -150,6 +168,7 @@ impl FullMeshPeeringStrategy { let strat = Arc::new(Self { netapp: netapp.clone(), known_hosts: RwLock::new(known_hosts), + public_peer_list: ArcSwap::new(Arc::new(Vec::new())), next_ping_id: AtomicU64::new(42), ping_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/Ping".into()), peer_list_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/PeerList".into()), @@ -173,6 +192,8 @@ impl FullMeshPeeringStrategy { strat } + /// Run the full mesh peering strategy. + /// This future exits when the `must_exit` watch becomes true. pub async fn run(self: Arc, must_exit: watch::Receiver) { while !*must_exit.borrow() { // 1. Read current state: get list of connected peers (ping them) @@ -229,6 +250,7 @@ impl FullMeshPeeringStrategy { } } } + self.update_public_peer_list(&known_hosts); } // 4. Sleep before next loop iteration @@ -236,6 +258,48 @@ impl FullMeshPeeringStrategy { } } + /// Returns a list of currently known peers in the network. + pub fn get_peer_list(&self) -> Arc> { + self.public_peer_list.load_full() + } + + // -- internal stuff -- + + fn update_public_peer_list(&self, known_hosts: &KnownHosts) { + let mut pub_peer_list = Vec::with_capacity(known_hosts.list.len()); + for (id, info) in known_hosts.list.iter() { + let mut pings = info.ping.iter().cloned().collect::>(); + pings.sort(); + if !pings.is_empty() { + pub_peer_list.push(PeerInfo { + id: *id, + addr: info.addr, + state: info.state, + last_seen: info.last_seen, + avg_ping: Some( + pings + .iter() + .fold(Duration::from_secs(0), |x, y| x + *y) + .div_f64(pings.len() as f64), + ), + max_ping: pings.last().cloned(), + med_ping: Some(pings[pings.len() / 2]), + }); + } else { + pub_peer_list.push(PeerInfo { + id: *id, + addr: info.addr, + state: info.state, + last_seen: info.last_seen, + avg_ping: None, + max_ping: None, + med_ping: None, + }); + } + } + self.public_peer_list.store(Arc::new(pub_peer_list)); + } + async fn ping(self: Arc, id: NodeID) { let peer_list_hash = self.known_hosts.read().unwrap().hash; let ping_id = self.next_ping_id.fetch_add(1u64, atomic::Ordering::Relaxed); @@ -268,6 +332,7 @@ impl FullMeshPeeringStrategy { while host.ping.len() > 10 { host.ping.pop_front(); } + self.update_public_peer_list(&known_hosts); } } if ping_resp.peer_list_hash != peer_list_hash { @@ -299,6 +364,7 @@ impl FullMeshPeeringStrategy { known_hosts.list.insert(*id, self.new_peer(id, *addr)); } } + self.update_public_peer_list(&known_hosts); } async fn try_connect(self: Arc, id: NodeID, addr: SocketAddr) { @@ -317,6 +383,7 @@ impl FullMeshPeeringStrategy { } _ => PeerConnState::Waiting(0, Instant::now() + CONN_RETRY_INTERVAL), }; + self.update_public_peer_list(&known_hosts); } } } @@ -336,6 +403,7 @@ impl FullMeshPeeringStrategy { if let Some(host) = known_hosts.list.get_mut(&id) { host.state = PeerConnState::Connected; known_hosts.update_hash(); + self.update_public_peer_list(&known_hosts); } } } @@ -347,53 +415,18 @@ impl FullMeshPeeringStrategy { if let Some(host) = known_hosts.list.get_mut(&id) { host.state = PeerConnState::Waiting(0, Instant::now()); known_hosts.update_hash(); + self.update_public_peer_list(&known_hosts); } } } - pub fn get_peer_list(&self) -> Vec { - let known_hosts = self.known_hosts.read().unwrap(); - let mut ret = Vec::with_capacity(known_hosts.list.len()); - for (id, info) in known_hosts.list.iter() { - let mut pings = info.ping.iter().cloned().collect::>(); - pings.sort(); - if !pings.is_empty() { - ret.push(PeerInfoPub { - id: *id, - addr: info.addr, - state: info.state, - last_seen: info.last_seen, - avg_ping: Some( - pings - .iter() - .fold(Duration::from_secs(0), |x, y| x + *y) - .div_f64(pings.len() as f64), - ), - max_ping: pings.last().cloned(), - med_ping: Some(pings[pings.len() / 2]), - }); - } else { - ret.push(PeerInfoPub { - id: *id, - addr: info.addr, - state: info.state, - last_seen: info.last_seen, - avg_ping: None, - max_ping: None, - med_ping: None, - }); - } - } - ret - } - - fn new_peer(&self, id: &NodeID, addr: SocketAddr) -> PeerInfo { + fn new_peer(&self, id: &NodeID, addr: SocketAddr) -> PeerInfoInternal { let state = if *id == self.netapp.id { PeerConnState::Ourself } else { PeerConnState::Waiting(0, Instant::now()) }; - PeerInfo { + PeerInfoInternal { addr, state, last_seen: None, diff --git a/src/util.rs b/src/util.rs index a87f779..bf8f4cb 100644 --- a/src/util.rs +++ b/src/util.rs @@ -4,6 +4,7 @@ use log::info; use tokio::sync::watch; +/// A node's identifier, which is also its public cryptographic key pub type NodeID = sodiumoxide::crypto::sign::ed25519::PublicKey; /// Utility function: encodes any serializable value in MessagePack binary format @@ -25,6 +26,7 @@ where /// This async function returns only when a true signal was received /// from a watcher that tells us when to exit. +/// /// Usefull in a select statement to interrupt another /// future: /// ```ignore @@ -41,6 +43,8 @@ pub async fn await_exit(mut must_exit: watch::Receiver) { } } +/// Creates a watch that contains `false`, and that changes +/// to `true` when a Ctrl+C signal is received. pub fn watch_ctrl_c() -> watch::Receiver { let (send_cancel, watch_cancel) = watch::channel(false); tokio::spawn(async move {