forked from cerc-io/plugeth
9123eceb0f
* p2p: add DialRatio for configuration of inbound vs. dialed connections * p2p: add connection flags to PeerInfo * p2p/netutil: add SameNet, DistinctNetSet * p2p/discover: improve revalidation and seeding This changes node revalidation to be periodic instead of on-demand. This should prevent issues where dead nodes get stuck in closer buckets because no other node will ever come along to replace them. Every 5 seconds (on average), the last node in a random bucket is checked and moved to the front of the bucket if it is still responding. If revalidation fails, the last node is replaced by an entry of the 'replacement list' containing recently-seen nodes. Most close buckets are removed because it's very unlikely we'll ever encounter a node that would fall into any of those buckets. Table seeding is also improved: we now require a few minutes of table membership before considering a node as a potential seed node. This should make it less likely to store short-lived nodes as potential seeds. * p2p/discover: fix nits in UDP transport We would skip sending neighbors replies if there were fewer than maxNeighbors results and CheckRelayIP returned an error for the last one. While here, also resolve a TODO about pong reply tokens.
865 lines
24 KiB
Go
865 lines
24 KiB
Go
// Copyright 2015 The go-ethereum Authors
|
|
// This file is part of the go-ethereum library.
|
|
//
|
|
// The go-ethereum library is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Lesser General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// The go-ethereum library is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Lesser General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Lesser General Public License
|
|
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
// Package discover implements the Node Discovery Protocol.
|
|
//
|
|
// The Node Discovery protocol provides a way to find RLPx nodes that
|
|
// can be connected to. It uses a Kademlia-like protocol to maintain a
|
|
// distributed database of the IDs and endpoints of all listening
|
|
// nodes.
|
|
package discover
|
|
|
|
import (
|
|
crand "crypto/rand"
|
|
"encoding/binary"
|
|
"errors"
|
|
"fmt"
|
|
mrand "math/rand"
|
|
"net"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/crypto"
|
|
"github.com/ethereum/go-ethereum/log"
|
|
"github.com/ethereum/go-ethereum/p2p/netutil"
|
|
)
|
|
|
|
const (
|
|
alpha = 3 // Kademlia concurrency factor
|
|
bucketSize = 16 // Kademlia bucket size
|
|
maxReplacements = 10 // Size of per-bucket replacement list
|
|
|
|
// We keep buckets for the upper 1/15 of distances because
|
|
// it's very unlikely we'll ever encounter a node that's closer.
|
|
hashBits = len(common.Hash{}) * 8
|
|
nBuckets = hashBits / 15 // Number of buckets
|
|
bucketMinDistance = hashBits - nBuckets // Log distance of closest bucket
|
|
|
|
// IP address limits.
|
|
bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24
|
|
tableIPLimit, tableSubnet = 10, 24
|
|
|
|
maxBondingPingPongs = 16 // Limit on the number of concurrent ping/pong interactions
|
|
maxFindnodeFailures = 5 // Nodes exceeding this limit are dropped
|
|
|
|
refreshInterval = 30 * time.Minute
|
|
revalidateInterval = 10 * time.Second
|
|
copyNodesInterval = 30 * time.Second
|
|
seedMinTableTime = 5 * time.Minute
|
|
seedCount = 30
|
|
seedMaxAge = 5 * 24 * time.Hour
|
|
)
|
|
|
|
type Table struct {
|
|
mutex sync.Mutex // protects buckets, bucket content, nursery, rand
|
|
buckets [nBuckets]*bucket // index of known nodes by distance
|
|
nursery []*Node // bootstrap nodes
|
|
rand *mrand.Rand // source of randomness, periodically reseeded
|
|
ips netutil.DistinctNetSet
|
|
|
|
db *nodeDB // database of known nodes
|
|
refreshReq chan chan struct{}
|
|
initDone chan struct{}
|
|
closeReq chan struct{}
|
|
closed chan struct{}
|
|
|
|
bondmu sync.Mutex
|
|
bonding map[NodeID]*bondproc
|
|
bondslots chan struct{} // limits total number of active bonding processes
|
|
|
|
nodeAddedHook func(*Node) // for testing
|
|
|
|
net transport
|
|
self *Node // metadata of the local node
|
|
}
|
|
|
|
type bondproc struct {
|
|
err error
|
|
n *Node
|
|
done chan struct{}
|
|
}
|
|
|
|
// transport is implemented by the UDP transport.
|
|
// it is an interface so we can test without opening lots of UDP
|
|
// sockets and without generating a private key.
|
|
type transport interface {
|
|
ping(NodeID, *net.UDPAddr) error
|
|
waitping(NodeID) error
|
|
findnode(toid NodeID, addr *net.UDPAddr, target NodeID) ([]*Node, error)
|
|
close()
|
|
}
|
|
|
|
// bucket contains nodes, ordered by their last activity. the entry
|
|
// that was most recently active is the first element in entries.
|
|
type bucket struct {
|
|
entries []*Node // live entries, sorted by time of last contact
|
|
replacements []*Node // recently seen nodes to be used if revalidation fails
|
|
ips netutil.DistinctNetSet
|
|
}
|
|
|
|
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, nodeDBPath string, bootnodes []*Node) (*Table, error) {
|
|
// If no node database was given, use an in-memory one
|
|
db, err := newNodeDB(nodeDBPath, Version, ourID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tab := &Table{
|
|
net: t,
|
|
db: db,
|
|
self: NewNode(ourID, ourAddr.IP, uint16(ourAddr.Port), uint16(ourAddr.Port)),
|
|
bonding: make(map[NodeID]*bondproc),
|
|
bondslots: make(chan struct{}, maxBondingPingPongs),
|
|
refreshReq: make(chan chan struct{}),
|
|
initDone: make(chan struct{}),
|
|
closeReq: make(chan struct{}),
|
|
closed: make(chan struct{}),
|
|
rand: mrand.New(mrand.NewSource(0)),
|
|
ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit},
|
|
}
|
|
if err := tab.setFallbackNodes(bootnodes); err != nil {
|
|
return nil, err
|
|
}
|
|
for i := 0; i < cap(tab.bondslots); i++ {
|
|
tab.bondslots <- struct{}{}
|
|
}
|
|
for i := range tab.buckets {
|
|
tab.buckets[i] = &bucket{
|
|
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
|
|
}
|
|
}
|
|
tab.seedRand()
|
|
tab.loadSeedNodes(false)
|
|
// Start the background expiration goroutine after loading seeds so that the search for
|
|
// seed nodes also considers older nodes that would otherwise be removed by the
|
|
// expiration.
|
|
tab.db.ensureExpirer()
|
|
go tab.loop()
|
|
return tab, nil
|
|
}
|
|
|
|
func (tab *Table) seedRand() {
|
|
var b [8]byte
|
|
crand.Read(b[:])
|
|
|
|
tab.mutex.Lock()
|
|
tab.rand.Seed(int64(binary.BigEndian.Uint64(b[:])))
|
|
tab.mutex.Unlock()
|
|
}
|
|
|
|
// Self returns the local node.
|
|
// The returned node should not be modified by the caller.
|
|
func (tab *Table) Self() *Node {
|
|
return tab.self
|
|
}
|
|
|
|
// ReadRandomNodes fills the given slice with random nodes from the
|
|
// table. It will not write the same node more than once. The nodes in
|
|
// the slice are copies and can be modified by the caller.
|
|
func (tab *Table) ReadRandomNodes(buf []*Node) (n int) {
|
|
if !tab.isInitDone() {
|
|
return 0
|
|
}
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
// Find all non-empty buckets and get a fresh slice of their entries.
|
|
var buckets [][]*Node
|
|
for _, b := range tab.buckets {
|
|
if len(b.entries) > 0 {
|
|
buckets = append(buckets, b.entries[:])
|
|
}
|
|
}
|
|
if len(buckets) == 0 {
|
|
return 0
|
|
}
|
|
// Shuffle the buckets.
|
|
for i := len(buckets) - 1; i > 0; i-- {
|
|
j := tab.rand.Intn(len(buckets))
|
|
buckets[i], buckets[j] = buckets[j], buckets[i]
|
|
}
|
|
// Move head of each bucket into buf, removing buckets that become empty.
|
|
var i, j int
|
|
for ; i < len(buf); i, j = i+1, (j+1)%len(buckets) {
|
|
b := buckets[j]
|
|
buf[i] = &(*b[0])
|
|
buckets[j] = b[1:]
|
|
if len(b) == 1 {
|
|
buckets = append(buckets[:j], buckets[j+1:]...)
|
|
}
|
|
if len(buckets) == 0 {
|
|
break
|
|
}
|
|
}
|
|
return i + 1
|
|
}
|
|
|
|
// Close terminates the network listener and flushes the node database.
|
|
func (tab *Table) Close() {
|
|
select {
|
|
case <-tab.closed:
|
|
// already closed.
|
|
case tab.closeReq <- struct{}{}:
|
|
<-tab.closed // wait for refreshLoop to end.
|
|
}
|
|
}
|
|
|
|
// setFallbackNodes sets the initial points of contact. These nodes
|
|
// are used to connect to the network if the table is empty and there
|
|
// are no known nodes in the database.
|
|
func (tab *Table) setFallbackNodes(nodes []*Node) error {
|
|
for _, n := range nodes {
|
|
if err := n.validateComplete(); err != nil {
|
|
return fmt.Errorf("bad bootstrap/fallback node %q (%v)", n, err)
|
|
}
|
|
}
|
|
tab.nursery = make([]*Node, 0, len(nodes))
|
|
for _, n := range nodes {
|
|
cpy := *n
|
|
// Recompute cpy.sha because the node might not have been
|
|
// created by NewNode or ParseNode.
|
|
cpy.sha = crypto.Keccak256Hash(n.ID[:])
|
|
tab.nursery = append(tab.nursery, &cpy)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// isInitDone returns whether the table's initial seeding procedure has completed.
|
|
func (tab *Table) isInitDone() bool {
|
|
select {
|
|
case <-tab.initDone:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// Resolve searches for a specific node with the given ID.
|
|
// It returns nil if the node could not be found.
|
|
func (tab *Table) Resolve(targetID NodeID) *Node {
|
|
// If the node is present in the local table, no
|
|
// network interaction is required.
|
|
hash := crypto.Keccak256Hash(targetID[:])
|
|
tab.mutex.Lock()
|
|
cl := tab.closest(hash, 1)
|
|
tab.mutex.Unlock()
|
|
if len(cl.entries) > 0 && cl.entries[0].ID == targetID {
|
|
return cl.entries[0]
|
|
}
|
|
// Otherwise, do a network lookup.
|
|
result := tab.Lookup(targetID)
|
|
for _, n := range result {
|
|
if n.ID == targetID {
|
|
return n
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Lookup performs a network search for nodes close
|
|
// to the given target. It approaches the target by querying
|
|
// nodes that are closer to it on each iteration.
|
|
// The given target does not need to be an actual node
|
|
// identifier.
|
|
func (tab *Table) Lookup(targetID NodeID) []*Node {
|
|
return tab.lookup(targetID, true)
|
|
}
|
|
|
|
func (tab *Table) lookup(targetID NodeID, refreshIfEmpty bool) []*Node {
|
|
var (
|
|
target = crypto.Keccak256Hash(targetID[:])
|
|
asked = make(map[NodeID]bool)
|
|
seen = make(map[NodeID]bool)
|
|
reply = make(chan []*Node, alpha)
|
|
pendingQueries = 0
|
|
result *nodesByDistance
|
|
)
|
|
// don't query further if we hit ourself.
|
|
// unlikely to happen often in practice.
|
|
asked[tab.self.ID] = true
|
|
|
|
for {
|
|
tab.mutex.Lock()
|
|
// generate initial result set
|
|
result = tab.closest(target, bucketSize)
|
|
tab.mutex.Unlock()
|
|
if len(result.entries) > 0 || !refreshIfEmpty {
|
|
break
|
|
}
|
|
// The result set is empty, all nodes were dropped, refresh.
|
|
// We actually wait for the refresh to complete here. The very
|
|
// first query will hit this case and run the bootstrapping
|
|
// logic.
|
|
<-tab.refresh()
|
|
refreshIfEmpty = false
|
|
}
|
|
|
|
for {
|
|
// ask the alpha closest nodes that we haven't asked yet
|
|
for i := 0; i < len(result.entries) && pendingQueries < alpha; i++ {
|
|
n := result.entries[i]
|
|
if !asked[n.ID] {
|
|
asked[n.ID] = true
|
|
pendingQueries++
|
|
go func() {
|
|
// Find potential neighbors to bond with
|
|
r, err := tab.net.findnode(n.ID, n.addr(), targetID)
|
|
if err != nil {
|
|
// Bump the failure counter to detect and evacuate non-bonded entries
|
|
fails := tab.db.findFails(n.ID) + 1
|
|
tab.db.updateFindFails(n.ID, fails)
|
|
log.Trace("Bumping findnode failure counter", "id", n.ID, "failcount", fails)
|
|
|
|
if fails >= maxFindnodeFailures {
|
|
log.Trace("Too many findnode failures, dropping", "id", n.ID, "failcount", fails)
|
|
tab.delete(n)
|
|
}
|
|
}
|
|
reply <- tab.bondall(r)
|
|
}()
|
|
}
|
|
}
|
|
if pendingQueries == 0 {
|
|
// we have asked all closest nodes, stop the search
|
|
break
|
|
}
|
|
// wait for the next reply
|
|
for _, n := range <-reply {
|
|
if n != nil && !seen[n.ID] {
|
|
seen[n.ID] = true
|
|
result.push(n, bucketSize)
|
|
}
|
|
}
|
|
pendingQueries--
|
|
}
|
|
return result.entries
|
|
}
|
|
|
|
func (tab *Table) refresh() <-chan struct{} {
|
|
done := make(chan struct{})
|
|
select {
|
|
case tab.refreshReq <- done:
|
|
case <-tab.closed:
|
|
close(done)
|
|
}
|
|
return done
|
|
}
|
|
|
|
// loop schedules refresh, revalidate runs and coordinates shutdown.
|
|
func (tab *Table) loop() {
|
|
var (
|
|
revalidate = time.NewTimer(tab.nextRevalidateTime())
|
|
refresh = time.NewTicker(refreshInterval)
|
|
copyNodes = time.NewTicker(copyNodesInterval)
|
|
revalidateDone = make(chan struct{})
|
|
refreshDone = make(chan struct{}) // where doRefresh reports completion
|
|
waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
|
|
)
|
|
defer refresh.Stop()
|
|
defer revalidate.Stop()
|
|
defer copyNodes.Stop()
|
|
|
|
// Start initial refresh.
|
|
go tab.doRefresh(refreshDone)
|
|
|
|
loop:
|
|
for {
|
|
select {
|
|
case <-refresh.C:
|
|
tab.seedRand()
|
|
if refreshDone == nil {
|
|
refreshDone = make(chan struct{})
|
|
go tab.doRefresh(refreshDone)
|
|
}
|
|
case req := <-tab.refreshReq:
|
|
waiting = append(waiting, req)
|
|
if refreshDone == nil {
|
|
refreshDone = make(chan struct{})
|
|
go tab.doRefresh(refreshDone)
|
|
}
|
|
case <-refreshDone:
|
|
for _, ch := range waiting {
|
|
close(ch)
|
|
}
|
|
waiting, refreshDone = nil, nil
|
|
case <-revalidate.C:
|
|
go tab.doRevalidate(revalidateDone)
|
|
case <-revalidateDone:
|
|
revalidate.Reset(tab.nextRevalidateTime())
|
|
case <-copyNodes.C:
|
|
go tab.copyBondedNodes()
|
|
case <-tab.closeReq:
|
|
break loop
|
|
}
|
|
}
|
|
|
|
if tab.net != nil {
|
|
tab.net.close()
|
|
}
|
|
if refreshDone != nil {
|
|
<-refreshDone
|
|
}
|
|
for _, ch := range waiting {
|
|
close(ch)
|
|
}
|
|
tab.db.close()
|
|
close(tab.closed)
|
|
}
|
|
|
|
// doRefresh performs a lookup for a random target to keep buckets
|
|
// full. seed nodes are inserted if the table is empty (initial
|
|
// bootstrap or discarded faulty peers).
|
|
func (tab *Table) doRefresh(done chan struct{}) {
|
|
defer close(done)
|
|
|
|
// Load nodes from the database and insert
|
|
// them. This should yield a few previously seen nodes that are
|
|
// (hopefully) still alive.
|
|
tab.loadSeedNodes(true)
|
|
|
|
// Run self lookup to discover new neighbor nodes.
|
|
tab.lookup(tab.self.ID, false)
|
|
|
|
// The Kademlia paper specifies that the bucket refresh should
|
|
// perform a lookup in the least recently used bucket. We cannot
|
|
// adhere to this because the findnode target is a 512bit value
|
|
// (not hash-sized) and it is not easily possible to generate a
|
|
// sha3 preimage that falls into a chosen bucket.
|
|
// We perform a few lookups with a random target instead.
|
|
for i := 0; i < 3; i++ {
|
|
var target NodeID
|
|
crand.Read(target[:])
|
|
tab.lookup(target, false)
|
|
}
|
|
}
|
|
|
|
func (tab *Table) loadSeedNodes(bond bool) {
|
|
seeds := tab.db.querySeeds(seedCount, seedMaxAge)
|
|
seeds = append(seeds, tab.nursery...)
|
|
if bond {
|
|
seeds = tab.bondall(seeds)
|
|
}
|
|
for i := range seeds {
|
|
seed := seeds[i]
|
|
age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.lastPong(seed.ID)) }}
|
|
log.Debug("Found seed node in database", "id", seed.ID, "addr", seed.addr(), "age", age)
|
|
tab.add(seed)
|
|
}
|
|
}
|
|
|
|
// doRevalidate checks that the last node in a random bucket is still live
|
|
// and replaces or deletes the node if it isn't.
|
|
func (tab *Table) doRevalidate(done chan<- struct{}) {
|
|
defer func() { done <- struct{}{} }()
|
|
|
|
last, bi := tab.nodeToRevalidate()
|
|
if last == nil {
|
|
// No non-empty bucket found.
|
|
return
|
|
}
|
|
|
|
// Ping the selected node and wait for a pong.
|
|
err := tab.ping(last.ID, last.addr())
|
|
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
b := tab.buckets[bi]
|
|
if err == nil {
|
|
// The node responded, move it to the front.
|
|
log.Debug("Revalidated node", "b", bi, "id", last.ID)
|
|
b.bump(last)
|
|
return
|
|
}
|
|
// No reply received, pick a replacement or delete the node if there aren't
|
|
// any replacements.
|
|
if r := tab.replace(b, last); r != nil {
|
|
log.Debug("Replaced dead node", "b", bi, "id", last.ID, "ip", last.IP, "r", r.ID, "rip", r.IP)
|
|
} else {
|
|
log.Debug("Removed dead node", "b", bi, "id", last.ID, "ip", last.IP)
|
|
}
|
|
}
|
|
|
|
// nodeToRevalidate returns the last node in a random, non-empty bucket.
|
|
func (tab *Table) nodeToRevalidate() (n *Node, bi int) {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
for _, bi = range tab.rand.Perm(len(tab.buckets)) {
|
|
b := tab.buckets[bi]
|
|
if len(b.entries) > 0 {
|
|
last := b.entries[len(b.entries)-1]
|
|
return last, bi
|
|
}
|
|
}
|
|
return nil, 0
|
|
}
|
|
|
|
func (tab *Table) nextRevalidateTime() time.Duration {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
return time.Duration(tab.rand.Int63n(int64(revalidateInterval)))
|
|
}
|
|
|
|
// copyBondedNodes adds nodes from the table to the database if they have been in the table
|
|
// longer then minTableTime.
|
|
func (tab *Table) copyBondedNodes() {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
now := time.Now()
|
|
for _, b := range tab.buckets {
|
|
for _, n := range b.entries {
|
|
if now.Sub(n.addedAt) >= seedMinTableTime {
|
|
tab.db.updateNode(n)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// closest returns the n nodes in the table that are closest to the
|
|
// given id. The caller must hold tab.mutex.
|
|
func (tab *Table) closest(target common.Hash, nresults int) *nodesByDistance {
|
|
// This is a very wasteful way to find the closest nodes but
|
|
// obviously correct. I believe that tree-based buckets would make
|
|
// this easier to implement efficiently.
|
|
close := &nodesByDistance{target: target}
|
|
for _, b := range tab.buckets {
|
|
for _, n := range b.entries {
|
|
close.push(n, nresults)
|
|
}
|
|
}
|
|
return close
|
|
}
|
|
|
|
func (tab *Table) len() (n int) {
|
|
for _, b := range tab.buckets {
|
|
n += len(b.entries)
|
|
}
|
|
return n
|
|
}
|
|
|
|
// bondall bonds with all given nodes concurrently and returns
|
|
// those nodes for which bonding has probably succeeded.
|
|
func (tab *Table) bondall(nodes []*Node) (result []*Node) {
|
|
rc := make(chan *Node, len(nodes))
|
|
for i := range nodes {
|
|
go func(n *Node) {
|
|
nn, _ := tab.bond(false, n.ID, n.addr(), n.TCP)
|
|
rc <- nn
|
|
}(nodes[i])
|
|
}
|
|
for range nodes {
|
|
if n := <-rc; n != nil {
|
|
result = append(result, n)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
// bond ensures the local node has a bond with the given remote node.
|
|
// It also attempts to insert the node into the table if bonding succeeds.
|
|
// The caller must not hold tab.mutex.
|
|
//
|
|
// A bond is must be established before sending findnode requests.
|
|
// Both sides must have completed a ping/pong exchange for a bond to
|
|
// exist. The total number of active bonding processes is limited in
|
|
// order to restrain network use.
|
|
//
|
|
// bond is meant to operate idempotently in that bonding with a remote
|
|
// node which still remembers a previously established bond will work.
|
|
// The remote node will simply not send a ping back, causing waitping
|
|
// to time out.
|
|
//
|
|
// If pinged is true, the remote node has just pinged us and one half
|
|
// of the process can be skipped.
|
|
func (tab *Table) bond(pinged bool, id NodeID, addr *net.UDPAddr, tcpPort uint16) (*Node, error) {
|
|
if id == tab.self.ID {
|
|
return nil, errors.New("is self")
|
|
}
|
|
if pinged && !tab.isInitDone() {
|
|
return nil, errors.New("still initializing")
|
|
}
|
|
// Start bonding if we haven't seen this node for a while or if it failed findnode too often.
|
|
node, fails := tab.db.node(id), tab.db.findFails(id)
|
|
age := time.Since(tab.db.lastPong(id))
|
|
var result error
|
|
if fails > 0 || age > nodeDBNodeExpiration {
|
|
log.Trace("Starting bonding ping/pong", "id", id, "known", node != nil, "failcount", fails, "age", age)
|
|
|
|
tab.bondmu.Lock()
|
|
w := tab.bonding[id]
|
|
if w != nil {
|
|
// Wait for an existing bonding process to complete.
|
|
tab.bondmu.Unlock()
|
|
<-w.done
|
|
} else {
|
|
// Register a new bonding process.
|
|
w = &bondproc{done: make(chan struct{})}
|
|
tab.bonding[id] = w
|
|
tab.bondmu.Unlock()
|
|
// Do the ping/pong. The result goes into w.
|
|
tab.pingpong(w, pinged, id, addr, tcpPort)
|
|
// Unregister the process after it's done.
|
|
tab.bondmu.Lock()
|
|
delete(tab.bonding, id)
|
|
tab.bondmu.Unlock()
|
|
}
|
|
// Retrieve the bonding results
|
|
result = w.err
|
|
if result == nil {
|
|
node = w.n
|
|
}
|
|
}
|
|
// Add the node to the table even if the bonding ping/pong
|
|
// fails. It will be relaced quickly if it continues to be
|
|
// unresponsive.
|
|
if node != nil {
|
|
tab.add(node)
|
|
tab.db.updateFindFails(id, 0)
|
|
}
|
|
return node, result
|
|
}
|
|
|
|
func (tab *Table) pingpong(w *bondproc, pinged bool, id NodeID, addr *net.UDPAddr, tcpPort uint16) {
|
|
// Request a bonding slot to limit network usage
|
|
<-tab.bondslots
|
|
defer func() { tab.bondslots <- struct{}{} }()
|
|
|
|
// Ping the remote side and wait for a pong.
|
|
if w.err = tab.ping(id, addr); w.err != nil {
|
|
close(w.done)
|
|
return
|
|
}
|
|
if !pinged {
|
|
// Give the remote node a chance to ping us before we start
|
|
// sending findnode requests. If they still remember us,
|
|
// waitping will simply time out.
|
|
tab.net.waitping(id)
|
|
}
|
|
// Bonding succeeded, update the node database.
|
|
w.n = NewNode(id, addr.IP, uint16(addr.Port), tcpPort)
|
|
close(w.done)
|
|
}
|
|
|
|
// ping a remote endpoint and wait for a reply, also updating the node
|
|
// database accordingly.
|
|
func (tab *Table) ping(id NodeID, addr *net.UDPAddr) error {
|
|
tab.db.updateLastPing(id, time.Now())
|
|
if err := tab.net.ping(id, addr); err != nil {
|
|
return err
|
|
}
|
|
tab.db.updateLastPong(id, time.Now())
|
|
return nil
|
|
}
|
|
|
|
// bucket returns the bucket for the given node ID hash.
|
|
func (tab *Table) bucket(sha common.Hash) *bucket {
|
|
d := logdist(tab.self.sha, sha)
|
|
if d <= bucketMinDistance {
|
|
return tab.buckets[0]
|
|
}
|
|
return tab.buckets[d-bucketMinDistance-1]
|
|
}
|
|
|
|
// add attempts to add the given node its corresponding bucket. If the
|
|
// bucket has space available, adding the node succeeds immediately.
|
|
// Otherwise, the node is added if the least recently active node in
|
|
// the bucket does not respond to a ping packet.
|
|
//
|
|
// The caller must not hold tab.mutex.
|
|
func (tab *Table) add(new *Node) {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
b := tab.bucket(new.sha)
|
|
if !tab.bumpOrAdd(b, new) {
|
|
// Node is not in table. Add it to the replacement list.
|
|
tab.addReplacement(b, new)
|
|
}
|
|
}
|
|
|
|
// stuff adds nodes the table to the end of their corresponding bucket
|
|
// if the bucket is not full. The caller must not hold tab.mutex.
|
|
func (tab *Table) stuff(nodes []*Node) {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
for _, n := range nodes {
|
|
if n.ID == tab.self.ID {
|
|
continue // don't add self
|
|
}
|
|
b := tab.bucket(n.sha)
|
|
if len(b.entries) < bucketSize {
|
|
tab.bumpOrAdd(b, n)
|
|
}
|
|
}
|
|
}
|
|
|
|
// delete removes an entry from the node table (used to evacuate
|
|
// failed/non-bonded discovery peers).
|
|
func (tab *Table) delete(node *Node) {
|
|
tab.mutex.Lock()
|
|
defer tab.mutex.Unlock()
|
|
|
|
tab.deleteInBucket(tab.bucket(node.sha), node)
|
|
}
|
|
|
|
func (tab *Table) addIP(b *bucket, ip net.IP) bool {
|
|
if netutil.IsLAN(ip) {
|
|
return true
|
|
}
|
|
if !tab.ips.Add(ip) {
|
|
log.Debug("IP exceeds table limit", "ip", ip)
|
|
return false
|
|
}
|
|
if !b.ips.Add(ip) {
|
|
log.Debug("IP exceeds bucket limit", "ip", ip)
|
|
tab.ips.Remove(ip)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (tab *Table) removeIP(b *bucket, ip net.IP) {
|
|
if netutil.IsLAN(ip) {
|
|
return
|
|
}
|
|
tab.ips.Remove(ip)
|
|
b.ips.Remove(ip)
|
|
}
|
|
|
|
func (tab *Table) addReplacement(b *bucket, n *Node) {
|
|
for _, e := range b.replacements {
|
|
if e.ID == n.ID {
|
|
return // already in list
|
|
}
|
|
}
|
|
if !tab.addIP(b, n.IP) {
|
|
return
|
|
}
|
|
var removed *Node
|
|
b.replacements, removed = pushNode(b.replacements, n, maxReplacements)
|
|
if removed != nil {
|
|
tab.removeIP(b, removed.IP)
|
|
}
|
|
}
|
|
|
|
// replace removes n from the replacement list and replaces 'last' with it if it is the
|
|
// last entry in the bucket. If 'last' isn't the last entry, it has either been replaced
|
|
// with someone else or became active.
|
|
func (tab *Table) replace(b *bucket, last *Node) *Node {
|
|
if len(b.entries) >= 0 && b.entries[len(b.entries)-1].ID != last.ID {
|
|
// Entry has moved, don't replace it.
|
|
return nil
|
|
}
|
|
// Still the last entry.
|
|
if len(b.replacements) == 0 {
|
|
tab.deleteInBucket(b, last)
|
|
return nil
|
|
}
|
|
r := b.replacements[tab.rand.Intn(len(b.replacements))]
|
|
b.replacements = deleteNode(b.replacements, r)
|
|
b.entries[len(b.entries)-1] = r
|
|
tab.removeIP(b, last.IP)
|
|
return r
|
|
}
|
|
|
|
// bump moves the given node to the front of the bucket entry list
|
|
// if it is contained in that list.
|
|
func (b *bucket) bump(n *Node) bool {
|
|
for i := range b.entries {
|
|
if b.entries[i].ID == n.ID {
|
|
// move it to the front
|
|
copy(b.entries[1:], b.entries[:i])
|
|
b.entries[0] = n
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// bumpOrAdd moves n to the front of the bucket entry list or adds it if the list isn't
|
|
// full. The return value is true if n is in the bucket.
|
|
func (tab *Table) bumpOrAdd(b *bucket, n *Node) bool {
|
|
if b.bump(n) {
|
|
return true
|
|
}
|
|
if len(b.entries) >= bucketSize || !tab.addIP(b, n.IP) {
|
|
return false
|
|
}
|
|
b.entries, _ = pushNode(b.entries, n, bucketSize)
|
|
b.replacements = deleteNode(b.replacements, n)
|
|
n.addedAt = time.Now()
|
|
if tab.nodeAddedHook != nil {
|
|
tab.nodeAddedHook(n)
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (tab *Table) deleteInBucket(b *bucket, n *Node) {
|
|
b.entries = deleteNode(b.entries, n)
|
|
tab.removeIP(b, n.IP)
|
|
}
|
|
|
|
// pushNode adds n to the front of list, keeping at most max items.
|
|
func pushNode(list []*Node, n *Node, max int) ([]*Node, *Node) {
|
|
if len(list) < max {
|
|
list = append(list, nil)
|
|
}
|
|
removed := list[len(list)-1]
|
|
copy(list[1:], list)
|
|
list[0] = n
|
|
return list, removed
|
|
}
|
|
|
|
// deleteNode removes n from list.
|
|
func deleteNode(list []*Node, n *Node) []*Node {
|
|
for i := range list {
|
|
if list[i].ID == n.ID {
|
|
return append(list[:i], list[i+1:]...)
|
|
}
|
|
}
|
|
return list
|
|
}
|
|
|
|
// nodesByDistance is a list of nodes, ordered by
|
|
// distance to target.
|
|
type nodesByDistance struct {
|
|
entries []*Node
|
|
target common.Hash
|
|
}
|
|
|
|
// push adds the given node to the list, keeping the total size below maxElems.
|
|
func (h *nodesByDistance) push(n *Node, maxElems int) {
|
|
ix := sort.Search(len(h.entries), func(i int) bool {
|
|
return distcmp(h.target, h.entries[i].sha, n.sha) > 0
|
|
})
|
|
if len(h.entries) < maxElems {
|
|
h.entries = append(h.entries, n)
|
|
}
|
|
if ix == len(h.entries) {
|
|
// farther away than all nodes we already have.
|
|
// if there was room for it, the node is now the last element.
|
|
} else {
|
|
// slide existing entries down to make room
|
|
// this will overwrite the entry we just appended.
|
|
copy(h.entries[ix+1:], h.entries[ix:])
|
|
h.entries[ix] = n
|
|
}
|
|
}
|