#![warn(missing_docs)]
use num_bigint::BigUint;
use super::Did;
pub const RING_BITS: usize = 160;
pub const DEFAULT_SUCCESSOR_CAPACITY: usize = 3;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TopologyState {
pub local: Did,
pub successors: Vec<Did>,
pub predecessor: Option<Did>,
pub fingers: Vec<Option<Did>>,
pub fix_finger_index: usize,
}
impl TopologyState {
pub fn new(
local: Did,
successors: Vec<Did>,
predecessor: Option<Did>,
fingers: Vec<Option<Did>>,
fix_finger_index: usize,
) -> Self {
Self {
local,
successors,
predecessor,
fingers,
fix_finger_index,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FindSuccessorStep {
Local(Did),
Remote {
next: Did,
did: Did,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TopologyEvent {
Join {
peer: Did,
},
Remove {
peer: Did,
},
UpdateSuccessor {
successor: Did,
},
Notify {
predecessor: Did,
},
Stabilize {
successors: Vec<Did>,
predecessor: Option<Did>,
},
FixFinger,
ApplyFinger {
index: usize,
successor: Did,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TopologyAction {
FindSuccessorForConnect {
next: Did,
did: Did,
},
FindSuccessorForFix {
next: Did,
did: Did,
index: usize,
},
QuerySuccessorList(Did),
Notify(Did),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TopologyStep {
pub state: TopologyState,
pub actions: Vec<TopologyAction>,
}
pub fn dist(a: Did, b: Did) -> BigUint {
BigUint::from(b - a)
}
fn push_unique(xs: &mut Vec<Did>, x: Did) {
if !xs.contains(&x) {
xs.push(x);
}
}
fn sorted_successors(mut candidates: Vec<Did>, local: Did, capacity: usize) -> Vec<Did> {
candidates.retain(|&did| did != local);
candidates.sort_by_key(|&did| dist(local, did));
candidates.dedup();
candidates.truncate(capacity);
candidates
}
pub fn successors(all: &[Did], n: Did, capacity: usize) -> Vec<Did> {
sorted_successors(all.to_vec(), n, capacity)
}
pub fn predecessor(all: &[Did], n: Did) -> Option<Did> {
all.iter()
.copied()
.filter(|&did| did != n)
.max_by_key(|&did| dist(n, did))
}
pub fn finger(all: &[Did], n: Did, bit: usize) -> Option<Did> {
let threshold = BigUint::from(1u8) << bit;
all.iter()
.copied()
.filter(|&did| did != n && dist(n, did) >= threshold)
.min_by_key(|&did| dist(n, did))
}
pub fn finger_table(all: &[Did], n: Did) -> Vec<Option<Did>> {
(0..RING_BITS).map(|bit| finger(all, n, bit)).collect()
}
pub fn update_successors(local: Did, current: &[Did], candidate: Did, capacity: usize) -> Vec<Did> {
let mut candidates = current.to_vec();
push_unique(&mut candidates, candidate);
sorted_successors(candidates, local, capacity)
}
fn finger_join(local: Did, current: &[Option<Did>], peer: Did) -> Vec<Option<Did>> {
let bias = dist(local, peer);
current
.iter()
.copied()
.enumerate()
.map(|(slot, old)| {
let pos = BigUint::from(Did::power_of_two(slot));
if bias < pos || peer == local {
old
} else {
match old {
Some(existing) if dist(local, existing) < bias => old,
_ => Some(peer),
}
}
})
.collect()
}
fn finger_remove(current: &[Option<Did>], peer: Did) -> Vec<Option<Did>> {
let mut next = current.to_vec();
let indexes = next
.iter()
.enumerate()
.filter(|(_, did)| **did == Some(peer))
.map(|(index, _)| index)
.collect::<Vec<_>>();
let (Some(first), Some(last)) = (indexes.first().copied(), indexes.last().copied()) else {
return next;
};
let replacement = next.get(last.saturating_add(1)).copied().flatten();
for slot in next.iter_mut().take(last.saturating_add(1)).skip(first) {
*slot = replacement;
}
next
}
fn finger_set(
local: Did,
current: &[Option<Did>],
index: usize,
successor: Did,
) -> Vec<Option<Did>> {
let mut next = current.to_vec();
if successor != local {
if let Some(slot) = next.get_mut(index) {
*slot = Some(successor);
}
}
next
}
pub fn find_successor(state: &TopologyState, did: Did) -> FindSuccessorStep {
let head = state.successors.first().copied().unwrap_or(state.local);
if state.successors.is_empty() || dist(state.local, did) <= dist(state.local, head) {
FindSuccessorStep::Local(head)
} else {
let next = state
.fingers
.iter()
.rev()
.flatten()
.copied()
.find(|peer| dist(state.local, *peer) < dist(state.local, did))
.unwrap_or(state.local);
FindSuccessorStep::Remote { next, did }
}
}
pub fn rectify_predecessor(local: Did, current: Option<Did>, candidate: Did) -> Option<Did> {
match current {
Some(cur) if dist(local, cur) >= dist(local, candidate) => Some(cur),
_ => Some(candidate),
}
}
pub fn stabilize_successors(
local: Did,
current: &[Did],
topo_successors: &[Did],
topo_predecessor: Option<Did>,
capacity: usize,
) -> Vec<Did> {
let mut known = vec![local];
for &did in current {
push_unique(&mut known, did);
}
if let Some(pred) = topo_predecessor {
push_unique(&mut known, pred);
}
for &did in topo_successors
.iter()
.take(topo_successors.len().saturating_sub(1))
{
push_unique(&mut known, did);
}
successors(&known, local, capacity)
}
pub fn stabilize_query(local: Did, current: &[Did], topo_predecessor: Option<Did>) -> Option<Did> {
let pred = topo_predecessor?;
if pred == local {
return None;
}
let old_head = current.iter().copied().min_by_key(|&did| dist(local, did));
match old_head {
Some(head) if dist(local, pred) >= dist(local, head) => None,
_ => Some(pred),
}
}
pub fn stabilize_notify(local: Did, next_successors: &[Did]) -> Option<Did> {
next_successors.first().copied().filter(|&did| did != local)
}
fn step_join(state: &TopologyState, peer: Did, capacity: usize) -> TopologyStep {
if peer == state.local {
return TopologyStep {
state: state.clone(),
actions: Vec::new(),
};
}
TopologyStep {
state: TopologyState {
successors: update_successors(state.local, &state.successors, peer, capacity),
fingers: finger_join(state.local, &state.fingers, peer),
..state.clone()
},
actions: vec![TopologyAction::FindSuccessorForConnect {
next: peer,
did: state.local,
}],
}
}
fn step_remove(state: &TopologyState, peer: Did, capacity: usize) -> TopologyStep {
let mut next_successors = state
.successors
.iter()
.copied()
.filter(|&did| did != peer)
.collect::<Vec<_>>();
let fingers = finger_remove(&state.fingers, peer);
if next_successors.is_empty() {
if let Some(first_finger) = fingers.iter().flatten().copied().next() {
next_successors =
update_successors(state.local, &next_successors, first_finger, capacity);
}
}
TopologyStep {
state: TopologyState {
successors: next_successors,
predecessor: state.predecessor.filter(|&did| did != peer),
fingers,
..state.clone()
},
actions: Vec::new(),
}
}
fn step_update_successor(state: &TopologyState, successor: Did, capacity: usize) -> TopologyStep {
let next_successors = update_successors(state.local, &state.successors, successor, capacity);
let inserted = !state.successors.contains(&successor) && next_successors.contains(&successor);
TopologyStep {
state: TopologyState {
successors: next_successors,
..state.clone()
},
actions: if inserted {
vec![TopologyAction::QuerySuccessorList(successor)]
} else {
Vec::new()
},
}
}
fn step_fix_finger(state: &TopologyState) -> TopologyStep {
if state.fingers.is_empty() {
return TopologyStep {
state: state.clone(),
actions: Vec::new(),
};
}
let index = (state.fix_finger_index + 1) % state.fingers.len();
let did = state.local + Did::power_of_two(index);
match find_successor(state, did) {
FindSuccessorStep::Local(successor) => TopologyStep {
state: TopologyState {
fingers: finger_set(state.local, &state.fingers, index, successor),
fix_finger_index: index,
..state.clone()
},
actions: Vec::new(),
},
FindSuccessorStep::Remote { next, did } => TopologyStep {
state: TopologyState {
fix_finger_index: index,
..state.clone()
},
actions: vec![TopologyAction::FindSuccessorForFix { next, did, index }],
},
}
}
pub fn step(state: &TopologyState, event: TopologyEvent, capacity: usize) -> TopologyStep {
match event {
TopologyEvent::Join { peer } => step_join(state, peer, capacity),
TopologyEvent::Remove { peer } => step_remove(state, peer, capacity),
TopologyEvent::UpdateSuccessor { successor } => {
step_update_successor(state, successor, capacity)
}
TopologyEvent::Notify { predecessor } => TopologyStep {
state: TopologyState {
predecessor: rectify_predecessor(state.local, state.predecessor, predecessor),
..state.clone()
},
actions: Vec::new(),
},
TopologyEvent::Stabilize {
successors: topo_successors,
predecessor: topo_predecessor,
} => {
let next_successors = stabilize_successors(
state.local,
&state.successors,
&topo_successors,
topo_predecessor,
capacity,
);
let mut actions = Vec::new();
if let Some(query) = stabilize_query(state.local, &state.successors, topo_predecessor) {
actions.push(TopologyAction::QuerySuccessorList(query));
}
if let Some(notify) = stabilize_notify(state.local, &next_successors) {
actions.push(TopologyAction::Notify(notify));
}
TopologyStep {
state: TopologyState {
successors: next_successors,
..state.clone()
},
actions,
}
}
TopologyEvent::FixFinger => step_fix_finger(state),
TopologyEvent::ApplyFinger { index, successor } => TopologyStep {
state: TopologyState {
fingers: finger_set(state.local, &state.fingers, index, successor),
..state.clone()
},
actions: Vec::new(),
},
}
}
#[cfg(test)]
mod tests {
use num_bigint::BigUint;
use super::*;
fn did(value: u32) -> Did {
Did::from(value)
}
fn state(
local: Did,
successors: Vec<Did>,
predecessor: Option<Did>,
fingers: Vec<Option<Did>>,
fix_finger_index: usize,
) -> TopologyState {
TopologyState::new(local, successors, predecessor, fingers, fix_finger_index)
}
fn successor_distances(local: Did, successors: &[Did], capacity: usize) -> Vec<BigUint> {
let infinity = BigUint::from(1u8) << RING_BITS;
(0..capacity)
.map(|index| {
successors
.get(index)
.map(|successor| dist(local, *successor))
.unwrap_or_else(|| infinity.clone())
})
.collect()
}
fn refines_successor_distances(before: &TopologyState, after: &TopologyState) -> bool {
let before_distances =
successor_distances(before.local, &before.successors, DEFAULT_SUCCESSOR_CAPACITY);
let after_distances =
successor_distances(after.local, &after.successors, DEFAULT_SUCCESSOR_CAPACITY);
before_distances
.iter()
.zip(after_distances.iter())
.all(|(before, after)| after <= before)
}
#[test]
fn join_step_updates_successors_fingers_and_connect_action() {
let local = did(0);
let peer = did(8);
let next = step(
&state(local, vec![], None, vec![None; 5], 0),
TopologyEvent::Join { peer },
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(next.state.successors, vec![peer]);
assert_eq!(next.state.fingers, vec![
Some(peer),
Some(peer),
Some(peer),
Some(peer),
None
]);
assert_eq!(next.actions, vec![
TopologyAction::FindSuccessorForConnect {
next: peer,
did: local
}
]);
}
#[test]
fn join_step_refines_successor_distance_vector() {
let local = did(0);
let current = state(local, vec![did(20), did(40)], None, vec![None; 5], 0);
let next = step(
¤t,
TopologyEvent::Join { peer: did(10) },
DEFAULT_SUCCESSOR_CAPACITY,
);
assert!(refines_successor_distances(¤t, &next.state));
}
#[test]
fn stabilize_step_refines_successor_distance_vector() {
let local = did(0);
let current = state(local, vec![did(40)], None, vec![None; 5], 0);
let next = step(
¤t,
TopologyEvent::Stabilize {
successors: vec![did(50), did(60)],
predecessor: Some(did(10)),
},
DEFAULT_SUCCESSOR_CAPACITY,
);
assert!(refines_successor_distances(¤t, &next.state));
}
#[test]
fn remove_step_removes_peer_from_every_topology_slot() {
let local = did(0);
let peer = did(8);
let next = step(
&state(
local,
vec![peer],
Some(peer),
vec![Some(peer), Some(peer)],
0,
),
TopologyEvent::Remove { peer },
DEFAULT_SUCCESSOR_CAPACITY,
);
assert!(next.state.successors.is_empty());
assert_eq!(next.state.predecessor, None);
assert_eq!(next.state.fingers, vec![None, None]);
assert!(next.actions.is_empty());
}
#[test]
fn fix_finger_step_updates_local_successor_slot() {
let local = did(0);
let successor = did(8);
let next = step(
&state(local, vec![successor], None, vec![None; 4], 2),
TopologyEvent::FixFinger,
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(next.state.fix_finger_index, 3);
assert_eq!(next.state.fingers, vec![None, None, None, Some(successor)]);
assert!(next.actions.is_empty());
}
#[test]
fn fix_finger_step_emits_indexed_remote_action() {
let local = did(0);
let successor = did(4);
let next_hop = did(6);
let next = step(
&state(
local,
vec![successor],
None,
vec![None, None, Some(next_hop), None],
2,
),
TopologyEvent::FixFinger,
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(next.state.fix_finger_index, 3);
assert_eq!(next.actions, vec![TopologyAction::FindSuccessorForFix {
next: next_hop,
did: Did::power_of_two(3),
index: 3
}]);
}
#[test]
fn fix_finger_step_queries_local_relative_probe() {
let local = did(100);
let successor = did(104);
let next_hop = did(106);
let next = step(
&state(
local,
vec![successor],
None,
vec![None, None, Some(next_hop), None],
2,
),
TopologyEvent::FixFinger,
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(next.state.fix_finger_index, 3);
assert_eq!(next.actions, vec![TopologyAction::FindSuccessorForFix {
next: next_hop,
did: local + Did::power_of_two(3),
index: 3
}]);
}
#[test]
fn apply_finger_step_updates_exact_slot() {
let local = did(0);
let successor = did(8);
let next = step(
&state(local, vec![], None, vec![None; 4], 0),
TopologyEvent::ApplyFinger {
index: 2,
successor,
},
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(next.state.fingers, vec![None, None, Some(successor), None]);
assert!(next.actions.is_empty());
}
#[test]
fn apply_finger_step_ignores_self_and_out_of_range_slot() {
let local = did(0);
let current = state(local, vec![], None, vec![None; 2], 0);
let self_update = step(
¤t,
TopologyEvent::ApplyFinger {
index: 1,
successor: local,
},
DEFAULT_SUCCESSOR_CAPACITY,
);
let out_of_range = step(
¤t,
TopologyEvent::ApplyFinger {
index: 9,
successor: did(9),
},
DEFAULT_SUCCESSOR_CAPACITY,
);
assert_eq!(self_update.state, current);
assert_eq!(out_of_range.state, current);
}
}