table.go 14 KB
Newer Older
1 2
// package kbucket implements a kademlia 'k-bucket' routing table.
package kbucket
3 4

import (
5
	"context"
6
	"errors"
7
	"fmt"
8
	"sync"
9
	"time"
10

11 12 13
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-core/peerstore"

George Antoniadis's avatar
George Antoniadis committed
14
	logging "github.com/ipfs/go-log"
15 16
)

Jeromy's avatar
Jeromy committed
17
var log = logging.Logger("table")
18

19 20 21
var ErrPeerRejectedHighLatency = errors.New("peer rejected; latency too high")
var ErrPeerRejectedNoCapacity = errors.New("peer rejected; insufficient capacity")

22 23
// RoutingTable defines the routing table.
type RoutingTable struct {
24
	// the routing table context
25
	ctx context.Context
26 27 28
	// function to cancel the RT context
	ctxCancel context.CancelFunc

29 30 31 32 33 34
	// ID of the local peer
	local ID

	// Blanket lock, refine later for better performance
	tabLock sync.RWMutex

35
	// latency metrics
36
	metrics peerstore.Metrics
37

38 39 40
	// Maximum acceptable latency for peers in this cluster
	maxLatency time.Duration

41
	// kBuckets define all the fingers to other nodes.
Aarsh Shah's avatar
Aarsh Shah committed
42
	buckets    []*bucket
43
	bucketsize int
Jeromy's avatar
Jeromy committed
44

Aarsh Shah's avatar
Aarsh Shah committed
45 46 47
	cplRefreshLk   sync.RWMutex
	cplRefreshedAt map[uint]time.Time

Jeromy's avatar
Jeromy committed
48 49 50
	// notification functions
	PeerRemoved func(peer.ID)
	PeerAdded   func(peer.ID)
51

Aarsh Shah's avatar
Aarsh Shah committed
52
	// usefulnessGracePeriod is the maximum grace period we will give to a
53 54 55
	// peer in the bucket to be useful to us, failing which, we will evict
	// it to make place for a new peer if the bucket is full
	usefulnessGracePeriod time.Duration
56 57
}

Chas Leichner's avatar
Chas Leichner committed
58
// NewRoutingTable creates a new routing table with a given bucketsize, local ID, and latency tolerance.
59
func NewRoutingTable(bucketsize int, localID ID, latency time.Duration, m peerstore.Metrics, usefulnessGracePeriod time.Duration) (*RoutingTable, error) {
Jeromy's avatar
Jeromy committed
60
	rt := &RoutingTable{
Aarsh Shah's avatar
Aarsh Shah committed
61
		buckets:    []*bucket{newBucket()},
62 63 64 65 66 67
		bucketsize: bucketsize,
		local:      localID,

		maxLatency: latency,
		metrics:    m,

Aarsh Shah's avatar
Aarsh Shah committed
68
		cplRefreshedAt: make(map[uint]time.Time),
69 70 71 72

		PeerRemoved: func(peer.ID) {},
		PeerAdded:   func(peer.ID) {},

Aarsh Shah's avatar
Aarsh Shah committed
73
		usefulnessGracePeriod: usefulnessGracePeriod,
Jeromy's avatar
Jeromy committed
74 75
	}

76
	rt.ctx, rt.ctxCancel = context.WithCancel(context.Background())
77 78 79 80

	return rt, nil
}

Aarsh Shah's avatar
Aarsh Shah committed
81 82 83
// Close shuts down the Routing Table & all associated processes.
// It is safe to call this multiple times.
func (rt *RoutingTable) Close() error {
84 85
	rt.ctxCancel()
	return nil
Aarsh Shah's avatar
Aarsh Shah committed
86 87
}

88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113
// NPeersForCPL returns the number of peers we have for a given Cpl
func (rt *RoutingTable) NPeersForCpl(cpl uint) int {
	rt.tabLock.RLock()
	defer rt.tabLock.RUnlock()

	// it's in the last bucket
	if int(cpl) >= len(rt.buckets)-1 {
		count := 0
		b := rt.buckets[len(rt.buckets)-1]
		for _, p := range b.peers() {
			if CommonPrefixLen(rt.local, p.dhtId) == int(cpl) {
				count++
			}
		}
		return count
	} else {
		return rt.buckets[cpl].len()
	}
}

