use arrayvec::ArrayVec;
use bigint::U512;
use multihash::Multihash;
use std::mem;
use std::slice::IterMut as SliceIterMut;
use std::time::{Duration, Instant};
use std::vec::IntoIter as VecIntoIter;
pub const MAX_NODES_PER_BUCKET: usize = 20;
#[derive(Debug, Clone)]
pub struct KBucketsTable<Id, Val> {
my_id: Id,
tables: Vec<KBucket<Id, Val>>,
ping_timeout: Duration,
}
#[derive(Debug, Clone)]
struct KBucket<Id, Val> {
nodes: ArrayVec<[Node<Id, Val>; MAX_NODES_PER_BUCKET]>,
pending_node: Option<(Node<Id, Val>, Instant)>,
last_update: Instant,
}
#[derive(Debug, Clone)]
struct Node<Id, Val> {
id: Id,
value: Val,
}
impl<Id, Val> KBucket<Id, Val> {
fn flush(&mut self, timeout: Duration) {
if let Some((pending_node, instant)) = self.pending_node.take() {
if instant.elapsed() >= timeout {
let _ = self.nodes.remove(0);
self.nodes.push(pending_node);
} else {
self.pending_node = Some((pending_node, instant));
}
}
}
}
pub trait KBucketsPeerId: Eq + Clone {
fn distance_with(&self, other: &Self) -> u32;
fn max_distance() -> usize;
}
impl KBucketsPeerId for Multihash {
#[inline]
fn distance_with(&self, other: &Self) -> u32 {
let my_hash = U512::from(self.digest());
let other_hash = U512::from(other.digest());
let xor = my_hash ^ other_hash;
512 - xor.leading_zeros()
}
#[inline]
fn max_distance() -> usize {
512
}
}
impl<Id, Val> KBucketsTable<Id, Val>
where
Id: KBucketsPeerId,
{
pub fn new(my_id: Id, ping_timeout: Duration) -> Self {
KBucketsTable {
my_id: my_id,
tables: (0..Id::max_distance())
.map(|_| KBucket {
nodes: ArrayVec::new(),
pending_node: None,
last_update: Instant::now(),
})
.collect(),
ping_timeout: ping_timeout,
}
}
#[inline]
fn bucket_num(&self, id: &Id) -> Option<usize> {
(self.my_id.distance_with(id) as usize).checked_sub(1)
}
#[inline]
pub fn buckets(&mut self) -> BucketsIter<Id, Val> {
BucketsIter(self.tables.iter_mut(), self.ping_timeout)
}
#[inline]
pub fn my_id(&self) -> &Id {
&self.my_id
}
pub fn find_closest(&mut self, id: &Id) -> VecIntoIter<Id>
where
Id: Clone,
{
let mut out = Vec::new();
for table in self.tables.iter_mut() {
table.flush(self.ping_timeout);
if table.last_update.elapsed() > self.ping_timeout {
continue;
}
for node in table.nodes.iter() {
out.push(node.id.clone());
}
}
out.sort_by(|a, b| b.distance_with(id).cmp(&a.distance_with(id)));
out.into_iter()
}
pub fn find_closest_with_self(&mut self, id: &Id) -> VecIntoIter<Id>
where
Id: Clone,
{
let mut intermediate: Vec<_> = self.find_closest(&id).collect();
if let Some(pos) = intermediate
.iter()
.position(|e| e.distance_with(&id) >= self.my_id.distance_with(&id))
{
if intermediate[pos] != self.my_id {
intermediate.insert(pos, self.my_id.clone());
}
} else {
intermediate.push(self.my_id.clone());
}
intermediate.into_iter()
}
pub fn update(&mut self, id: Id, value: Val) -> UpdateOutcome<Id, Val> {
let table = match self.bucket_num(&id) {
Some(n) => &mut self.tables[n],
None => return UpdateOutcome::FailSelfUpdate,
};
table.flush(self.ping_timeout);
if let Some(pos) = table.nodes.iter().position(|n| n.id == id) {
let mut existing = table.nodes.remove(pos);
let old_val = mem::replace(&mut existing.value, value);
if pos == 0 {
table.nodes.truncate(MAX_NODES_PER_BUCKET - 1);
table.pending_node = None;
}
table.nodes.push(existing);
table.last_update = Instant::now();
UpdateOutcome::Refreshed(old_val)
} else if table.nodes.len() < MAX_NODES_PER_BUCKET {
table.nodes.push(Node {
id: id,
value: value,
});
table.last_update = Instant::now();
UpdateOutcome::Added
} else {
if table.pending_node.is_none() {
table.pending_node = Some((
Node {
id: id,
value: value,
},
Instant::now(),
));
UpdateOutcome::NeedPing(table.nodes[0].id.clone())
} else {
UpdateOutcome::Discarded
}
}
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[must_use]
pub enum UpdateOutcome<Id, Val> {
Added,
Refreshed(Val),
NeedPing(Id),
Discarded,
FailSelfUpdate,
}
pub struct BucketsIter<'a, Id: 'a, Val: 'a>(SliceIterMut<'a, KBucket<Id, Val>>, Duration);
impl<'a, Id: 'a, Val: 'a> Iterator for BucketsIter<'a, Id, Val> {
type Item = Bucket<'a, Id, Val>;
#[inline]
fn next(&mut self) -> Option<Self::Item> {
self.0.next().map(|bucket| {
bucket.flush(self.1);
Bucket(bucket)
})
}
#[inline]
fn size_hint(&self) -> (usize, Option<usize>) {
self.0.size_hint()
}
}
impl<'a, Id: 'a, Val: 'a> ExactSizeIterator for BucketsIter<'a, Id, Val> {}
pub struct Bucket<'a, Id: 'a, Val: 'a>(&'a mut KBucket<Id, Val>);
impl<'a, Id: 'a, Val: 'a> Bucket<'a, Id, Val> {
#[inline]
pub fn num_entries(&self) -> usize {
self.0.nodes.len()
}
#[inline]
pub fn has_pending(&self) -> bool {
self.0.pending_node.is_some()
}
#[inline]
pub fn last_update(&self) -> Instant {
self.0.last_update.clone()
}
}
#[cfg(test)]
mod tests {
extern crate rand;
use self::rand::random;
use kbucket::{KBucketsPeerId, KBucketsTable, UpdateOutcome, MAX_NODES_PER_BUCKET};
use multihash::{Multihash, Hash};
use std::thread;
use std::time::Duration;
#[test]
fn basic_closest() {
let my_id = Multihash::random(Hash::SHA2256);
let other_id = Multihash::random(Hash::SHA2256);
let mut table = KBucketsTable::new(my_id, Duration::from_secs(5));
let _ = table.update(other_id.clone(), ());
let res = table.find_closest(&other_id).collect::<Vec<_>>();
assert_eq!(res.len(), 1);
assert_eq!(res[0], other_id);
}
#[test]
fn update_local_id_fails() {
let my_id = Multihash::random(Hash::SHA2256);
let mut table = KBucketsTable::new(my_id.clone(), Duration::from_secs(5));
match table.update(my_id, ()) {
UpdateOutcome::FailSelfUpdate => (),
_ => panic!(),
}
}
#[test]
fn update_time_last_refresh() {
let my_id = Multihash::random(Hash::SHA2256);
let other_ids = (0..random::<usize>() % 20)
.map(|_| {
let bit_num = random::<usize>() % 256;
let mut id = my_id.as_bytes().to_vec().clone();
id[33 - (bit_num / 8)] ^= 1 << (bit_num % 8);
(Multihash::from_bytes(id).unwrap(), bit_num)
})
.collect::<Vec<_>>();
let mut table = KBucketsTable::new(my_id, Duration::from_secs(5));
let before_update = table.buckets().map(|b| b.last_update()).collect::<Vec<_>>();
thread::sleep(Duration::from_secs(2));
for &(ref id, _) in &other_ids {
let _ = table.update(id.clone(), ());
}
let after_update = table.buckets().map(|b| b.last_update()).collect::<Vec<_>>();
for (offset, (bef, aft)) in before_update.iter().zip(after_update.iter()).enumerate() {
if other_ids.iter().any(|&(_, bucket)| bucket == offset) {
assert_ne!(bef, aft);
} else {
assert_eq!(bef, aft);
}
}
}
#[test]
fn full_kbucket() {
let my_id = Multihash::random(Hash::SHA2256);
assert!(MAX_NODES_PER_BUCKET <= 251);
let mut fill_ids = (0..MAX_NODES_PER_BUCKET + 3)
.map(|n| {
let mut id = my_id.clone().into_bytes();
id[2] ^= 0x80;
id[33] = id[33].wrapping_add(n as u8);
Multihash::from_bytes(id).unwrap()
})
.collect::<Vec<_>>();
let first_node = fill_ids[0].clone();
let second_node = fill_ids[1].clone();
let mut table = KBucketsTable::new(my_id.clone(), Duration::from_secs(1));
for (num, id) in fill_ids.drain(..MAX_NODES_PER_BUCKET).enumerate() {
assert_eq!(table.update(id, ()), UpdateOutcome::Added);
assert_eq!(table.buckets().nth(255).unwrap().num_entries(), num + 1);
}
assert_eq!(
table.buckets().nth(255).unwrap().num_entries(),
MAX_NODES_PER_BUCKET
);
assert!(!table.buckets().nth(255).unwrap().has_pending());
assert_eq!(
table.update(fill_ids.remove(0), ()),
UpdateOutcome::NeedPing(first_node)
);
assert_eq!(
table.buckets().nth(255).unwrap().num_entries(),
MAX_NODES_PER_BUCKET
);
assert!(table.buckets().nth(255).unwrap().has_pending());
assert_eq!(
table.update(fill_ids.remove(0), ()),
UpdateOutcome::Discarded
);
thread::sleep(Duration::from_secs(2));
assert!(!table.buckets().nth(255).unwrap().has_pending());
assert_eq!(
table.update(fill_ids.remove(0), ()),
UpdateOutcome::NeedPing(second_node)
);
}
#[test]
fn self_distance_zero() {
let a = Multihash::random(Hash::SHA2256);
assert_eq!(a.distance_with(&a), 0);
}
#[test]
fn distance_correct_order() {
let a = Multihash::random(Hash::SHA2256);
let b = Multihash::random(Hash::SHA2256);
assert!(a.distance_with(&a) < b.distance_with(&a));
assert!(a.distance_with(&b) > b.distance_with(&b));
}
}