netapp/src/peering/fullmesh.rs

608 lines
16 KiB
Rust
Raw Normal View History

use std::collections::{HashMap, VecDeque};
2020-12-02 12:30:47 +00:00
use std::net::SocketAddr;
use std::sync::atomic::{self, AtomicU64};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
2021-10-14 09:35:05 +00:00
use arc_swap::ArcSwap;
2021-10-12 15:59:46 +00:00
use async_trait::async_trait;
2020-12-02 12:30:47 +00:00
use log::{debug, info, trace, warn};
use serde::{Deserialize, Serialize};
use tokio::select;
use tokio::sync::watch;
2020-12-02 12:30:47 +00:00
use sodiumoxide::crypto::hash;
2021-10-12 15:59:46 +00:00
use crate::endpoint::*;
use crate::error::*;
2020-12-02 12:30:47 +00:00
use crate::netapp::*;
2022-07-21 15:34:53 +00:00
use crate::message::*;
use crate::NodeID;
2020-12-02 12:30:47 +00:00
const CONN_RETRY_INTERVAL: Duration = Duration::from_secs(30);
const CONN_MAX_RETRIES: usize = 10;
const PING_INTERVAL: Duration = Duration::from_secs(10);
const LOOP_DELAY: Duration = Duration::from_secs(1);
const PING_TIMEOUT: Duration = Duration::from_secs(5);
const FAILED_PING_THRESHOLD: usize = 3;
2020-12-02 12:30:47 +00:00
// -- Protocol messages --
#[derive(Serialize, Deserialize, Clone)]
2020-12-02 12:30:47 +00:00
struct PingMessage {
pub id: u64,
pub peer_list_hash: hash::Digest,
}
impl Message for PingMessage {
type Response = PingMessage;
}
#[derive(Serialize, Deserialize, Clone)]
2020-12-02 12:30:47 +00:00
struct PeerListMessage {
pub list: Vec<(NodeID, SocketAddr)>,
2020-12-02 12:30:47 +00:00
}
impl Message for PeerListMessage {
type Response = PeerListMessage;
}
// -- Algorithm data structures --
#[derive(Debug)]
2021-10-14 09:35:05 +00:00
struct PeerInfoInternal {
// addr is the currently connected address,
// or the last address we were connected to,
// or an arbitrary address some other peer gave us
2020-12-02 12:30:47 +00:00
addr: SocketAddr,
// all_addrs contains all of the addresses everyone gave us
all_addrs: Vec<SocketAddr>,
2020-12-02 12:30:47 +00:00
state: PeerConnState,
last_seen: Option<Instant>,
ping: VecDeque<Duration>,
failed_pings: usize,
2020-12-02 12:30:47 +00:00
}
#[derive(Copy, Clone, Debug)]
2021-10-14 09:35:05 +00:00
pub struct PeerInfo {
/// The node's identifier (its public key)
pub id: NodeID,
2021-10-14 09:35:05 +00:00
/// The node's network address
2020-12-02 12:30:47 +00:00
pub addr: SocketAddr,
2021-10-14 09:35:05 +00:00
/// The current status of our connection to this node
2020-12-02 12:30:47 +00:00
pub state: PeerConnState,
2021-10-14 09:35:05 +00:00
/// The last time at which the node was seen
2020-12-02 12:30:47 +00:00
pub last_seen: Option<Instant>,
2021-10-14 09:35:05 +00:00
/// The average ping to this node on recent observations (if at least one ping value is known)
2020-12-02 12:30:47 +00:00
pub avg_ping: Option<Duration>,
2021-10-14 09:35:05 +00:00
/// The maximum observed ping to this node on recent observations (if at least one
/// ping value is known)
2020-12-02 12:30:47 +00:00
pub max_ping: Option<Duration>,
2021-10-14 09:35:05 +00:00
/// The median ping to this node on recent observations (if at least one ping value
/// is known)
2020-12-02 12:30:47 +00:00
pub med_ping: Option<Duration>,
}
2021-10-14 12:13:44 +00:00
impl PeerInfo {
/// Returns true if we can currently send requests to this peer
pub fn is_up(&self) -> bool {
self.state.is_up()
}
}
2021-10-14 09:35:05 +00:00
/// PeerConnState: possible states for our tentative connections to given peer
/// This structure is only interested in recording connection info for outgoing
/// TCP connections
2020-12-02 12:30:47 +00:00
#[derive(Copy, Clone, Debug, PartialEq)]
pub enum PeerConnState {
2021-10-14 09:35:05 +00:00
/// This entry represents ourself (the local node)
2020-12-02 12:30:47 +00:00
Ourself,
2021-10-14 09:35:05 +00:00
/// We currently have a connection to this peer
2020-12-02 12:30:47 +00:00
Connected,
2021-10-14 09:35:05 +00:00
/// Our next connection tentative (the nth, where n is the first value of the tuple)
/// will be at given Instant
2020-12-02 12:30:47 +00:00
Waiting(usize, Instant),
2021-10-14 09:35:05 +00:00
/// A connection tentative is in progress (the nth, where n is the value stored)
2020-12-02 12:30:47 +00:00
Trying(usize),
2021-10-14 09:35:05 +00:00
/// We abandonned trying to connect to this peer (too many failed attempts)
2020-12-02 12:30:47 +00:00
Abandonned,
}
2021-10-14 12:13:44 +00:00
impl PeerConnState {
/// Returns true if we can currently send requests to this peer
pub fn is_up(&self) -> bool {
matches!(self, Self::Ourself | Self::Connected)
}
}
2020-12-02 12:30:47 +00:00
struct KnownHosts {
2021-10-14 09:35:05 +00:00
list: HashMap<NodeID, PeerInfoInternal>,
2020-12-02 12:30:47 +00:00
hash: hash::Digest,
}
impl KnownHosts {
fn new() -> Self {
let list = HashMap::new();
let hash = Self::calculate_hash(&list);
Self { list, hash }
}
fn update_hash(&mut self) {
self.hash = Self::calculate_hash(&self.list);
}
2021-10-14 09:35:05 +00:00
fn map_into_vec(input: &HashMap<NodeID, PeerInfoInternal>) -> Vec<(NodeID, SocketAddr)> {
2020-12-02 12:30:47 +00:00
let mut list = Vec::with_capacity(input.len());
for (id, peer) in input.iter() {
if peer.state == PeerConnState::Connected || peer.state == PeerConnState::Ourself {
2021-10-12 11:18:24 +00:00
list.push((*id, peer.addr));
2020-12-02 12:30:47 +00:00
}
}
list
}
2021-10-14 09:35:05 +00:00
fn calculate_hash(input: &HashMap<NodeID, PeerInfoInternal>) -> hash::Digest {
2020-12-02 12:30:47 +00:00
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[..]);
2021-10-14 09:35:05 +00:00
hash_state.update(&format!("{}\n", addr).into_bytes()[..]);
2020-12-02 12:30:47 +00:00
}
hash_state.finalize()
}
}
2021-10-14 09:35:05 +00:00
/// 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.
2020-12-02 12:30:47 +00:00
pub struct FullMeshPeeringStrategy {
netapp: Arc<NetApp>,
known_hosts: RwLock<KnownHosts>,
2021-10-14 09:35:05 +00:00
public_peer_list: ArcSwap<Vec<PeerInfo>>,
2021-10-12 15:59:46 +00:00
2021-10-14 09:35:05 +00:00
next_ping_id: AtomicU64,
2021-10-12 15:59:46 +00:00
ping_endpoint: Arc<Endpoint<PingMessage, Self>>,
peer_list_endpoint: Arc<Endpoint<PeerListMessage, Self>>,
2020-12-02 12:30:47 +00:00
}
impl FullMeshPeeringStrategy {
2021-10-14 09:35:05 +00:00
/// 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<NetApp>,
bootstrap_list: Vec<(NodeID, SocketAddr)>,
our_addr: Option<SocketAddr>,
) -> Arc<Self> {
2020-12-02 12:30:47 +00:00
let mut known_hosts = KnownHosts::new();
for (id, addr) in bootstrap_list {
if id != netapp.id {
2020-12-02 12:30:47 +00:00
known_hosts.list.insert(
id,
2021-10-14 09:35:05 +00:00
PeerInfoInternal {
2021-10-12 11:18:24 +00:00
addr,
all_addrs: vec![addr],
2020-12-02 12:30:47 +00:00
state: PeerConnState::Waiting(0, Instant::now()),
last_seen: None,
ping: VecDeque::new(),
failed_pings: 0,
2020-12-02 12:30:47 +00:00
},
);
}
}
if let Some(addr) = our_addr {
known_hosts.list.insert(
netapp.id,
PeerInfoInternal {
addr,
all_addrs: vec![addr],
state: PeerConnState::Ourself,
last_seen: None,
ping: VecDeque::new(),
failed_pings: 0,
},
);
}
2020-12-02 12:30:47 +00:00
let strat = Arc::new(Self {
netapp: netapp.clone(),
known_hosts: RwLock::new(known_hosts),
2021-10-14 09:35:05 +00:00
public_peer_list: ArcSwap::new(Arc::new(Vec::new())),
2020-12-02 12:30:47 +00:00
next_ping_id: AtomicU64::new(42),
2021-10-12 15:59:46 +00:00
ping_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/Ping".into()),
peer_list_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/PeerList".into()),
2020-12-02 12:30:47 +00:00
});
strat.update_public_peer_list(&strat.known_hosts.read().unwrap());
2021-10-12 15:59:46 +00:00
strat.ping_endpoint.set_handler(strat.clone());
strat.peer_list_endpoint.set_handler(strat.clone());
2020-12-02 12:30:47 +00:00
let strat2 = strat.clone();
netapp.on_connected(move |id: NodeID, addr: SocketAddr, is_incoming: bool| {
let strat2 = strat2.clone();
2021-10-21 10:24:42 +00:00
strat2.on_connected(id, addr, is_incoming);
});
2020-12-02 12:30:47 +00:00
let strat2 = strat.clone();
netapp.on_disconnected(move |id: NodeID, is_incoming: bool| {
let strat2 = strat2.clone();
2021-10-21 10:24:42 +00:00
strat2.on_disconnected(id, is_incoming);
});
2020-12-02 12:30:47 +00:00
strat
}
2021-10-14 09:35:05 +00:00
/// Run the full mesh peering strategy.
/// This future exits when the `must_exit` watch becomes true.
pub async fn run(self: Arc<Self>, must_exit: watch::Receiver<bool>) {
while !*must_exit.borrow() {
2020-12-02 12:30:47 +00:00
// 1. Read current state: get list of connected peers (ping them)
let (to_ping, to_retry) = {
let known_hosts = self.known_hosts.read().unwrap();
2021-10-13 15:30:41 +00:00
trace!("known_hosts: {} peers", known_hosts.list.len());
let mut to_ping = vec![];
let mut to_retry = vec![];
for (id, info) in known_hosts.list.iter() {
2022-02-21 15:57:07 +00:00
trace!("{}, {:?}", hex::encode(&id[..8]), info);
match info.state {
PeerConnState::Connected => {
let must_ping = match info.last_seen {
None => true,
Some(t) => Instant::now() - t > PING_INTERVAL,
};
if must_ping {
to_ping.push(*id);
}
2020-12-02 12:30:47 +00:00
}
PeerConnState::Waiting(_, t) => {
if Instant::now() >= t {
to_retry.push(*id);
}
2020-12-02 12:30:47 +00:00
}
_ => (),
2020-12-02 12:30:47 +00:00
}
}
(to_ping, to_retry)
};
2020-12-02 12:30:47 +00:00
// 2. Dispatch ping to hosts
trace!("to_ping: {} peers", to_retry.len());
for id in to_ping {
tokio::spawn(self.clone().ping(id));
}
// 3. Try reconnects
trace!("to_retry: {} peers", to_retry.len());
if !to_retry.is_empty() {
let mut known_hosts = self.known_hosts.write().unwrap();
for id in to_retry {
if let Some(h) = known_hosts.list.get_mut(&id) {
if let PeerConnState::Waiting(i, _) = h.state {
info!(
"Retrying connection to {} at {} ({})",
2022-02-21 15:57:07 +00:00
hex::encode(&id[..8]),
h.all_addrs
.iter()
.map(|x| format!("{}", x))
.collect::<Vec<_>>()
.join(", "),
2020-12-02 12:30:47 +00:00
i + 1
);
h.state = PeerConnState::Trying(i);
let alternate_addrs = h
.all_addrs
.iter()
.filter(|x| **x != h.addr)
.cloned()
.collect::<Vec<_>>();
tokio::spawn(self.clone().try_connect(id, h.addr, alternate_addrs));
2020-12-02 12:30:47 +00:00
}
}
}
2021-10-14 09:35:05 +00:00
self.update_public_peer_list(&known_hosts);
2020-12-02 12:30:47 +00:00
}
// 4. Sleep before next loop iteration
2021-10-12 11:07:34 +00:00
tokio::time::sleep(LOOP_DELAY).await;
2020-12-02 12:30:47 +00:00
}
}
2021-10-14 09:35:05 +00:00
/// Returns a list of currently known peers in the network.
pub fn get_peer_list(&self) -> Arc<Vec<PeerInfo>> {
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::<Vec<_>>();
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<Self>, id: NodeID) {
2020-12-02 12:30:47 +00:00
let peer_list_hash = self.known_hosts.read().unwrap().hash;
let ping_id = self.next_ping_id.fetch_add(1u64, atomic::Ordering::Relaxed);
let ping_time = Instant::now();
let ping_msg = PingMessage {
id: ping_id,
peer_list_hash,
};
debug!(
"Sending ping {} to {} at {:?}",
ping_id,
2022-02-21 15:57:07 +00:00
hex::encode(&id[..8]),
2020-12-02 12:30:47 +00:00
ping_time
);
let ping_response = select! {
r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH) => r,
_ = tokio::time::sleep(PING_TIMEOUT) => Err(Error::Message("Ping timeout".into())),
};
match ping_response {
Err(e) => {
warn!("Error pinging {}: {}", hex::encode(&id[..8]), e);
let mut known_hosts = self.known_hosts.write().unwrap();
if let Some(host) = known_hosts.list.get_mut(&id) {
host.failed_pings += 1;
if host.failed_pings > FAILED_PING_THRESHOLD {
warn!(
"Too many failed pings from {}, closing connection.",
hex::encode(&id[..8])
);
// this will later update info in known_hosts
// through the disconnection handler
self.netapp.disconnect(&id);
}
}
}
2020-12-02 12:30:47 +00:00
Ok(ping_resp) => {
let resp_time = Instant::now();
debug!(
"Got ping response from {} at {:?}",
2022-02-21 15:57:07 +00:00
hex::encode(&id[..8]),
2020-12-02 12:30:47 +00:00
resp_time
);
{
let mut known_hosts = self.known_hosts.write().unwrap();
if let Some(host) = known_hosts.list.get_mut(&id) {
host.failed_pings = 0;
2020-12-02 12:30:47 +00:00
host.last_seen = Some(resp_time);
host.ping.push_back(resp_time - ping_time);
while host.ping.len() > 10 {
host.ping.pop_front();
}
2021-10-14 09:35:05 +00:00
self.update_public_peer_list(&known_hosts);
2020-12-02 12:30:47 +00:00
}
}
if ping_resp.peer_list_hash != peer_list_hash {
2020-12-02 17:10:07 +00:00
self.exchange_peers(&id).await;
2020-12-02 12:30:47 +00:00
}
}
}
}
async fn exchange_peers(self: Arc<Self>, id: &NodeID) {
2020-12-02 12:30:47 +00:00
let peer_list = KnownHosts::map_into_vec(&self.known_hosts.read().unwrap().list);
let pex_message = PeerListMessage { list: peer_list };
2021-10-12 15:59:46 +00:00
match self
.peer_list_endpoint
.call(id, pex_message, PRIO_BACKGROUND)
2021-10-12 15:59:46 +00:00
.await
{
2020-12-02 12:30:47 +00:00
Err(e) => warn!("Error doing peer exchange: {}", e),
Ok(resp) => {
self.handle_peer_list(&resp.list[..]);
}
}
}
fn handle_peer_list(&self, list: &[(NodeID, SocketAddr)]) {
2020-12-02 12:30:47 +00:00
let mut known_hosts = self.known_hosts.write().unwrap();
2021-10-21 10:14:19 +00:00
let mut changed = false;
2020-12-02 12:30:47 +00:00
for (id, addr) in list.iter() {
if let Some(kh) = known_hosts.list.get_mut(id) {
if !kh.all_addrs.contains(addr) {
kh.all_addrs.push(*addr);
changed = true;
}
} else {
2020-12-02 12:30:47 +00:00
known_hosts.list.insert(*id, self.new_peer(id, *addr));
2021-10-21 10:14:19 +00:00
changed = true;
2020-12-02 12:30:47 +00:00
}
}
2021-10-21 10:14:19 +00:00
if changed {
known_hosts.update_hash();
self.update_public_peer_list(&known_hosts);
}
2020-12-02 12:30:47 +00:00
}
async fn try_connect(
self: Arc<Self>,
id: NodeID,
default_addr: SocketAddr,
alternate_addrs: Vec<SocketAddr>,
) {
let conn_addr = {
let mut ret = None;
for addr in [default_addr].iter().chain(alternate_addrs.iter()) {
debug!("Trying address {} for peer {}", addr, hex::encode(&id[..8]));
match self.netapp.clone().try_connect(*addr, id).await {
Ok(()) => {
ret = Some(*addr);
break;
}
Err(e) => {
debug!(
"Error connecting to {} at {}: {}",
hex::encode(&id[..8]),
addr,
e
);
}
}
}
ret
};
if let Some(ok_addr) = conn_addr {
self.on_connected(id, ok_addr, false);
} else {
warn!(
"Could not connect to peer {} ({} addresses tried)",
hex::encode(&id[..8]),
1 + alternate_addrs.len()
);
2020-12-02 12:30:47 +00:00
let mut known_hosts = self.known_hosts.write().unwrap();
if let Some(host) = known_hosts.list.get_mut(&id) {
host.state = match host.state {
PeerConnState::Trying(i) => {
if i >= CONN_MAX_RETRIES {
PeerConnState::Abandonned
} else {
PeerConnState::Waiting(i + 1, Instant::now() + CONN_RETRY_INTERVAL)
}
}
_ => PeerConnState::Waiting(0, Instant::now() + CONN_RETRY_INTERVAL),
};
2021-10-14 09:35:05 +00:00
self.update_public_peer_list(&known_hosts);
2020-12-02 12:30:47 +00:00
}
}
}
2021-10-21 10:24:42 +00:00
fn on_connected(self: Arc<Self>, id: NodeID, addr: SocketAddr, is_incoming: bool) {
let mut known_hosts = self.known_hosts.write().unwrap();
2020-12-02 12:30:47 +00:00
if is_incoming {
if let Some(host) = known_hosts.list.get_mut(&id) {
if !host.all_addrs.contains(&addr) {
host.all_addrs.push(addr);
}
} else {
known_hosts.list.insert(id, self.new_peer(&id, addr));
2020-12-02 12:30:47 +00:00
}
} else {
2022-02-21 15:57:07 +00:00
info!(
"Successfully connected to {} at {}",
hex::encode(&id[..8]),
addr
);
if let Some(host) = known_hosts.list.get_mut(&id) {
2020-12-02 12:30:47 +00:00
host.state = PeerConnState::Connected;
2021-10-21 10:14:19 +00:00
host.addr = addr;
if !host.all_addrs.contains(&addr) {
host.all_addrs.push(addr);
}
2021-10-21 10:33:35 +00:00
} else {
2021-10-22 13:20:07 +00:00
known_hosts.list.insert(
id,
PeerInfoInternal {
state: PeerConnState::Connected,
addr,
all_addrs: vec![addr],
2021-10-22 13:20:07 +00:00
last_seen: None,
ping: VecDeque::new(),
failed_pings: 0,
2021-10-22 13:20:07 +00:00
},
);
2020-12-02 12:30:47 +00:00
}
}
known_hosts.update_hash();
self.update_public_peer_list(&known_hosts);
2020-12-02 12:30:47 +00:00
}
2021-10-21 10:24:42 +00:00
fn on_disconnected(self: Arc<Self>, id: NodeID, is_incoming: bool) {
2020-12-02 12:30:47 +00:00
if !is_incoming {
2022-02-21 15:57:07 +00:00
info!("Connection to {} was closed", hex::encode(&id[..8]));
2020-12-02 12:30:47 +00:00
let mut known_hosts = self.known_hosts.write().unwrap();
if let Some(host) = known_hosts.list.get_mut(&id) {
2020-12-02 12:30:47 +00:00
host.state = PeerConnState::Waiting(0, Instant::now());
known_hosts.update_hash();
2021-10-14 09:35:05 +00:00
self.update_public_peer_list(&known_hosts);
2020-12-02 12:30:47 +00:00
}
}
}
2021-10-14 09:35:05 +00:00
fn new_peer(&self, id: &NodeID, addr: SocketAddr) -> PeerInfoInternal {
let state = if *id == self.netapp.id {
2020-12-02 12:30:47 +00:00
PeerConnState::Ourself
} else {
PeerConnState::Waiting(0, Instant::now())
};
2021-10-14 09:35:05 +00:00
PeerInfoInternal {
2020-12-02 12:30:47 +00:00
addr,
all_addrs: vec![addr],
2020-12-02 12:30:47 +00:00
state,
last_seen: None,
ping: VecDeque::new(),
failed_pings: 0,
2020-12-02 12:30:47 +00:00
}
}
}
2021-10-12 15:59:46 +00:00
#[async_trait]
impl EndpointHandler<PingMessage> for FullMeshPeeringStrategy {
async fn handle(self: &Arc<Self>, ping: PingMessage, from: NodeID) -> PingMessage {
2021-10-12 15:59:46 +00:00
let ping_resp = PingMessage {
id: ping.id,
peer_list_hash: self.known_hosts.read().unwrap().hash,
};
2022-02-21 15:57:07 +00:00
debug!("Ping from {}", hex::encode(&from[..8]));
2021-10-12 15:59:46 +00:00
ping_resp
}
}
#[async_trait]
impl EndpointHandler<PeerListMessage> for FullMeshPeeringStrategy {
async fn handle(
self: &Arc<Self>,
peer_list: PeerListMessage,
2021-10-12 15:59:46 +00:00
_from: NodeID,
) -> PeerListMessage {
self.handle_peer_list(&peer_list.list[..]);
let peer_list = KnownHosts::map_into_vec(&self.known_hosts.read().unwrap().list);
PeerListMessage { list: peer_list }
}
}