// TryAddPeer tries to add a peer to the Routing table.
// If the peer ALREADY exists in the Routing Table and has been queried before, this call is a no-op.
// If the peer ALREADY exists in the Routing Table but hasn't been queried before, we set it's LastUsefulAt value to
// the current time. This needs to done because we don't mark peers as "Useful"(by setting the LastUsefulAt value)
// when we first connect to them.
//
Aarsh Shah's avatar
Aarsh Shah committed
114
// If the peer is a queryPeer i.e. we queried it or it queried us, we set the LastSuccessfulOutboundQuery to the current time.
115
// If the peer is just a peer that we connect to/it connected to us without any DHT query, we consider it as having
Aarsh Shah's avatar
Aarsh Shah committed
116
// no LastSuccessfulOutboundQuery.
117
//
118
//
119
// If the logical bucket to which the peer belongs is full and it's not the last bucket, we try to replace an existing peer
Aarsh Shah's avatar
Aarsh Shah committed
120
// whose LastSuccessfulOutboundQuery is above the maximum allowed threshold in that bucket with the new peer.
121 122 123 124 125 126 127 128
// If no such peer exists in that bucket, we do NOT add the peer to the Routing Table and return error "ErrPeerRejectedNoCapacity".

// It returns a boolean value set to true if the peer was newly added to the Routing Table, false otherwise.
// It also returns any error that occurred while adding the peer to the Routing Table. If the error is not nil,
// the boolean value will ALWAYS be false i.e. the peer wont be added to the Routing Table it it's not already there.
//
// A return value of false with error=nil indicates that the peer ALREADY exists in the Routing Table.
func (rt *RoutingTable) TryAddPeer(p peer.ID, queryPeer bool) (bool, error) {
129 130
	rt.tabLock.Lock()
	defer rt.tabLock.Unlock()
131

132
	return rt.addPeer(p, queryPeer)
133 134 135
}

// locking is the responsibility of the caller
136
func (rt *RoutingTable) addPeer(p peer.ID, queryPeer bool) (bool, error) {
137
	bucketID := rt.bucketIdForPeer(p)
Aarsh Shah's avatar
Aarsh Shah committed
138
	bucket := rt.buckets[bucketID]
139 140

	now := time.Now()
Aarsh Shah's avatar
Aarsh Shah committed
141
	var lastUsefulAt time.Time
142
	if queryPeer {
143
		lastUsefulAt = now
144
	}
145

146 147
	// peer already exists in the Routing Table.
	if peer := bucket.getPeer(p); peer != nil {
148 149 150 151 152
		// if we're querying the peer first time after adding it, let's give it a
		// usefulness bump. This will ONLY happen once.
		if peer.LastUsefulAt.IsZero() && queryPeer {
			peer.LastUsefulAt = lastUsefulAt
		}
153
		return false, nil
154 155
	}

156
	// peer's latency threshold is NOT acceptable
157 158
	if rt.metrics.LatencyEWMA(p) > rt.maxLatency {
		// Connection doesnt meet requirements, skip!
159
		return false, ErrPeerRejectedHighLatency
160 161 162
	}

	// We have enough space in the bucket (whether spawned or grouped).
163
	if bucket.len() < rt.bucketsize {
164 165 166 167 168 169 170
		bucket.pushFront(&PeerInfo{
			Id:                            p,
			LastUsefulAt:                  lastUsefulAt,
			LastSuccessfulOutboundQueryAt: now,
			AddedAt:                       now,
			dhtId:                         ConvertPeerID(p),
		})
171
		rt.PeerAdded(p)
172
		return true, nil
173 174
	}

Aarsh Shah's avatar
Aarsh Shah committed
175
	if bucketID == len(rt.buckets)-1 {
176 177 178
		// if the bucket is too large and this is the last bucket (i.e. wildcard), unfold it.
		rt.nextBucket()
		// the structure of the table has changed, so let's recheck if the peer now has a dedicated bucket.
179
		bucketID = rt.bucketIdForPeer(p)
Aarsh Shah's avatar
Aarsh Shah committed
180
		bucket = rt.buckets[bucketID]
181 182 183

		// push the peer only if the bucket isn't overflowing after slitting
		if bucket.len() < rt.bucketsize {
184 185 186 187 188 189 190
			bucket.pushFront(&PeerInfo{
				Id:                            p,
				LastUsefulAt:                  lastUsefulAt,
				LastSuccessfulOutboundQueryAt: now,
				AddedAt:                       now,
				dhtId:                         ConvertPeerID(p),
			})
191
			rt.PeerAdded(p)
192 193 194 195 196
			return true, nil
		}
	}

	// the bucket to which the peer belongs is full. Let's try to find a peer
Aarsh Shah's avatar
Aarsh Shah committed
197
	// in that bucket with a LastSuccessfulOutboundQuery value above the maximum threshold and replace it.
Aarsh Shah's avatar
Aarsh Shah committed
198 199 200 201 202 203
	minLast := bucket.min(func(first *PeerInfo, second *PeerInfo) bool {
		return first.LastUsefulAt.Before(second.LastUsefulAt)
	})

	if time.Since(minLast.LastUsefulAt) > rt.usefulnessGracePeriod {
		// let's evict it and add the new peer
Aarsh Shah's avatar
Aarsh Shah committed
204
		if rt.removePeer(minLast.Id) {
205 206 207 208 209 210 211
			bucket.pushFront(&PeerInfo{
				Id:                            p,
				LastUsefulAt:                  lastUsefulAt,
				LastSuccessfulOutboundQueryAt: now,
				AddedAt:                       now,
				dhtId:                         ConvertPeerID(p),
			})
Aarsh Shah's avatar
Aarsh Shah committed
212 213
			rt.PeerAdded(p)
			return true, nil
214 215
		}
	}
216

217 218 219
	return false, ErrPeerRejectedNoCapacity
}

