veilid/veilid-core/src/routing_table/bucket_entry.rs

699 lines
26 KiB
Rust
Raw Normal View History

2021-11-22 16:28:30 +00:00
use super::*;
2022-06-25 14:57:33 +00:00
use core::sync::atomic::{AtomicU32, Ordering};
2021-11-22 16:28:30 +00:00
2022-08-27 02:52:08 +00:00
/// Reliable pings are done with increased spacing between pings
/// - Start secs is the number of seconds between the first two pings
2021-11-22 16:28:30 +00:00
const RELIABLE_PING_INTERVAL_START_SECS: u32 = 10;
2022-08-27 02:52:08 +00:00
/// - Max secs is the maximum number of seconds between consecutive pings
2021-11-22 16:28:30 +00:00
const RELIABLE_PING_INTERVAL_MAX_SECS: u32 = 10 * 60;
2022-08-27 02:52:08 +00:00
/// - Multiplier changes the number of seconds between pings over time
/// making it longer as the node becomes more reliable
2021-11-22 16:28:30 +00:00
const RELIABLE_PING_INTERVAL_MULTIPLIER: f64 = 2.0;
2022-08-27 02:52:08 +00:00
/// Unreliable pings are done for a fixed amount of time while the
/// node is given a chance to come back online before it is made dead
/// If a node misses a single ping, it is marked unreliable and must
/// return reliable pings for the duration of the span before being
/// marked reliable again
///
/// - Span is the number of seconds total to attempt to validate the node
2021-11-22 16:28:30 +00:00
const UNRELIABLE_PING_SPAN_SECS: u32 = 60;
2022-08-27 02:52:08 +00:00
/// - Interval is the number of seconds between each ping
2021-11-22 16:28:30 +00:00
const UNRELIABLE_PING_INTERVAL_SECS: u32 = 5;
2022-08-27 02:52:08 +00:00
/// Keepalive pings are done occasionally to ensure holepunched public dialinfo
/// remains valid, as well as to make sure we remain in any relay node's routing table
2022-07-05 23:47:25 +00:00
const KEEPALIVE_PING_INTERVAL_SECS: u32 = 10;
2022-04-07 13:55:09 +00:00
2022-08-27 02:52:08 +00:00
/// How many times do we try to ping a never-reached node before we call it dead
const NEVER_REACHED_PING_COUNT: u32 = 3;
2022-04-07 13:55:09 +00:00
// Do not change order here, it will mess up other sorts
2022-08-27 02:52:08 +00:00
2021-12-11 01:14:33 +00:00
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
2021-11-22 16:28:30 +00:00
pub enum BucketEntryState {
Dead,
2021-12-11 01:14:33 +00:00
Unreliable,
Reliable,
2021-11-22 16:28:30 +00:00
}
2022-08-09 00:42:27 +00:00
#[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord, Hash)]
2022-09-04 18:17:28 +00:00
struct LastConnectionKey(ProtocolType, AddressType);
2022-09-01 01:41:48 +00:00
/// Bucket entry information specific to the LocalNetwork RoutingDomain
#[derive(Debug)]
pub struct BucketEntryPublicInternet {
/// The PublicInternet node info
signed_node_info: Option<Box<SignedNodeInfo>>,
/// If this node has seen our publicinternet node info
seen_our_node_info: bool,
/// Last known node status
node_status: Option<PublicInternetNodeStatus>,
}
/// Bucket entry information specific to the LocalNetwork RoutingDomain
#[derive(Debug)]
pub struct BucketEntryLocalNetwork {
/// The LocalNetwork node info
signed_node_info: Option<Box<SignedNodeInfo>>,
/// If this node has seen our localnetwork node info
seen_our_node_info: bool,
/// Last known node status
node_status: Option<LocalNetworkNodeStatus>,
}
2022-08-09 00:42:27 +00:00
2022-06-25 14:57:33 +00:00
#[derive(Debug)]
pub struct BucketEntryInner {
2021-11-22 16:28:30 +00:00
min_max_version: Option<(u8, u8)>,
2022-08-27 02:52:08 +00:00
updated_since_last_network_change: bool,
2022-08-09 00:42:27 +00:00
last_connections: BTreeMap<LastConnectionKey, (ConnectionDescriptor, u64)>,
2022-09-01 01:41:48 +00:00
public_internet: BucketEntryPublicInternet,
local_network: BucketEntryLocalNetwork,
2022-04-08 14:17:09 +00:00
peer_stats: PeerStats,
2022-03-19 22:19:40 +00:00
latency_stats_accounting: LatencyStatsAccounting,
transfer_stats_accounting: TransferStatsAccounting,
2022-05-25 15:12:19 +00:00
#[cfg(feature = "tracking")]
next_track_id: usize,
#[cfg(feature = "tracking")]
node_ref_tracks: HashMap<usize, backtrace::Backtrace>,
2021-11-22 16:28:30 +00:00
}
2022-06-25 14:57:33 +00:00
impl BucketEntryInner {
2022-05-25 15:12:19 +00:00
#[cfg(feature = "tracking")]
pub fn track(&mut self) -> usize {
let track_id = self.next_track_id;
self.next_track_id += 1;
self.node_ref_tracks
.insert(track_id, backtrace::Backtrace::new_unresolved());
track_id
}
#[cfg(feature = "tracking")]
pub fn untrack(&mut self, track_id: usize) {
self.node_ref_tracks.remove(&track_id);
}
2022-05-26 00:56:13 +00:00
// Less is faster
pub fn cmp_fastest(e1: &Self, e2: &Self) -> std::cmp::Ordering {
2022-04-07 13:55:09 +00:00
// Lower latency to the front
if let Some(e1_latency) = &e1.peer_stats.latency {
if let Some(e2_latency) = &e2.peer_stats.latency {
e1_latency.average.cmp(&e2_latency.average)
} else {
std::cmp::Ordering::Less
}
} else if e2.peer_stats.latency.is_some() {
std::cmp::Ordering::Greater
} else {
std::cmp::Ordering::Equal
}
}
2022-05-26 00:56:13 +00:00
// Less is more reliable then faster
2022-04-07 13:55:09 +00:00
pub fn cmp_fastest_reliable(cur_ts: u64, e1: &Self, e2: &Self) -> std::cmp::Ordering {
// Reverse compare so most reliable is at front
let ret = e2.state(cur_ts).cmp(&e1.state(cur_ts));
if ret != std::cmp::Ordering::Equal {
return ret;
}
// Lower latency to the front
if let Some(e1_latency) = &e1.peer_stats.latency {
if let Some(e2_latency) = &e2.peer_stats.latency {
e1_latency.average.cmp(&e2_latency.average)
} else {
std::cmp::Ordering::Less
}
} else if e2.peer_stats.latency.is_some() {
std::cmp::Ordering::Greater
} else {
std::cmp::Ordering::Equal
}
}
2022-10-09 18:59:01 +00:00
// Less is more reliable then older
pub fn cmp_oldest_reliable(cur_ts: u64, e1: &Self, e2: &Self) -> std::cmp::Ordering {
// Reverse compare so most reliable is at front
let ret = e2.state(cur_ts).cmp(&e1.state(cur_ts));
if ret != std::cmp::Ordering::Equal {
return ret;
}
// Lower timestamp to the front, recent or no timestamp is at the end
if let Some(e1_ts) = &e1.peer_stats.rpc_stats.first_consecutive_seen_ts {
if let Some(e2_ts) = &e2.peer_stats.rpc_stats.first_consecutive_seen_ts {
e1_ts.cmp(&e2_ts)
} else {
std::cmp::Ordering::Less
}
} else if e2.peer_stats.rpc_stats.first_consecutive_seen_ts.is_some() {
std::cmp::Ordering::Greater
} else {
std::cmp::Ordering::Equal
}
}
2022-04-07 13:55:09 +00:00
pub fn sort_fastest_reliable_fn(cur_ts: u64) -> impl FnMut(&Self, &Self) -> std::cmp::Ordering {
move |e1, e2| Self::cmp_fastest_reliable(cur_ts, e1, e2)
}
2022-09-03 17:57:25 +00:00
pub fn clear_signed_node_info(&mut self, routing_domain: RoutingDomain) {
// Get the correct signed_node_info for the chosen routing domain
let opt_current_sni = match routing_domain {
RoutingDomain::LocalNetwork => &mut self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &mut self.public_internet.signed_node_info,
};
*opt_current_sni = None;
}
2022-08-05 22:07:32 +00:00
// Retuns true if the node info changed
2022-08-27 02:52:08 +00:00
pub fn update_signed_node_info(
&mut self,
2022-09-04 18:17:28 +00:00
routing_domain: RoutingDomain,
2022-08-27 02:52:08 +00:00
signed_node_info: SignedNodeInfo,
2022-08-29 02:34:24 +00:00
) {
2022-09-01 01:41:48 +00:00
// Get the correct signed_node_info for the chosen routing domain
2022-09-04 18:17:28 +00:00
let opt_current_sni = match routing_domain {
2022-09-01 01:41:48 +00:00
RoutingDomain::LocalNetwork => &mut self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &mut self.public_internet.signed_node_info,
};
2022-08-27 02:52:08 +00:00
// See if we have an existing signed_node_info to update or not
2022-09-01 01:41:48 +00:00
if let Some(current_sni) = opt_current_sni {
2022-08-27 02:52:08 +00:00
// If the timestamp hasn't changed or is less, ignore this update
if signed_node_info.timestamp <= current_sni.timestamp {
// If we received a node update with the same timestamp
2022-09-03 17:57:25 +00:00
// we can make this node live again, but only if our network has recently changed
// which may make nodes that were unreachable now reachable with the same dialinfo
2022-08-27 02:52:08 +00:00
if !self.updated_since_last_network_change
&& signed_node_info.timestamp == current_sni.timestamp
{
// No need to update the signednodeinfo though since the timestamp is the same
2022-08-31 01:21:16 +00:00
// Touch the node and let it try to live again
2022-08-27 02:52:08 +00:00
self.updated_since_last_network_change = true;
2022-08-29 02:34:24 +00:00
self.touch_last_seen(intf::get_timestamp());
2022-08-27 02:52:08 +00:00
}
2022-08-29 02:34:24 +00:00
return;
}
}
2022-08-27 02:52:08 +00:00
// Update the protocol min/max version we have
2022-05-28 14:07:57 +00:00
self.min_max_version = Some((
signed_node_info.node_info.min_version,
signed_node_info.node_info.max_version,
));
2022-08-27 02:52:08 +00:00
// Update the signed node info
2022-09-01 01:41:48 +00:00
*opt_current_sni = Some(Box::new(signed_node_info));
2022-08-27 02:52:08 +00:00
self.updated_since_last_network_change = true;
2022-08-29 02:34:24 +00:00
self.touch_last_seen(intf::get_timestamp());
2021-11-22 16:28:30 +00:00
}
2022-05-11 01:49:42 +00:00
2022-09-03 17:57:25 +00:00
pub fn has_node_info(&self, routing_domain_set: RoutingDomainSet) -> bool {
for routing_domain in routing_domain_set {
2022-09-01 01:41:48 +00:00
// Get the correct signed_node_info for the chosen routing domain
let opt_current_sni = match routing_domain {
2022-09-04 19:40:35 +00:00
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
2022-09-01 01:41:48 +00:00
};
2022-09-03 17:57:25 +00:00
if opt_current_sni.is_some() {
return true;
2022-08-31 01:21:16 +00:00
}
2022-05-11 01:49:42 +00:00
}
2022-09-03 17:57:25 +00:00
false
2022-05-11 01:49:42 +00:00
}
2022-09-04 18:17:28 +00:00
pub fn node_info(&self, routing_domain: RoutingDomain) -> Option<&NodeInfo> {
let opt_current_sni = match routing_domain {
2022-09-04 19:40:35 +00:00
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
2022-09-04 18:17:28 +00:00
};
opt_current_sni.as_ref().map(|s| &s.node_info)
2021-11-22 16:28:30 +00:00
}
2022-05-11 01:49:42 +00:00
2022-09-04 18:17:28 +00:00
pub fn signed_node_info(&self, routing_domain: RoutingDomain) -> Option<&SignedNodeInfo> {
2022-09-01 01:41:48 +00:00
let opt_current_sni = match routing_domain {
2022-09-04 19:40:35 +00:00
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
2022-09-01 01:41:48 +00:00
};
2022-09-04 18:17:28 +00:00
opt_current_sni.as_ref().map(|s| s.as_ref())
2022-05-11 01:49:42 +00:00
}
2022-08-31 01:21:16 +00:00
2022-09-04 18:17:28 +00:00
pub fn make_peer_info(&self, key: DHTKey, routing_domain: RoutingDomain) -> Option<PeerInfo> {
2022-09-01 01:41:48 +00:00
let opt_current_sni = match routing_domain {
2022-09-04 19:40:35 +00:00
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
2022-09-01 01:41:48 +00:00
};
opt_current_sni.as_ref().map(|s| PeerInfo {
node_id: NodeId::new(key),
signed_node_info: *s.clone(),
})
2021-11-22 16:28:30 +00:00
}
2022-09-03 17:57:25 +00:00
pub fn best_routing_domain(
&self,
routing_domain_set: RoutingDomainSet,
) -> Option<RoutingDomain> {
for routing_domain in routing_domain_set {
let opt_current_sni = match routing_domain {
2022-09-04 19:40:35 +00:00
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
2022-09-03 17:57:25 +00:00
};
2022-09-04 19:40:35 +00:00
if opt_current_sni.is_some() {
2022-09-04 18:17:28 +00:00
return Some(routing_domain);
2022-09-03 17:57:25 +00:00
}
}
None
}
fn descriptor_to_key(&self, last_connection: ConnectionDescriptor) -> LastConnectionKey {
2022-08-09 00:42:27 +00:00
LastConnectionKey(
last_connection.protocol_type(),
last_connection.address_type(),
)
2021-11-22 16:28:30 +00:00
}
2022-08-09 00:42:27 +00:00
// Stores a connection descriptor in this entry's table of last connections
pub fn set_last_connection(&mut self, last_connection: ConnectionDescriptor, timestamp: u64) {
2022-09-03 17:57:25 +00:00
let key = self.descriptor_to_key(last_connection);
2022-08-09 00:42:27 +00:00
self.last_connections
.insert(key, (last_connection, timestamp));
2022-08-06 00:34:00 +00:00
}
2022-08-09 00:42:27 +00:00
// Clears the table of last connections to ensure we create new ones and drop any existing ones
pub fn clear_last_connections(&mut self) {
self.last_connections.clear();
2021-11-22 16:28:30 +00:00
}
// Gets all the 'last connections' that match a particular filter
pub(super) fn last_connections(
2022-08-09 00:42:27 +00:00
&self,
2022-10-19 01:53:45 +00:00
rti: &RoutingTableInner,
filter: Option<NodeRefFilter>,
) -> Vec<(ConnectionDescriptor, u64)> {
let mut out: Vec<(ConnectionDescriptor, u64)> = self
.last_connections
.iter()
.filter_map(|(k, v)| {
let include = if let Some(filter) = &filter {
let remote_address = v.0.remote_address().address();
2022-10-19 01:53:45 +00:00
if let Some(routing_domain) = rti.routing_domain_for_address(remote_address) {
if filter.routing_domain_set.contains(routing_domain)
&& filter.dial_info_filter.protocol_type_set.contains(k.0)
&& filter.dial_info_filter.address_type_set.contains(k.1)
{
// matches filter
true
} else {
// does not match filter
false
2022-09-04 18:17:28 +00:00
}
} else {
// no valid routing domain
false
2022-09-03 17:57:25 +00:00
}
} else {
// no filter
true
};
if include {
Some(v.clone())
} else {
None
2022-08-09 00:42:27 +00:00
}
})
.collect();
// Sort with newest timestamps first
out.sort_by(|a, b| b.1.cmp(&a.1));
out
2022-08-09 00:42:27 +00:00
}
2021-11-22 16:28:30 +00:00
pub fn set_min_max_version(&mut self, min_max_version: (u8, u8)) {
self.min_max_version = Some(min_max_version);
}
pub fn min_max_version(&self) -> Option<(u8, u8)> {
self.min_max_version
}
pub fn state(&self, cur_ts: u64) -> BucketEntryState {
if self.check_reliable(cur_ts) {
BucketEntryState::Reliable
} else if self.check_dead(cur_ts) {
BucketEntryState::Dead
} else {
BucketEntryState::Unreliable
}
}
pub fn peer_stats(&self) -> &PeerStats {
&self.peer_stats
}
2022-04-08 14:17:09 +00:00
pub fn update_node_status(&mut self, status: NodeStatus) {
2022-09-01 01:41:48 +00:00
match status {
NodeStatus::LocalNetwork(ln) => {
self.local_network.node_status = Some(ln);
}
NodeStatus::PublicInternet(pi) => {
self.public_internet.node_status = Some(pi);
}
}
}
pub fn node_status(&self, routing_domain: RoutingDomain) -> Option<NodeStatus> {
match routing_domain {
RoutingDomain::LocalNetwork => self
.local_network
.node_status
2022-09-04 19:40:35 +00:00
.as_ref()
.map(|ln| NodeStatus::LocalNetwork(ln.clone())),
2022-09-01 01:41:48 +00:00
RoutingDomain::PublicInternet => self
2022-09-03 17:57:25 +00:00
.public_internet
2022-09-01 01:41:48 +00:00
.node_status
2022-09-04 19:40:35 +00:00
.as_ref()
.map(|pi| NodeStatus::PublicInternet(pi.clone())),
2022-09-01 01:41:48 +00:00
}
2021-11-22 16:28:30 +00:00
}
2022-08-31 01:21:16 +00:00
pub fn set_seen_our_node_info(&mut self, routing_domain: RoutingDomain, seen: bool) {
2022-09-01 01:41:48 +00:00
match routing_domain {
RoutingDomain::LocalNetwork => {
self.local_network.seen_our_node_info = seen;
}
RoutingDomain::PublicInternet => {
self.public_internet.seen_our_node_info = seen;
}
}
2022-03-25 02:07:55 +00:00
}
2022-08-31 01:21:16 +00:00
pub fn has_seen_our_node_info(&self, routing_domain: RoutingDomain) -> bool {
2022-09-01 01:41:48 +00:00
match routing_domain {
RoutingDomain::LocalNetwork => self.local_network.seen_our_node_info,
RoutingDomain::PublicInternet => self.public_internet.seen_our_node_info,
}
2022-03-25 02:07:55 +00:00
}
2022-08-27 02:52:08 +00:00
pub fn set_updated_since_last_network_change(&mut self, updated: bool) {
self.updated_since_last_network_change = updated;
}
pub fn has_updated_since_last_network_change(&self) -> bool {
self.updated_since_last_network_change
}
2021-11-22 16:28:30 +00:00
///// stats methods
// called every ROLLING_TRANSFERS_INTERVAL_SECS seconds
pub(super) fn roll_transfers(&mut self, last_ts: u64, cur_ts: u64) {
2022-03-19 22:19:40 +00:00
self.transfer_stats_accounting.roll_transfers(
last_ts,
cur_ts,
&mut self.peer_stats.transfer,
);
2021-11-22 16:28:30 +00:00
}
// Called for every round trip packet we receive
fn record_latency(&mut self, latency: u64) {
2022-03-19 22:19:40 +00:00
self.peer_stats.latency = Some(self.latency_stats_accounting.record_latency(latency));
2021-11-22 16:28:30 +00:00
}
///// state machine handling
pub(super) fn check_reliable(&self, cur_ts: u64) -> bool {
// If we have had any failures to send, this is not reliable
if self.peer_stats.rpc_stats.failed_to_send > 0 {
return false;
}
// if we have seen the node consistently for longer that UNRELIABLE_PING_SPAN_SECS
match self.peer_stats.rpc_stats.first_consecutive_seen_ts {
2021-11-22 16:28:30 +00:00
None => false,
2021-12-11 01:14:33 +00:00
Some(ts) => {
cur_ts.saturating_sub(ts) >= (UNRELIABLE_PING_SPAN_SECS as u64 * 1000000u64)
}
2021-11-22 16:28:30 +00:00
}
}
pub(super) fn check_dead(&self, cur_ts: u64) -> bool {
// If we have failured to send NEVER_REACHED_PING_COUNT times in a row, the node is dead
if self.peer_stats.rpc_stats.failed_to_send >= NEVER_REACHED_PING_COUNT {
return true;
}
2021-11-22 16:28:30 +00:00
// if we have not heard from the node at all for the duration of the unreliable ping span
// a node is not dead if we haven't heard from it yet,
// but we give it NEVER_REACHED_PING_COUNT chances to ping before we say it's dead
match self.peer_stats.rpc_stats.last_seen_ts {
None => self.peer_stats.rpc_stats.recent_lost_answers < NEVER_REACHED_PING_COUNT,
2021-12-11 01:14:33 +00:00
Some(ts) => {
cur_ts.saturating_sub(ts) >= (UNRELIABLE_PING_SPAN_SECS as u64 * 1000000u64)
}
2021-11-22 16:28:30 +00:00
}
}
2022-10-06 16:39:30 +00:00
/// Return the last time we either saw a node, or asked it a question
fn latest_contact_time(&self) -> Option<u64> {
self.peer_stats
.rpc_stats
.last_seen_ts
2022-10-06 16:39:30 +00:00
.max(self.peer_stats.rpc_stats.last_question)
}
fn needs_constant_ping(&self, cur_ts: u64, interval: u64) -> bool {
// If we have not either seen the node in the last 'interval' then we should ping it
let latest_contact_time = self.latest_contact_time();
match latest_contact_time {
2022-04-07 13:55:09 +00:00
None => true,
Some(latest_contact_time) => {
// If we haven't done anything with this node in 'interval' seconds
cur_ts.saturating_sub(latest_contact_time) >= (interval * 1000000u64)
}
2022-04-07 13:55:09 +00:00
}
}
2022-03-25 02:07:55 +00:00
// Check if this node needs a ping right now to validate it is still reachable
2022-09-04 19:40:35 +00:00
pub(super) fn needs_ping(&self, cur_ts: u64, needs_keepalive: bool) -> bool {
2022-03-25 02:07:55 +00:00
// See which ping pattern we are to use
2022-04-07 13:55:09 +00:00
let state = self.state(cur_ts);
2022-03-25 02:07:55 +00:00
2022-08-31 01:21:16 +00:00
// If this entry needs a keepalive (like a relay node),
// then we should ping it regularly to keep our association alive
if needs_keepalive {
return self.needs_constant_ping(cur_ts, KEEPALIVE_PING_INTERVAL_SECS as u64);
2022-03-25 02:07:55 +00:00
}
2022-10-06 16:39:30 +00:00
// If we don't have node status for this node, then we should ping it to get some node status
for routing_domain in RoutingDomainSet::all() {
if self.has_node_info(routing_domain.into()) {
if self.node_status(routing_domain).is_none() {
return true;
}
}
}
2022-03-27 01:25:24 +00:00
match state {
2021-11-22 16:28:30 +00:00
BucketEntryState::Reliable => {
// If we are in a reliable state, we need a ping on an exponential scale
2022-10-06 16:39:30 +00:00
let latest_contact_time = self.latest_contact_time();
match latest_contact_time {
None => {
error!("Peer is reliable, but not seen!");
true
}
Some(latest_contact_time) => {
let first_consecutive_seen_ts =
self.peer_stats.rpc_stats.first_consecutive_seen_ts.unwrap();
let start_of_reliable_time = first_consecutive_seen_ts
2021-12-11 01:14:33 +00:00
+ ((UNRELIABLE_PING_SPAN_SECS - UNRELIABLE_PING_INTERVAL_SECS) as u64
2022-04-18 22:49:33 +00:00
* 1_000_000u64);
2021-12-11 01:14:33 +00:00
let reliable_cur = cur_ts.saturating_sub(start_of_reliable_time);
let reliable_last =
latest_contact_time.saturating_sub(start_of_reliable_time);
2021-11-22 16:28:30 +00:00
retry_falloff_log(
reliable_last,
reliable_cur,
2022-04-18 22:49:33 +00:00
RELIABLE_PING_INTERVAL_START_SECS as u64 * 1_000_000u64,
RELIABLE_PING_INTERVAL_MAX_SECS as u64 * 1_000_000u64,
2021-11-22 16:28:30 +00:00
RELIABLE_PING_INTERVAL_MULTIPLIER,
)
}
}
}
BucketEntryState::Unreliable => {
// If we are in an unreliable state, we need a ping every UNRELIABLE_PING_INTERVAL_SECS seconds
2022-04-07 13:55:09 +00:00
self.needs_constant_ping(cur_ts, UNRELIABLE_PING_INTERVAL_SECS as u64)
2021-11-22 16:28:30 +00:00
}
BucketEntryState::Dead => false,
}
}
pub(super) fn touch_last_seen(&mut self, ts: u64) {
// Mark the node as seen
if self
.peer_stats
.rpc_stats
.first_consecutive_seen_ts
.is_none()
{
self.peer_stats.rpc_stats.first_consecutive_seen_ts = Some(ts);
}
self.peer_stats.rpc_stats.last_seen_ts = Some(ts);
2021-11-22 16:28:30 +00:00
}
2022-06-25 14:57:33 +00:00
pub(super) fn _state_debug_info(&self, cur_ts: u64) -> String {
let first_consecutive_seen_ts = if let Some(first_consecutive_seen_ts) =
self.peer_stats.rpc_stats.first_consecutive_seen_ts
2021-12-11 01:14:33 +00:00
{
format!(
"{}s ago",
timestamp_to_secs(cur_ts.saturating_sub(first_consecutive_seen_ts))
2021-12-11 01:14:33 +00:00
)
} else {
"never".to_owned()
};
let last_seen_ts_str = if let Some(last_seen_ts) = self.peer_stats.rpc_stats.last_seen_ts {
2021-12-11 01:14:33 +00:00
format!(
"{}s ago",
timestamp_to_secs(cur_ts.saturating_sub(last_seen_ts))
2021-12-11 01:14:33 +00:00
)
} else {
"never".to_owned()
};
format!(
2022-06-25 14:57:33 +00:00
"state: {:?}, first_consecutive_seen_ts: {}, last_seen_ts: {}",
2021-12-11 01:14:33 +00:00
self.state(cur_ts),
first_consecutive_seen_ts,
last_seen_ts_str
2021-12-11 01:14:33 +00:00
)
}
2021-11-22 16:28:30 +00:00
////////////////////////////////////////////////////////////////
/// Called when rpc processor things happen
2021-11-22 16:28:30 +00:00
2022-04-18 22:49:33 +00:00
pub(super) fn question_sent(&mut self, ts: u64, bytes: u64, expects_answer: bool) {
2022-03-19 22:19:40 +00:00
self.transfer_stats_accounting.add_up(bytes);
2022-04-18 22:49:33 +00:00
self.peer_stats.rpc_stats.messages_sent += 1;
self.peer_stats.rpc_stats.failed_to_send = 0;
2022-04-18 22:49:33 +00:00
if expects_answer {
self.peer_stats.rpc_stats.questions_in_flight += 1;
self.peer_stats.rpc_stats.last_question = Some(ts);
2021-12-14 14:48:33 +00:00
}
2021-11-22 16:28:30 +00:00
}
2022-04-18 22:49:33 +00:00
pub(super) fn question_rcvd(&mut self, ts: u64, bytes: u64) {
2022-03-19 22:19:40 +00:00
self.transfer_stats_accounting.add_down(bytes);
2022-04-18 22:49:33 +00:00
self.peer_stats.rpc_stats.messages_rcvd += 1;
2021-11-22 16:28:30 +00:00
self.touch_last_seen(ts);
}
pub(super) fn answer_sent(&mut self, bytes: u64) {
2022-03-19 22:19:40 +00:00
self.transfer_stats_accounting.add_up(bytes);
2022-04-18 22:49:33 +00:00
self.peer_stats.rpc_stats.messages_sent += 1;
self.peer_stats.rpc_stats.failed_to_send = 0;
2021-11-22 16:28:30 +00:00
}
2022-04-18 22:49:33 +00:00
pub(super) fn answer_rcvd(&mut self, send_ts: u64, recv_ts: u64, bytes: u64) {
2022-03-19 22:19:40 +00:00
self.transfer_stats_accounting.add_down(bytes);
2022-04-18 22:49:33 +00:00
self.peer_stats.rpc_stats.messages_rcvd += 1;
self.peer_stats.rpc_stats.questions_in_flight -= 1;
2021-11-22 16:28:30 +00:00
self.record_latency(recv_ts - send_ts);
self.touch_last_seen(recv_ts);
self.peer_stats.rpc_stats.recent_lost_answers = 0;
2021-11-22 16:28:30 +00:00
}
pub(super) fn question_lost(&mut self) {
self.peer_stats.rpc_stats.first_consecutive_seen_ts = None;
2022-04-18 22:49:33 +00:00
self.peer_stats.rpc_stats.questions_in_flight -= 1;
self.peer_stats.rpc_stats.recent_lost_answers += 1;
}
pub(super) fn failed_to_send(&mut self, ts: u64, expects_answer: bool) {
if expects_answer {
self.peer_stats.rpc_stats.last_question = Some(ts);
}
self.peer_stats.rpc_stats.failed_to_send += 1;
self.peer_stats.rpc_stats.first_consecutive_seen_ts = None;
2021-11-22 16:28:30 +00:00
}
}
2022-06-25 14:57:33 +00:00
#[derive(Debug)]
pub struct BucketEntry {
pub(super) ref_count: AtomicU32,
inner: RwLock<BucketEntryInner>,
}
impl BucketEntry {
pub(super) fn new() -> Self {
2022-06-28 03:46:29 +00:00
let now = intf::get_timestamp();
2022-06-25 14:57:33 +00:00
Self {
ref_count: AtomicU32::new(0),
inner: RwLock::new(BucketEntryInner {
min_max_version: None,
2022-08-27 02:52:08 +00:00
updated_since_last_network_change: false,
2022-08-09 00:42:27 +00:00
last_connections: BTreeMap::new(),
2022-09-01 01:41:48 +00:00
local_network: BucketEntryLocalNetwork {
seen_our_node_info: false,
signed_node_info: None,
node_status: None,
},
public_internet: BucketEntryPublicInternet {
seen_our_node_info: false,
signed_node_info: None,
node_status: None,
},
2022-06-25 14:57:33 +00:00
peer_stats: PeerStats {
time_added: now,
rpc_stats: RPCStats::default(),
latency: None,
transfer: TransferStatsDownUp::default(),
},
latency_stats_accounting: LatencyStatsAccounting::new(),
transfer_stats_accounting: TransferStatsAccounting::new(),
#[cfg(feature = "tracking")]
next_track_id: 0,
#[cfg(feature = "tracking")]
node_ref_tracks: HashMap::new(),
}),
}
}
2022-10-09 18:59:01 +00:00
// Note, that this requires -also- holding the RoutingTable read lock, as an
// immutable reference to RoutingTableInner must be passed in to get this
// This ensures that an operation on the routing table can not change entries
// while it is being read from
pub(super) fn with<F, R>(&self, rti: &RoutingTableInner, f: F) -> R
2022-06-25 14:57:33 +00:00
where
2022-10-09 18:59:01 +00:00
F: FnOnce(&RoutingTableInner, &BucketEntryInner) -> R,
2022-06-25 14:57:33 +00:00
{
let inner = self.inner.read();
2022-10-09 18:59:01 +00:00
f(rti, &*inner)
2022-06-25 14:57:33 +00:00
}
2022-10-09 18:59:01 +00:00
// Note, that this requires -also- holding the RoutingTable write lock, as a
// mutable reference to RoutingTableInner must be passed in to get this
pub(super) fn with_mut<F, R>(&self, rti: &mut RoutingTableInner, f: F) -> R
2022-06-25 14:57:33 +00:00
where
2022-10-09 18:59:01 +00:00
F: FnOnce(&mut RoutingTableInner, &mut BucketEntryInner) -> R,
2022-06-25 14:57:33 +00:00
{
let mut inner = self.inner.write();
2022-10-09 18:59:01 +00:00
f(rti, &mut *inner)
2022-06-25 14:57:33 +00:00
}
}
2021-11-22 16:28:30 +00:00
impl Drop for BucketEntry {
fn drop(&mut self) {
2022-06-25 14:57:33 +00:00
if self.ref_count.load(Ordering::Relaxed) != 0 {
2022-05-25 15:12:19 +00:00
#[cfg(feature = "tracking")]
{
println!("NodeRef Tracking");
for (id, bt) in &mut self.node_ref_tracks {
bt.resolve();
println!("Id: {}\n----------------\n{:#?}", id, bt);
}
}
panic!(
"bucket entry dropped with non-zero refcount: {:#?}",
2022-08-31 01:21:16 +00:00
&*self.inner.read()
)
}
2021-11-22 16:28:30 +00:00
}
}