libp2p_kad/kbucket.rs
1// Copyright 2018 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21//! Implementation of a Kademlia routing table as used by a single peer
22//! participating in a Kademlia DHT.
23//!
24//! The entry point for the API of this module is a [`KBucketsTable`].
25//!
26//! ## Pending Insertions
27//!
28//! When the bucket associated with the `Key` of an inserted entry is full
29//! but contains disconnected nodes, it accepts a [`PendingEntry`].
30//! Pending entries are inserted lazily when their timeout is found to be expired
31//! upon querying the `KBucketsTable`. When that happens, the `KBucketsTable` records
32//! an [`AppliedPending`] result which must be consumed by calling [`take_applied_pending`]
33//! regularly and / or after performing lookup operations like [`entry`] and [`closest`].
34//!
35//! [`entry`]: KBucketsTable::entry
36//! [`closest`]: KBucketsTable::closest
37//! [`AppliedPending`]: bucket::AppliedPending
38//! [`take_applied_pending`]: KBucketsTable::take_applied_pending
39//! [`PendingEntry`]: entry::PendingEntry
40
41// [Implementation Notes]
42//
43// 1. Routing Table Layout
44//
45// The routing table is currently implemented as a fixed-size "array" of
46// buckets, ordered by increasing distance relative to a local key
47// that identifies the local peer. This is an often-used, simplified
48// implementation that approximates the properties of the b-tree (or prefix tree)
49// implementation described in the full paper [0], whereby buckets are split on-demand.
50// This should be treated as an implementation detail, however, so that the
51// implementation may change in the future without breaking the API.
52//
53// 2. Replacement Cache
54//
55// In this implementation, the "replacement cache" for unresponsive peers
56// consists of a single entry per bucket. Furthermore, this implementation is
57// currently tailored to connection-oriented transports, meaning that the
58// "LRU"-based ordering of entries in a bucket is actually based on the last reported
59// connection status of the corresponding peers, from least-recently (dis)connected to
60// most-recently (dis)connected, and controlled through the `Entry` API. As a result,
61// the nodes in the buckets are not reordered as a result of RPC activity, but only as a
62// result of nodes being marked as connected or disconnected. In particular,
63// if a bucket is full and contains only entries for peers that are considered
64// connected, no pending entry is accepted. See the `bucket` submodule for
65// further details.
66//
67// [0]: https://pdos.csail.mit.edu/~petar/papers/maymounkov-kademlia-lncs.pdf
68
69mod bucket;
70mod entry;
71#[allow(clippy::ptr_offset_with_cast)]
72#[allow(clippy::assign_op_pattern)]
73mod key;
74mod sub_bucket;
75mod swamp;
76mod weighted;
77mod weighted_iter;
78
79pub use entry::*;
80pub use sub_bucket::*;
81
82use crate::kbucket::weighted_iter::WeightedIter;
83use bucket::KBucket;
84use libp2p_core::identity::{Keypair, PublicKey};
85use log::debug;
86use std::collections::VecDeque;
87use std::fmt::Debug;
88use std::time::Duration;
89use derivative::Derivative;
90
91/// Maximum number of k-buckets.
92const NUM_BUCKETS: usize = 256;
93
94/// A `KBucketsTable` represents a Kademlia routing table.
95#[derive(Derivative)]
96#[derivative(Debug, Clone)]
97pub struct KBucketsTable<TKey, TVal> {
98 #[derivative(Debug="ignore")]
99 local_kp: Keypair,
100 /// The key identifying the local peer that owns the routing table.
101 local_key: TKey,
102 /// The buckets comprising the routing table.
103 pub(super) buckets: Vec<KBucket<TKey, TVal>>,
104 /// The list of evicted entries that have been replaced with pending
105 /// entries since the last call to [`KBucketsTable::take_applied_pending`].
106 applied_pending: VecDeque<AppliedPending<TKey, TVal>>,
107}
108
109/// A (type-safe) index into a `KBucketsTable`, i.e. a non-negative integer in the
110/// interval `[0, NUM_BUCKETS)`.
111#[derive(Debug, Copy, Clone, PartialEq, Eq)]
112pub struct BucketIndex(usize);
113
114impl BucketIndex {
115 /// Creates a new `BucketIndex` for a `Distance`.
116 ///
117 /// The given distance is interpreted as the distance from a `local_key` of
118 /// a `KBucketsTable`. If the distance is zero, `None` is returned, in
119 /// recognition of the fact that the only key with distance `0` to a
120 /// `local_key` is the `local_key` itself, which does not belong in any
121 /// bucket.
122 fn new(d: &Distance) -> Option<BucketIndex> {
123 d.ilog2().map(|i| BucketIndex(i as usize))
124 }
125
126 /// Gets the index value as an unsigned integer.
127 pub fn get(&self) -> usize {
128 self.0
129 }
130
131 /// Returns the minimum inclusive and maximum inclusive [`Distance`]
132 /// included in the bucket for this index.
133 fn range(&self) -> (Distance, Distance) {
134 let min = Distance(U256::pow(U256::from(2), U256::from(self.0)));
135 if self.0 == usize::from(u8::MAX) {
136 (min, Distance(U256::MAX))
137 } else {
138 let max = Distance(U256::pow(U256::from(2), U256::from(self.0 + 1)) - 1);
139 (min, max)
140 }
141 }
142
143 /// Generates a random distance that falls into the bucket for this index.
144 fn rand_distance(&self, rng: &mut impl rand::Rng) -> Distance {
145 let mut bytes = [0u8; 32];
146 let quot = self.0 / 8;
147 for i in 0..quot {
148 bytes[31 - i] = rng.gen();
149 }
150 let rem = (self.0 % 8) as u32;
151 let lower = usize::pow(2, rem);
152 let upper = usize::pow(2, rem + 1);
153 bytes[31 - quot] = rng.gen_range(lower, upper) as u8;
154 Distance(U256::from(bytes))
155 }
156}
157
158impl<TKey, TVal> KBucketsTable<TKey, TVal>
159where
160 TKey: Clone + AsRef<KeyBytes>,
161 TVal: Clone,
162{
163 /// Creates a new, empty Kademlia routing table with entries partitioned
164 /// into buckets as per the Kademlia protocol.
165 ///
166 /// The given `pending_timeout` specifies the duration after creation of
167 /// a [`PendingEntry`] after which it becomes eligible for insertion into
168 /// a full bucket, replacing the least-recently (dis)connected node.
169 pub fn new(local_kp: Keypair, local_key: TKey, pending_timeout: Duration) -> Self {
170 KBucketsTable {
171 local_kp,
172 local_key,
173 buckets: (0..NUM_BUCKETS)
174 .map(|_| KBucket::new(pending_timeout))
175 .collect(),
176 applied_pending: VecDeque::new(),
177 }
178 }
179
180 /// Returns the local key.
181 pub fn local_key(&self) -> &TKey {
182 &self.local_key
183 }
184
185 pub fn local_public_key(&self) -> PublicKey {
186 self.local_kp.public()
187 }
188
189 /// Returns an `Entry` for the given key, representing the state of the entry
190 /// in the routing table.
191 pub fn entry<'a>(&'a mut self, key: &'a TKey) -> Entry<'a, TKey, TVal> {
192 let index = BucketIndex::new(&self.local_key.as_ref().distance(key));
193 if let Some(i) = index {
194 debug!(
195 "Node {} belongs to bucket {}",
196 bs58::encode(key.as_ref()).into_string(),
197 i.get()
198 );
199 let bucket = &mut self.buckets[i.get()];
200 self.applied_pending.extend(bucket.apply_pending());
201 Entry::new(bucket, key)
202 } else {
203 Entry::SelfEntry
204 }
205 }
206
207 /// Returns an iterator over all buckets.
208 ///
209 /// The buckets are ordered by proximity to the `local_key`, i.e. the first
210 /// bucket is the closest bucket (containing at most one key).
211 pub fn iter<'a>(&'a mut self) -> impl Iterator<Item = KBucketRef<'a, TKey, TVal>> + 'a {
212 let applied_pending = &mut self.applied_pending;
213 self.buckets.iter_mut().enumerate().map(move |(i, b)| {
214 applied_pending.extend(b.apply_pending());
215 KBucketRef {
216 index: BucketIndex(i),
217 bucket: b,
218 }
219 })
220 }
221
222 /// Returns the bucket for the distance to the given key.
223 ///
224 /// Returns `None` if the given key refers to the local key.
225 pub fn bucket<K>(&mut self, key: &K) -> Option<KBucketRef<'_, TKey, TVal>>
226 where
227 K: AsRef<KeyBytes>,
228 {
229 let d = self.local_key.as_ref().distance(key);
230 if let Some(index) = BucketIndex::new(&d) {
231 let bucket = &mut self.buckets[index.0];
232 self.applied_pending.extend(bucket.apply_pending());
233 Some(KBucketRef { bucket, index })
234 } else {
235 None
236 }
237 }
238
239 /// Consumes the next applied pending entry, if any.
240 ///
241 /// When an entry is attempted to be inserted and the respective bucket is full,
242 /// it may be recorded as pending insertion after a timeout, see [`InsertResult::Pending`].
243 ///
244 /// If the oldest currently disconnected entry in the respective bucket does not change
245 /// its status until the timeout of pending entry expires, it is evicted and
246 /// the pending entry inserted instead. These insertions of pending entries
247 /// happens lazily, whenever the `KBucketsTable` is accessed, and the corresponding
248 /// buckets are updated accordingly. The fact that a pending entry was applied is
249 /// recorded in the `KBucketsTable` in the form of `AppliedPending` results, which must be
250 /// consumed by calling this function.
251 pub fn take_applied_pending(&mut self) -> Option<AppliedPending<TKey, TVal>> {
252 self.applied_pending.pop_front()
253 }
254
255 /// Returns an iterator over the keys closest to `target`, ordered by
256 /// increasing distance.
257 pub fn closest_keys<'a, T>(&'a mut self, target: &'a T) -> impl Iterator<Item = TKey> + 'a
258 where
259 T: Clone + AsRef<KeyBytes>,
260 {
261 let distance = self.local_key.as_ref().distance(target);
262 WeightedIter::new(self, distance, target.as_ref()).map(|(n, _)| n.key.clone())
263 }
264
265 /// Returns an iterator over the nodes closest to the `target` key, ordered by
266 /// increasing distance.
267 pub fn closest<'a, T>(
268 &'a mut self,
269 target: &'a T,
270 ) -> impl Iterator<Item = EntryView<TKey, TVal>> + 'a
271 where
272 T: Clone + AsRef<KeyBytes>,
273 TVal: Clone,
274 {
275 let distance = self.local_key.as_ref().distance(target);
276 WeightedIter::new(self, distance, target.as_ref()).map(|(n, status)| EntryView {
277 node: n.clone(),
278 status,
279 })
280 }
281
282 /// Counts the number of nodes between the local node and the node
283 /// closest to `target`.
284 ///
285 /// The number of nodes between the local node and the target are
286 /// calculated by backtracking from the target towards the local key.
287 pub fn count_nodes_between<T>(&mut self, target: &T) -> usize
288 where
289 T: AsRef<KeyBytes>,
290 {
291 let local_key = self.local_key.clone();
292 let distance = target.as_ref().distance(&local_key);
293 let mut iter = ClosestBucketsIter::new(distance).take_while(|i| i.get() != 0);
294 if let Some(i) = iter.next() {
295 let num_first = self.buckets[i.get()]
296 .iter()
297 .filter(|(n, _)| n.key.as_ref().distance(&local_key) <= distance)
298 .count();
299 let num_rest: usize = iter.map(|i| self.buckets[i.get()].num_entries()).sum();
300 let result = num_first + num_rest;
301 debug!(
302 "There are {} nodes between local {} and remote {}",
303 result,
304 bs58::encode(local_key.as_ref()).into_string(),
305 bs58::encode(target.as_ref()).into_string()
306 );
307 result
308 } else {
309 0
310 }
311 }
312}
313
314/// An iterator over (some projection of) the closest entries in a
315/// `KBucketsTable` w.r.t. some target `Key`.
316struct ClosestIter<'a, TTarget, TKey, TVal, TMap, TOut> {
317 /// A reference to the target key whose distance to the local key determines
318 /// the order in which the buckets are traversed. The resulting
319 /// array from projecting the entries of each bucket using `fmap` is
320 /// sorted according to the distance to the target.
321 target: &'a TTarget,
322 /// A reference to all buckets of the `KBucketsTable`.
323 table: &'a mut KBucketsTable<TKey, TVal>,
324 /// The iterator over the bucket indices in the order determined by the
325 /// distance of the local key to the target.
326 buckets_iter: ClosestBucketsIter,
327 /// The iterator over the entries in the currently traversed bucket.
328 iter: Option<std::vec::IntoIter<TOut>>,
329 /// The projection function / mapping applied on each bucket as
330 /// it is encountered, producing the next `iter`ator.
331 fmap: TMap,
332}
333
334/// An iterator over the bucket indices, in the order determined by the `Distance` of
335/// a target from the `local_key`, such that the entries in the buckets are incrementally
336/// further away from the target, starting with the bucket covering the target.
337pub(super) struct ClosestBucketsIter {
338 /// The distance to the `local_key`.
339 distance: Distance,
340 /// The current state of the iterator.
341 state: ClosestBucketsIterState,
342}
343
344/// Operating states of a `ClosestBucketsIter`.
345#[derive(Debug)]
346enum ClosestBucketsIterState {
347 /// The starting state of the iterator yields the first bucket index and
348 /// then transitions to `ZoomIn`.
349 Start(BucketIndex),
350 /// The iterator "zooms in" to to yield the next bucket containing nodes that
351 /// are incrementally closer to the local node but further from the `target`.
352 /// These buckets are identified by a `1` in the corresponding bit position
353 /// of the distance bit string. When bucket `0` is reached, the iterator
354 /// transitions to `ZoomOut`.
355 ZoomIn(BucketIndex),
356 /// Once bucket `0` has been reached, the iterator starts "zooming out"
357 /// to buckets containing nodes that are incrementally further away from
358 /// both the local key and the target. These are identified by a `0` in
359 /// the corresponding bit position of the distance bit string. When bucket
360 /// `255` is reached, the iterator transitions to state `Done`.
361 ZoomOut(BucketIndex),
362 /// The iterator is in this state once it has visited all buckets.
363 Done,
364}
365
366impl ClosestBucketsIter {
367 pub(super) fn new(distance: Distance) -> Self {
368 let state = match BucketIndex::new(&distance) {
369 Some(i) => ClosestBucketsIterState::Start(i),
370 None => ClosestBucketsIterState::Start(BucketIndex(0)),
371 };
372 // println!("ClosestBucketsIter new: distance {} {}, state {:?}", Self::u256_binary(&distance.0, 256), distance.0.leading_zeros(), state);
373 Self { distance, state }
374 }
375
376 fn next_in(&self, idx: BucketIndex) -> Option<BucketIndex> {
377 (0..idx.get()).rev().find_map(|i| {
378 let bit = self.distance.0.bit(i);
379 if bit {
380 // println!("next_in {} [{}th = {}] bucket_idx: {:?}", Self::u256_binary(&self.distance.0, i), i, (bit as usize), idx);
381 Some(BucketIndex(i))
382 } else {
383 None
384 }
385 })
386 }
387
388 fn next_out(&self, idx: BucketIndex) -> Option<BucketIndex> {
389 (idx.get() + 1..NUM_BUCKETS).find_map(|i| {
390 let bit = self.distance.0.bit(i);
391 if !bit {
392 // println!("next_out {} [{}th = !{}] bucket_idx: {:?}", Self::u256_binary(&self.distance.0, i), i, (bit as usize), idx);
393 Some(BucketIndex(i))
394 } else {
395 None
396 }
397 })
398 }
399
400 fn u256_binary(u: &U256, highlight: usize) -> String {
401 let mut arr: [u8; 256] = [0; 256];
402 for i in 0..256 {
403 arr[i] = u.bit(i) as u8;
404 }
405
406 arr.iter()
407 .enumerate()
408 .map(|(i, u)| {
409 if i == highlight {
410 format!("-[{}]-", u)
411 } else {
412 u.to_string()
413 }
414 })
415 .collect()
416 }
417}
418
419impl Iterator for ClosestBucketsIter {
420 type Item = BucketIndex;
421
422 fn next(&mut self) -> Option<Self::Item> {
423 match self.state {
424 ClosestBucketsIterState::Start(i) => {
425 self.state = ClosestBucketsIterState::ZoomIn(i);
426 Some(i)
427 }
428 ClosestBucketsIterState::ZoomIn(i) => {
429 if let Some(i) = self.next_in(i) {
430 self.state = ClosestBucketsIterState::ZoomIn(i);
431 Some(i)
432 } else {
433 let i = BucketIndex(0);
434 self.state = ClosestBucketsIterState::ZoomOut(i);
435 Some(i)
436 }
437 }
438 ClosestBucketsIterState::ZoomOut(i) => {
439 if let Some(i) = self.next_out(i) {
440 self.state = ClosestBucketsIterState::ZoomOut(i);
441 Some(i)
442 } else {
443 self.state = ClosestBucketsIterState::Done;
444 None
445 }
446 }
447 ClosestBucketsIterState::Done => None,
448 }
449 }
450}
451
452impl<TTarget, TKey, TVal, TMap, TOut> Iterator for ClosestIter<'_, TTarget, TKey, TVal, TMap, TOut>
453where
454 TTarget: AsRef<KeyBytes>,
455 TKey: Clone + AsRef<KeyBytes>,
456 TVal: Clone,
457 TMap: Fn(&KBucket<TKey, TVal>) -> Vec<TOut>,
458 TOut: AsRef<KeyBytes>,
459{
460 type Item = TOut;
461
462 fn next(&mut self) -> Option<Self::Item> {
463 loop {
464 match &mut self.iter {
465 Some(iter) => match iter.next() {
466 Some(k) => {
467 debug!(
468 "ClosestIter: target = {}; next node {}",
469 bs58::encode(&self.target.as_ref()).into_string(),
470 bs58::encode(k.as_ref()).into_string()
471 );
472 return Some(k);
473 }
474 None => self.iter = None,
475 },
476 None => {
477 if let Some(i) = self.buckets_iter.next() {
478 let bucket = &mut self.table.buckets[i.get()];
479 self.table.applied_pending.extend(bucket.apply_pending());
480 let mut v = (self.fmap)(bucket);
481 v.sort_by(|a, b| {
482 Ord::cmp(
483 &self.target.as_ref().distance(a.as_ref()),
484 &self.target.as_ref().distance(b.as_ref()),
485 )
486 });
487 debug!(
488 "ClosestIter: target = {}; next bucket {} with {} nodes",
489 bs58::encode(&self.target.as_ref()).into_string(),
490 i.0,
491 v.len()
492 );
493 self.iter = Some(v.into_iter());
494 } else {
495 debug!(
496 "ClosestIter: target = {}; Finished.",
497 bs58::encode(&self.target.as_ref()).into_string()
498 );
499 return None;
500 }
501 }
502 }
503 }
504 }
505}
506
507/// A reference to a bucket in a [`KBucketsTable`].
508pub struct KBucketRef<'a, TKey, TVal> {
509 pub index: BucketIndex,
510 pub bucket: &'a mut KBucket<TKey, TVal>,
511}
512
513impl<'a, TKey, TVal> KBucketRef<'a, TKey, TVal>
514where
515 TKey: Clone + AsRef<KeyBytes>,
516 TVal: Clone,
517{
518 /// Returns the minimum inclusive and maximum inclusive [`Distance`] for
519 /// this bucket.
520 pub fn range(&self) -> (Distance, Distance) {
521 self.index.range()
522 }
523
524 /// Checks whether the bucket is empty.
525 pub fn is_empty(&self) -> bool {
526 self.num_entries() == 0
527 }
528
529 /// Returns the number of entries in the bucket.
530 pub fn num_entries(&self) -> usize {
531 self.bucket.num_entries()
532 }
533
534 /// Returns true if the bucket has a pending node.
535 pub fn has_pending(&self) -> bool {
536 self.bucket.has_pending()
537 }
538
539 /// Tests whether the given distance falls into this bucket.
540 pub fn contains(&self, d: &Distance) -> bool {
541 BucketIndex::new(d).map_or(false, |i| i == self.index)
542 }
543
544 /// Generates a random distance that falls into this bucket.
545 ///
546 /// Together with a known key `a` (e.g. the local key), a random distance `d` for
547 /// this bucket w.r.t `k` gives rise to the corresponding (random) key `b` s.t.
548 /// the XOR distance between `a` and `b` is `d`. In other words, it gives
549 /// rise to a random key falling into this bucket. See [`key::Key::for_distance`].
550 pub fn rand_distance(&self, rng: &mut impl rand::Rng) -> Distance {
551 self.index.rand_distance(rng)
552 }
553
554 /// Returns an iterator over the entries in the bucket.
555 pub fn iter(&'a self) -> impl Iterator<Item = EntryRefView<'a, TKey, TVal>> {
556 self.bucket.iter().map(move |(n, status)| {
557 EntryRefView {
558 node: NodeRefView {
559 key: &n.key,
560 value: &n.value
561 },
562 status
563 }
564 })
565 }
566}
567
568#[cfg(test)]
569mod tests {
570 use super::*;
571 use libp2p_core::PeerId;
572 use quickcheck::*;
573 use rand::Rng;
574 use std::time::Instant;
575
576 type TestTable = KBucketsTable<KeyBytes, ()>;
577
578 impl Arbitrary for TestTable {
579 fn arbitrary<G: Gen>(g: &mut G) -> TestTable {
580 let keypair = Keypair::generate_ed25519();
581 let public_key = keypair.public();
582 let local_key = Key::from(PeerId::from(public_key));
583 let timeout = Duration::from_secs(g.gen_range(1, 360));
584
585 let mut table = TestTable::new(keypair, local_key.clone().into(), timeout);
586 let mut num_total = g.gen_range(0, 100);
587 for (i, b) in &mut table.buckets.iter_mut().enumerate().rev() {
588 let ix = BucketIndex(i);
589 let num = g.gen_range(0, usize::min(K_VALUE.get(), num_total) + 1);
590 num_total -= num;
591 for _ in 0..num {
592 let distance = ix.rand_distance(g);
593 let key = local_key.for_distance(distance);
594 let node = Node {
595 key: key.clone(),
596 value: (),
597 weight: 0,
598 }; // TODO: arbitrary weight
599 let status = NodeStatus::arbitrary(g);
600 match b.insert(node, status) {
601 InsertResult::Inserted => {}
602 _ => panic!(),
603 }
604 }
605 }
606 table
607 }
608 }
609
610 #[test]
611 fn buckets_are_non_overlapping_and_exhaustive() {
612 let keypair = Keypair::generate_ed25519();
613 let public_key = keypair.public();
614 let local_key = Key::from(PeerId::from(public_key));
615 let timeout = Duration::from_secs(0);
616 let mut table = KBucketsTable::<KeyBytes, ()>::new(keypair, local_key.into(), timeout);
617
618 let mut prev_max = U256::from(0);
619
620 for bucket in table.iter() {
621 let (min, max) = bucket.range();
622 assert_eq!(Distance(prev_max + U256::from(1)), min);
623 prev_max = max.0;
624 }
625
626 assert_eq!(U256::MAX, prev_max);
627 }
628
629 #[test]
630 fn bucket_contains_range() {
631 fn prop(ix: u8) {
632 let index = BucketIndex(ix as usize);
633 let mut bucket = KBucket::<Key<PeerId>, ()>::new(Duration::from_secs(0));
634 let bucket_ref = KBucketRef {
635 index,
636 bucket: &mut bucket,
637 };
638
639 let (min, max) = bucket_ref.range();
640
641 assert!(min <= max);
642
643 assert!(bucket_ref.contains(&min));
644 assert!(bucket_ref.contains(&max));
645
646 assert!(!bucket_ref.contains(&Distance(min.0 - 1)));
647 assert!(!bucket_ref.contains(&Distance(max.0 + 1)));
648 }
649
650 quickcheck(prop as fn(_));
651 }
652
653 #[test]
654 fn rand_distance() {
655 fn prop(ix: u8) -> bool {
656 let d = BucketIndex(ix as usize).rand_distance(&mut rand::thread_rng());
657 let n = U256::from(<[u8; 32]>::from(d.0));
658 let b = U256::from(2);
659 let e = U256::from(ix);
660 let lower = b.pow(e);
661 let upper = b.pow(e + U256::from(1)) - U256::from(1);
662 lower <= n && n <= upper
663 }
664 quickcheck(prop as fn(_) -> _);
665 }
666
667 #[test]
668 fn entry_inserted() {
669 let keypair = Keypair::generate_ed25519();
670 let public_key = keypair.public();
671 let local_key = Key::from(PeerId::from(public_key));
672 let other_id = Key::from(PeerId::random());
673 let other_weight = 0; // TODO: random weight
674
675 let mut table = KBucketsTable::<_, ()>::new(keypair, local_key, Duration::from_secs(5));
676 if let Entry::Absent(entry) = table.entry(&other_id) {
677 match entry.insert((), NodeStatus::Connected, other_weight) {
678 InsertResult::Inserted => (),
679 _ => panic!(),
680 }
681 } else {
682 panic!()
683 }
684
685 let res = table.closest_keys(&other_id).collect::<Vec<_>>();
686 assert_eq!(res.len(), 1);
687 assert_eq!(res[0], other_id);
688 }
689
690 #[test]
691 fn entry_self() {
692 let keypair = Keypair::generate_ed25519();
693 let public_key = keypair.public();
694 let local_key = Key::from(PeerId::from(public_key));
695 let mut table =
696 KBucketsTable::<_, ()>::new(keypair, local_key.clone(), Duration::from_secs(5));
697 match table.entry(&local_key) {
698 Entry::SelfEntry => (),
699 _ => panic!(),
700 }
701 }
702
703 #[test]
704 fn closest() {
705 let keypair = Keypair::generate_ed25519();
706 let public_key = keypair.public();
707 let local_key = Key::from(PeerId::from(public_key));
708 let mut table = KBucketsTable::<_, ()>::new(keypair, local_key, Duration::from_secs(5));
709 let mut count = 0;
710 loop {
711 if count == 100 {
712 break;
713 }
714 let key = Key::from(PeerId::random());
715 if let Entry::Absent(e) = table.entry(&key) {
716 match e.insert((), NodeStatus::Connected, 0) {
717 // TODO: random weight
718 InsertResult::Inserted => count += 1,
719 _ => continue,
720 }
721 } else {
722 panic!("entry exists")
723 }
724 }
725
726 let mut expected_keys: Vec<_> = table
727 .buckets
728 .iter()
729 .flat_map(|t| t.iter().map(|(n, _)| n.key.clone()))
730 .collect();
731
732 for _ in 0..10 {
733 let target_key = Key::from(PeerId::random());
734 let keys = table.closest_keys(&target_key).collect::<Vec<_>>();
735 // The list of keys is expected to match the result of a full-table scan.
736 expected_keys.sort_by_key(|k| k.distance(&target_key));
737 assert_eq!(keys, expected_keys);
738 }
739 }
740
741 #[test]
742 fn applied_pending() {
743 let keypair = Keypair::generate_ed25519();
744 let public_key = keypair.public();
745 let local_key = Key::from(PeerId::from(public_key));
746 let mut table =
747 KBucketsTable::<_, ()>::new(keypair, local_key.clone(), Duration::from_millis(1));
748
749 let expected_applied;
750 let full_bucket_index;
751 loop {
752 let key = Key::from(PeerId::random()); // generate random peer_id
753 if let Entry::Absent(e) = table.entry(&key) {
754 // check it's not yet in any bucket
755 // TODO: random weight
756 match e.insert((), NodeStatus::Disconnected, 0) {
757 // insert it into some bucket (node Disconnected status)
758 InsertResult::Full => {
759 // keep inserting until some random bucket is full (see continue below)
760 if let Entry::Absent(e) = table.entry(&key) {
761 // insertion didn't succeeded => no such key in a table
762 // TODO: random weight
763 match e.insert((), NodeStatus::Connected, 0) {
764 // insert it but now with Connected status
765 InsertResult::Pending { disconnected } => {
766 // insertion of a connected node into full bucket should produce Pending
767 expected_applied = AppliedPending {
768 inserted: Node {
769 key: key.clone(),
770 value: (),
771 weight: 0,
772 }, // TODO: random weight
773 evicted: Some(Node {
774 key: disconnected,
775 value: (),
776 weight: 0,
777 }), // TODO: random weight
778 };
779 full_bucket_index = BucketIndex::new(&key.distance(&local_key));
780 break;
781 }
782 _ => panic!(),
783 }
784 } else {
785 panic!()
786 }
787 }
788 _ => continue,
789 }
790 } else {
791 panic!("entry exists")
792 }
793 }
794
795 // Expire the timeout for the pending entry on the full bucket.`
796 let full_bucket = &mut table.buckets[full_bucket_index.unwrap().get()];
797 let elapsed = Instant::now() - Duration::from_secs(1);
798 full_bucket
799 .pending_mut(&expected_applied.inserted.key)
800 .unwrap()
801 .set_ready_at(elapsed);
802
803 // Calling table.entry() has a side-effect of applying pending nodes
804 match table.entry(&expected_applied.inserted.key) {
805 Entry::Present(_, NodeStatus::Connected) => {}
806 x => panic!("Unexpected entry: {:?}", x),
807 }
808
809 match table.entry(&expected_applied.evicted.as_ref().unwrap().key) {
810 Entry::Absent(_) => {}
811 x => panic!("Unexpected entry: {:?}", x),
812 }
813
814 assert_eq!(Some(expected_applied), table.take_applied_pending());
815 assert_eq!(None, table.take_applied_pending());
816 }
817
818 #[test]
819 fn count_nodes_between() {
820 fn prop(mut table: TestTable, target: Key<PeerId>) -> bool {
821 let num_to_target = table.count_nodes_between(&target);
822 let distance = table.local_key.distance(&target);
823 let base2 = U256::from(2);
824 let mut iter = ClosestBucketsIter::new(distance);
825 iter.all(|i| {
826 // Flip the distance bit related to the bucket.
827 let d = Distance(distance.0 ^ (base2.pow(U256::from(i.get()))));
828 let k = table.local_key.for_distance(d);
829 if distance.0.bit(i.get()) {
830 // Bit flip `1` -> `0`, the key must be closer than `target`.
831 d < distance && table.count_nodes_between(&k) <= num_to_target
832 } else {
833 // Bit flip `0` -> `1`, the key must be farther than `target`.
834 d > distance && table.count_nodes_between(&k) >= num_to_target
835 }
836 })
837 }
838
839 QuickCheck::new()
840 .tests(10)
841 .quickcheck(prop as fn(_, _) -> _)
842 }
843}