Aarsh Shah's avatar
Aarsh Shah committed
220 221 222 223 224 225 226 227 228 229 230 231 232 233
// GetPeerInfos returns the peer information that we've stored in the buckets
func (rt *RoutingTable) GetPeerInfos() []PeerInfo {
	rt.tabLock.RLock()
	defer rt.tabLock.RUnlock()

	var pis []PeerInfo
	for _, b := range rt.buckets {
		for _, p := range b.peers() {
			pis = append(pis, p)
		}
	}
	return pis
}

Aarsh Shah's avatar
Aarsh Shah committed
234
// UpdateLastSuccessfulOutboundQuery updates the LastSuccessfulOutboundQueryAt time of the peer.
235
// Returns true if the update was successful, false otherwise.
Aarsh Shah's avatar
Aarsh Shah committed
236
func (rt *RoutingTable) UpdateLastSuccessfulOutboundQueryAt(p peer.ID, t time.Time) bool {
237 238 239 240 241 242 243
	rt.tabLock.Lock()
	defer rt.tabLock.Unlock()

	bucketID := rt.bucketIdForPeer(p)
	bucket := rt.buckets[bucketID]

	if pc := bucket.getPeer(p); pc != nil {
Aarsh Shah's avatar
Aarsh Shah committed
244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260
		pc.LastSuccessfulOutboundQueryAt = t
		return true
	}
	return false
}

// UpdateLastUsefulAt updates the LastUsefulAt time of the peer.
// Returns true if the update was successful, false otherwise.
func (rt *RoutingTable) UpdateLastUsefulAt(p peer.ID, t time.Time) bool {
	rt.tabLock.Lock()
	defer rt.tabLock.Unlock()

	bucketID := rt.bucketIdForPeer(p)
	bucket := rt.buckets[bucketID]

	if pc := bucket.getPeer(p); pc != nil {
		pc.LastUsefulAt = t
261 262 263
		return true
	}
	return false
264 265
}

266 267 268 269
// RemovePeer should be called when the caller is sure that a peer is not useful for queries.
// For eg: the peer could have stopped supporting the DHT protocol.
// It evicts the peer from the Routing Table.
func (rt *RoutingTable) RemovePeer(p peer.ID) {
Steven Allen's avatar
Steven Allen committed
270 271
	rt.tabLock.Lock()
	defer rt.tabLock.Unlock()
272 273 274 275
	rt.removePeer(p)
}

// locking is the responsibility of the caller
Aarsh Shah's avatar
Aarsh Shah committed
276
func (rt *RoutingTable) removePeer(p peer.ID) bool {
277
	bucketID := rt.bucketIdForPeer(p)
Aarsh Shah's avatar
Aarsh Shah committed
278
	bucket := rt.buckets[bucketID]
279
	if bucket.remove(p) {
280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296
		for {
			lastBucketIndex := len(rt.buckets) - 1

			// remove the last bucket if it's empty and it isn't the only bucket we have
			if len(rt.buckets) > 1 && rt.buckets[lastBucketIndex].len() == 0 {
				rt.buckets[lastBucketIndex] = nil
				rt.buckets = rt.buckets[:lastBucketIndex]
			} else if len(rt.buckets) >= 2 && rt.buckets[lastBucketIndex-1].len() == 0 {
				// if the second last bucket just became empty, remove and replace it with the last bucket.
				rt.buckets[lastBucketIndex-1] = rt.buckets[lastBucketIndex]
				rt.buckets[lastBucketIndex] = nil
				rt.buckets = rt.buckets[:lastBucketIndex]
			} else {
				break
			}
		}

297
		// peer removed callback
298
		rt.PeerRemoved(p)
Aarsh Shah's avatar
Aarsh Shah committed
299
		return true
300
	}
Aarsh Shah's avatar
Aarsh Shah committed
301
	return false
302 303
}

304
func (rt *RoutingTable) nextBucket() {
305 306 307
	// This is the last bucket, which allegedly is a mixed bag containing peers not belonging in dedicated (unfolded) buckets.
	// _allegedly_ is used here to denote that *all* peers in the last bucket might feasibly belong to another bucket.
	// This could happen if e.g. we've unfolded 4 buckets, and all peers in folded bucket 5 really belong in bucket 8.
Aarsh Shah's avatar
Aarsh Shah committed
308 309 310
	bucket := rt.buckets[len(rt.buckets)-1]
	newBucket := bucket.split(len(rt.buckets)-1, rt.local)
	rt.buckets = append(rt.buckets, newBucket)
311

312
	// The newly formed bucket still contains too many peers. We probably just unfolded a empty bucket.
313
	if newBucket.len() >= rt.bucketsize {
314 315
		// Keep unfolding the table until the last bucket is not overflowing.
		rt.nextBucket()
316 317 318
	}
}

Jeromy's avatar
Jeromy committed
319
// Find a specific peer by ID or return nil
320
func (rt *RoutingTable) Find(id peer.ID) peer.ID {
Chas Leichner's avatar
Chas Leichner committed
321
	srch := rt.NearestPeers(ConvertPeerID(id), 1)
322 323
	if len(srch) == 0 || srch[0] != id {
		return ""
Jeromy's avatar
Jeromy committed
324 325 326 327
	}
	return srch[0]
}

328
// NearestPeer returns a single peer that is nearest to the given ID
329
func (rt *RoutingTable) NearestPeer(id ID) peer.ID {
330 331 332 333
	peers := rt.NearestPeers(id, 1)
	if len(peers) > 0 {
		return peers[0]
	}
334

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
335
	log.Debugf("NearestPeer: Returning nil, table size = %d", rt.Size())
336
	return ""
337 338
}

339
// NearestPeers returns a list of the 'count' closest peers to the given ID
340
func (rt *RoutingTable) NearestPeers(id ID, count int) []peer.ID {
341 342 343 344
	// This is the number of bits _we_ share with the key. All peers in this
	// bucket share cpl bits with us and will therefore share at least cpl+1
	// bits with the given key. +1 because both the target and all peers in
	// this bucket differ from us in the cpl bit.
Matt Joiner's avatar
Matt Joiner committed
345
	cpl := CommonPrefixLen(id, rt.local)
346

347
	// It's assumed that this also protects the buckets.
Jeromy's avatar
Jeromy committed
348 349
	rt.tabLock.RLock()

Aarsh Shah's avatar
Aarsh Shah committed
350 351 352
	// Get bucket index or last bucket
	if cpl >= len(rt.buckets) {
		cpl = len(rt.buckets) - 1
353 354
	}

355
	pds := peerDistanceSorter{
356
		peers:  make([]peerDistance, 0, count+rt.bucketsize),
357 358
		target: id,
	}
359

360
	// Add peers from the target bucket (cpl+1 shared bits).
Aarsh Shah's avatar
Aarsh Shah committed
361
	pds.appendPeersFromList(rt.buckets[cpl].list)
362

Steven Allen's avatar
Steven Allen committed
363 364 365
	// If we're short, add peers from all buckets to the right. All buckets
	// to the right share exactly cpl bits (as opposed to the cpl+1 bits
	// shared by the peers in the cpl bucket).
366
	//
Steven Allen's avatar
Steven Allen committed
367 368 369
	// This is, unfortunately, less efficient than we'd like. We will switch
	// to a trie implementation eventually which will allow us to find the
	// closest N peers to any target key.
370

Steven Allen's avatar
Steven Allen committed
371 372 373 374
	if pds.Len() < count {
		for i := cpl + 1; i < len(rt.buckets); i++ {
			pds.appendPeersFromList(rt.buckets[i].list)
		}
375 376
	}

377 378 379 380
	// If we're still short, add in buckets that share _fewer_ bits. We can
	// do this bucket by bucket because each bucket will share 1 fewer bit
	// than the last.
	//
Aarsh Shah's avatar
fix doc  
Aarsh Shah committed
381 382
	// * bucket cpl-1: cpl-1 shared bits.
	// * bucket cpl-2: cpl-2 shared bits.
383
	// ...
384
	for i := cpl - 1; i >= 0 && pds.Len() < count; i-- {
Aarsh Shah's avatar
Aarsh Shah committed
385
		pds.appendPeersFromList(rt.buckets[i].list)
386
	}
Jeromy's avatar
Jeromy committed
387
	rt.tabLock.RUnlock()
388 389

	// Sort by distance to local peer
390
	pds.sort()
391

392 393
	if count < pds.Len() {
		pds.peers = pds.peers[:count]
394 395
	}

396 397
	out := make([]peer.ID, 0, pds.Len())
	for _, p := range pds.peers {
398
		out = append(out, p.p)
399 400 401 402 403
	}

	return out
}

404
// Size returns the total number of peers in the routing table
405 406
func (rt *RoutingTable) Size() int {
	var tot int
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
407
	rt.tabLock.RLock()
Aarsh Shah's avatar
Aarsh Shah committed
408
	for _, buck := range rt.buckets {
409
		tot += buck.len()
410
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
411
	rt.tabLock.RUnlock()
412 413 414
	return tot
}

Chas Leichner's avatar
Chas Leichner committed
415
// ListPeers takes a RoutingTable and returns a list of all peers from all buckets in the table.
416
func (rt *RoutingTable) ListPeers() []peer.ID {
417
	rt.tabLock.RLock()
418 419 420
	defer rt.tabLock.RUnlock()

	var peers []peer.ID
Aarsh Shah's avatar
Aarsh Shah committed
421
	for _, buck := range rt.buckets {
422
		peers = append(peers, buck.peerIds()...)
423 424 425
	}
	return peers
}
426

Chas Leichner's avatar
Chas Leichner committed
427 428
// Print prints a descriptive statement about the provided RoutingTable
func (rt *RoutingTable) Print() {
429 430
	fmt.Printf("Routing Table, bs = %d, Max latency = %d\n", rt.bucketsize, rt.maxLatency)
	rt.tabLock.RLock()
431

Aarsh Shah's avatar
Aarsh Shah committed
432
	for i, b := range rt.buckets {
433 434 435
		fmt.Printf("\tbucket: %d\n", i)

		for e := b.list.Front(); e != nil; e = e.Next() {
Aarsh Shah's avatar
Aarsh Shah committed
436
			p := e.Value.(*PeerInfo).Id
437 438
			fmt.Printf("\t\t- %s %s\n", p.Pretty(), rt.metrics.LatencyEWMA(p).String())
		}
439
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
440
	rt.tabLock.RUnlock()
441
}
442 443 444 445 446 447

// the caller is responsible for the locking
func (rt *RoutingTable) bucketIdForPeer(p peer.ID) int {
	peerID := ConvertPeerID(p)
	cpl := CommonPrefixLen(peerID, rt.local)
	bucketID := cpl
Aarsh Shah's avatar
Aarsh Shah committed
448 449
	if bucketID >= len(rt.buckets) {
		bucketID = len(rt.buckets) - 1
450 451 452
	}
	return bucketID
}
453 454 455 456 457 458 459 460 461 462 463 464 465 466

// maxCommonPrefix returns the maximum common prefix length between any peer in
// the table and the current peer.
func (rt *RoutingTable) maxCommonPrefix() uint {
	rt.tabLock.RLock()
	defer rt.tabLock.RUnlock()

	for i := len(rt.buckets) - 1; i >= 0; i-- {
		if rt.buckets[i].len() > 0 {
			return rt.buckets[i].maxCommonPrefix(rt.local)
		}
	}
	return 0
}