forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

revert table modification

Browse source 
Signed-off-by: Chen Kai <281165273grape@gmail.com>
This commit is contained in:
Chen Kai 2024-12-10 14:13:18 +08:00
parent c3669fd24c
commit 5095282864
12 changed files with 220 additions and 220 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
p2p/discover/lookup.go
View file

@ -24,30 +24,30 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
) )
// Lookup performs a network search for nodes close to the given target. It approaches the // 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 // target by querying nodes that are closer to it on each iteration. The given target does
// not need to be an actual node identifier. // not need to be an actual node identifier.
type Lookup struct { type lookup struct {
tab *Table tab *Table
queryfunc queryFunc queryfunc queryFunc
replyCh chan []*enode.Node replyCh chan []*enode.Node
cancelCh <-chan struct{} cancelCh <-chan struct{}
asked, seen map[enode.ID]bool asked, seen map[enode.ID]bool
result NodesByDistance result nodesByDistance
replyBuffer []*enode.Node replyBuffer []*enode.Node
queries int queries int
} }
type queryFunc func(*enode.Node) ([]*enode.Node, error) type queryFunc func(*enode.Node) ([]*enode.Node, error)
func NewLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *Lookup { func newLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *lookup {
it := &Lookup{ it := &lookup{
tab: tab, tab: tab,
queryfunc: q, queryfunc: q,
asked: make(map[enode.ID]bool), asked: make(map[enode.ID]bool),
seen: make(map[enode.ID]bool), seen: make(map[enode.ID]bool),
result: NodesByDistance{Target: target}, result: nodesByDistance{target: target},
replyCh: make(chan []*enode.Node, Alpha), replyCh: make(chan []*enode.Node, alpha),
cancelCh: ctx.Done(), cancelCh: ctx.Done(),
queries: -1, queries: -1,
} }
@ -57,16 +57,16 @@ func NewLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *L
return it return it
} }
// Run runs the lookup to completion and returns the closest nodes found. // run runs the lookup to completion and returns the closest nodes found.
func (it *Lookup) Run() []*enode.Node { func (it *lookup) run() []*enode.Node {
for it.advance() { for it.advance() {
} }
return it.result.Entries return it.result.entries
} }
// advance advances the lookup until any new nodes have been found. // advance advances the lookup until any new nodes have been found.
// It returns false when the lookup has ended. // It returns false when the lookup has ended.
func (it *Lookup) advance() bool { func (it *lookup) advance() bool {
for it.startQueries() { for it.startQueries() {
select { select {
case nodes := <-it.replyCh: case nodes := <-it.replyCh:
@ -74,7 +74,7 @@ func (it *Lookup) advance() bool {
for _, n := range nodes { for _, n := range nodes {
if n != nil && !it.seen[n.ID()] { if n != nil && !it.seen[n.ID()] {
it.seen[n.ID()] = true it.seen[n.ID()] = true
it.result.Push(n, BucketSize) it.result.push(n, bucketSize)
it.replyBuffer = append(it.replyBuffer, n) it.replyBuffer = append(it.replyBuffer, n)
} }
} }
@ -89,7 +89,7 @@ func (it *Lookup) advance() bool {
return false return false
} }
func (it *Lookup) shutdown() { func (it *lookup) shutdown() {
for it.queries > 0 { for it.queries > 0 {
<-it.replyCh <-it.replyCh
it.queries-- it.queries--
@ -98,28 +98,28 @@ func (it *Lookup) shutdown() {
it.replyBuffer = nil it.replyBuffer = nil
} }
func (it *Lookup) startQueries() bool { func (it *lookup) startQueries() bool {
if it.queryfunc == nil { if it.queryfunc == nil {
return false return false
} }
// The first query returns nodes from the local table. // The first query returns nodes from the local table.
if it.queries == -1 { if it.queries == -1 {
closest := it.tab.FindnodeByID(it.result.Target, BucketSize, false) closest := it.tab.findnodeByID(it.result.target, bucketSize, false)
// Avoid finishing the lookup too quickly if table is empty. It'd be better to wait // Avoid finishing the lookup too quickly if table is empty. It'd be better to wait
// for the table to fill in this case, but there is no good mechanism for that // for the table to fill in this case, but there is no good mechanism for that
// yet. // yet.
if len(closest.Entries) == 0 { if len(closest.entries) == 0 {
it.slowdown() it.slowdown()
} }
it.queries = 1 it.queries = 1
it.replyCh <- closest.Entries it.replyCh <- closest.entries
return true return true
} }
// Ask the closest nodes that we haven't asked yet. // Ask the closest nodes that we haven't asked yet.
for i := 0; i < len(it.result.Entries) && it.queries < Alpha; i++ { for i := 0; i < len(it.result.entries) && it.queries < alpha; i++ {
n := it.result.Entries[i] n := it.result.entries[i]
if !it.asked[n.ID()] { if !it.asked[n.ID()] {
it.asked[n.ID()] = true it.asked[n.ID()] = true
it.queries++ it.queries++
@ -130,7 +130,7 @@ func (it *Lookup) startQueries() bool {
return it.queries > 0 return it.queries > 0
} }
func (it *Lookup) slowdown() { func (it *lookup) slowdown() {
sleep := time.NewTimer(1 * time.Second) sleep := time.NewTimer(1 * time.Second)
defer sleep.Stop() defer sleep.Stop()
select { select {
@ -139,9 +139,9 @@ func (it *Lookup) slowdown() {
} }
} }
func (it *Lookup) query(n *enode.Node, reply chan<- []*enode.Node) { func (it *lookup) query(n *enode.Node, reply chan<- []*enode.Node) {
r, err := it.queryfunc(n) r, err := it.queryfunc(n)
if !errors.Is(err, ErrClosed) { // avoid recording failures on shutdown. if !errors.Is(err, errClosed) { // avoid recording failures on shutdown.
success := len(r) > 0 success := len(r) > 0
it.tab.trackRequest(n, success, r) it.tab.trackRequest(n, success, r)
if err != nil { if err != nil {
@ -158,10 +158,10 @@ type lookupIterator struct {
nextLookup lookupFunc nextLookup lookupFunc
ctx context.Context ctx context.Context
cancel func() cancel func()
lookup *Lookup lookup *lookup
} }
type lookupFunc func(ctx context.Context) *Lookup type lookupFunc func(ctx context.Context) *lookup
func newLookupIterator(ctx context.Context, next lookupFunc) *lookupIterator { func newLookupIterator(ctx context.Context, next lookupFunc) *lookupIterator {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)

26
p2p/discover/node.go
View file

@ -54,27 +54,27 @@ func (n *tableNode) String() string {
return n.Node.String() return n.Node.String()
} }
// NodesByDistance is a list of nodes, ordered by distance to target. // nodesByDistance is a list of nodes, ordered by distance to target.
type NodesByDistance struct { type nodesByDistance struct {
Entries []*enode.Node entries []*enode.Node
Target enode.ID target enode.ID
} }
// Push adds the given node to the list, keeping the total size below maxElems. // push adds the given node to the list, keeping the total size below maxElems.
func (h *NodesByDistance) Push(n *enode.Node, maxElems int) { func (h *nodesByDistance) push(n *enode.Node, maxElems int) {
ix := sort.Search(len(h.Entries), func(i int) bool { ix := sort.Search(len(h.entries), func(i int) bool {
return enode.DistCmp(h.Target, h.Entries[i].ID(), n.ID()) > 0 return enode.DistCmp(h.target, h.entries[i].ID(), n.ID()) > 0
}) })
end := len(h.Entries) end := len(h.entries)
if len(h.Entries) < maxElems { if len(h.entries) < maxElems {
h.Entries = append(h.Entries, n) h.entries = append(h.entries, n)
} }
if ix < end { if ix < end {
// Slide existing entries down to make room. // Slide existing entries down to make room.
// This will overwrite the entry we just appended. // This will overwrite the entry we just appended.
copy(h.Entries[ix+1:], h.Entries[ix:]) copy(h.entries[ix+1:], h.entries[ix:])
h.Entries[ix] = n h.entries[ix] = n
} }
} }

117
p2p/discover/table.go
View file

@ -39,8 +39,8 @@ import (
) )
const ( const (
Alpha = 3 // Kademlia concurrency factor alpha = 3 // Kademlia concurrency factor
BucketSize = 16 // Kademlia bucket size bucketSize = 16 // Kademlia bucket size
maxReplacements = 10 // Size of per-bucket replacement list maxReplacements = 10 // Size of per-bucket replacement list
// We keep buckets for the upper 1/15 of distances because // We keep buckets for the upper 1/15 of distances because
@ -92,9 +92,9 @@ type Table struct {
type transport interface { type transport interface {
Self() *enode.Node Self() *enode.Node
RequestENR(*enode.Node) (*enode.Node, error) RequestENR(*enode.Node) (*enode.Node, error)
LookupRandom() []*enode.Node lookupRandom() []*enode.Node
LookupSelf() []*enode.Node lookupSelf() []*enode.Node
Ping(*enode.Node) (seq uint64, err error) ping(*enode.Node) (seq uint64, err error)
} }
// bucket contains nodes, ordered by their last activity. the entry // bucket contains nodes, ordered by their last activity. the entry
@ -118,7 +118,7 @@ type trackRequestOp struct {
success bool success bool
} }
func NewTable(t transport, db *enode.DB, cfg Config) (*Table, error) { func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
cfg = cfg.withDefaults() cfg = cfg.withDefaults()
tab := &Table{ tab := &Table{
net: t, net: t,
@ -178,8 +178,8 @@ func (tab *Table) self() *enode.Node {
return tab.net.Self() return tab.net.Self()
} }
// GetNode returns the node with the given ID or nil if it isn't in the table. // getNode returns the node with the given ID or nil if it isn't in the table.
func (tab *Table) GetNode(id enode.ID) *enode.Node { func (tab *Table) getNode(id enode.ID) *enode.Node {
tab.mutex.Lock() tab.mutex.Lock()
defer tab.mutex.Unlock() defer tab.mutex.Unlock()
@ -192,8 +192,8 @@ func (tab *Table) GetNode(id enode.ID) *enode.Node {
return nil return nil
} }
// Close terminates the network listener and flushes the node database. // close terminates the network listener and flushes the node database.
func (tab *Table) Close() { func (tab *Table) close() {
close(tab.closeReq) close(tab.closeReq)
<-tab.closed <-tab.closed
} }
@ -237,40 +237,40 @@ func (tab *Table) refresh() <-chan struct{} {
return done return done
} }
// FindnodeByID returns the n nodes in the table that are closest to the given id. // findnodeByID returns the n nodes in the table that are closest to the given id.
// This is used by the FINDNODE/v4 handler. // This is used by the FINDNODE/v4 handler.
// //
// The preferLive parameter says whether the caller wants liveness-checked results. If // The preferLive parameter says whether the caller wants liveness-checked results. If
// preferLive is true and the table contains any verified nodes, the result will not // preferLive is true and the table contains any verified nodes, the result will not
// contain unverified nodes. However, if there are no verified nodes at all, the result // contain unverified nodes. However, if there are no verified nodes at all, the result
// will contain unverified nodes. // will contain unverified nodes.
func (tab *Table) FindnodeByID(target enode.ID, nresults int, preferLive bool) *NodesByDistance { func (tab *Table) findnodeByID(target enode.ID, nresults int, preferLive bool) *nodesByDistance {
tab.mutex.Lock() tab.mutex.Lock()
defer tab.mutex.Unlock() defer tab.mutex.Unlock()
// Scan all buckets. There might be a better way to do this, but there aren't that many // Scan all buckets. There might be a better way to do this, but there aren't that many
// buckets, so this solution should be fine. The worst-case complexity of this loop // buckets, so this solution should be fine. The worst-case complexity of this loop
// is O(tab.len() * nresults). // is O(tab.len() * nresults).
nodes := &NodesByDistance{Target: target} nodes := &nodesByDistance{target: target}
liveNodes := &NodesByDistance{Target: target} liveNodes := &nodesByDistance{target: target}
for _, b := range &tab.buckets { for _, b := range &tab.buckets {
for _, n := range b.entries { for _, n := range b.entries {
nodes.Push(n.Node, nresults) nodes.push(n.Node, nresults)
if preferLive && n.isValidatedLive { if preferLive && n.isValidatedLive {
liveNodes.Push(n.Node, nresults) liveNodes.push(n.Node, nresults)
} }
} }
} }
if preferLive && len(liveNodes.Entries) > 0 { if preferLive && len(liveNodes.entries) > 0 {
return liveNodes return liveNodes
} }
return nodes return nodes
} }
// AppendBucketNodes adds nodes at the given distance to the result slice. // appendBucketNodes adds nodes at the given distance to the result slice.
// This is used by the FINDNODE/v5 handler. // This is used by the FINDNODE/v5 handler.
func (tab *Table) AppendBucketNodes(dist uint, result []*enode.Node, checkLive bool) []*enode.Node { func (tab *Table) appendBucketNodes(dist uint, result []*enode.Node, checkLive bool) []*enode.Node {
if dist > 256 { if dist > 256 {
return result return result
} }
@ -304,12 +304,12 @@ func (tab *Table) len() (n int) {
return n return n
} }
// AddFoundNode adds a node which may not be live. If the bucket has space available, // addFoundNode adds a node which may not be live. If the bucket has space available,
// adding the node succeeds immediately. Otherwise, the node is added to the replacements // adding the node succeeds immediately. Otherwise, the node is added to the replacements
// list. // list.
// //
// The caller must not hold tab.mutex. // The caller must not hold tab.mutex.
func (tab *Table) AddFoundNode(n *enode.Node, forceSetLive bool) bool { func (tab *Table) addFoundNode(n *enode.Node, forceSetLive bool) bool {
op := addNodeOp{node: n, isInbound: false, forceSetLive: forceSetLive} op := addNodeOp{node: n, isInbound: false, forceSetLive: forceSetLive}
select { select {
case tab.addNodeCh <- op: case tab.addNodeCh <- op:
@ -319,7 +319,7 @@ func (tab *Table) AddFoundNode(n *enode.Node, forceSetLive bool) bool {
} }
} }
// AddInboundNode adds a node from an inbound contact. If the bucket has no space, the // addInboundNode adds a node from an inbound contact. If the bucket has no space, the
// node is added to the replacements list. // node is added to the replacements list.
// //
// There is an additional safety measure: if the table is still initializing the node is // There is an additional safety measure: if the table is still initializing the node is
@ -327,7 +327,7 @@ func (tab *Table) AddFoundNode(n *enode.Node, forceSetLive bool) bool {
// repeatedly. // repeatedly.
// //
// The caller must not hold tab.mutex. // The caller must not hold tab.mutex.
func (tab *Table) AddInboundNode(n *enode.Node) bool { func (tab *Table) addInboundNode(n *enode.Node) bool {
op := addNodeOp{node: n, isInbound: true} op := addNodeOp{node: n, isInbound: true}
select { select {
case tab.addNodeCh <- op: case tab.addNodeCh <- op:
@ -345,8 +345,8 @@ func (tab *Table) trackRequest(n *enode.Node, success bool, foundNodes []*enode.
} }
} }
// Loop is the main loop of Table. // loop is the main loop of Table.
func (tab *Table) Loop() { func (tab *Table) loop() {
var ( var (
refresh = time.NewTimer(tab.nextRefreshTime()) refresh = time.NewTimer(tab.nextRefreshTime())
refreshDone = make(chan struct{}) // where doRefresh reports completion refreshDone = make(chan struct{}) // where doRefresh reports completion
@ -429,7 +429,7 @@ func (tab *Table) doRefresh(done chan struct{}) {
tab.loadSeedNodes() tab.loadSeedNodes()
// Run self lookup to discover new neighbor nodes. // Run self lookup to discover new neighbor nodes.
tab.net.LookupSelf() tab.net.lookupSelf()
// The Kademlia paper specifies that the bucket refresh should // The Kademlia paper specifies that the bucket refresh should
// perform a lookup in the least recently used bucket. We cannot // perform a lookup in the least recently used bucket. We cannot
@ -438,7 +438,7 @@ func (tab *Table) doRefresh(done chan struct{}) {
// sha3 preimage that falls into a chosen bucket. // sha3 preimage that falls into a chosen bucket.
// We perform a few lookups with a random target instead. // We perform a few lookups with a random target instead.
for i := 0; i < 3; i++ { for i := 0; i < 3; i++ {
tab.net.LookupRandom() tab.net.lookupRandom()
} }
} }
@ -521,7 +521,7 @@ func (tab *Table) handleAddNode(req addNodeOp) bool {
// Already in bucket. // Already in bucket.
return false return false
} }
if len(b.entries) >= BucketSize { if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement. // Bucket full, maybe add as replacement.
tab.addReplacement(b, req.node) tab.addReplacement(b, req.node)
return false return false
@ -674,7 +674,7 @@ func (tab *Table) handleTrackRequest(op trackRequestOp) {
// many times, but only if there are enough other nodes in the bucket. This latter // many times, but only if there are enough other nodes in the bucket. This latter
// condition specifically exists to make bootstrapping in smaller test networks more // condition specifically exists to make bootstrapping in smaller test networks more
// reliable. // reliable.
if fails >= maxFindnodeFailures && len(b.entries) >= BucketSize/4 { if fails >= maxFindnodeFailures && len(b.entries) >= bucketSize/4 {
tab.deleteInBucket(b, op.node.ID()) tab.deleteInBucket(b, op.node.ID())
} }
@ -695,41 +695,8 @@ func pushNode(list []*tableNode, n *tableNode, max int) ([]*tableNode, *tableNod
return list, removed return list, removed
} }
// WaitInit waits until the table is initialized. // nodeList returns all nodes contained in the table.
func (tab *Table) WaitInit() { func (tab *Table) nodeList() []*enode.Node {
<-tab.initDone
}
// NodeIds returns the node IDs in the table.
func (tab *Table) NodeIds() [][]string {
tab.mutex.Lock()
defer tab.mutex.Unlock()
nodes := make([][]string, 0)
for _, b := range &tab.buckets {
bucketNodes := make([]string, 0)
for _, n := range b.entries {
bucketNodes = append(bucketNodes, "0x"+n.ID().String())
}
nodes = append(nodes, bucketNodes)
}
return nodes
}
// Config returns the table's configuration.
func (tab *Table) Config() Config {
return tab.cfg
}
// DeleteNode removes a node from the table.
func (tab *Table) DeleteNode(n *enode.Node) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
tab.deleteInBucket(b, n.ID())
}
// NodeList returns all nodes contained in the table.
func (tab *Table) NodeList() []*enode.Node {
if !tab.isInitDone() { if !tab.isInitDone() {
return nil return nil
} }
@ -745,3 +712,25 @@ func (tab *Table) NodeList() []*enode.Node {
} }
return nodes return nodes
} }
// nodeIds returns the node IDs in the table.
func (tab *Table) nodeIds() [][]string {
tab.mutex.Lock()
defer tab.mutex.Unlock()
nodes := make([][]string, 0)
for _, b := range &tab.buckets {
bucketNodes := make([]string, 0)
for _, n := range b.entries {
bucketNodes = append(bucketNodes, "0x"+n.ID().String())
}
nodes = append(nodes, bucketNodes)
}
return nodes
}
func (tab *Table) deleteNode(n *enode.Node) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
tab.deleteInBucket(b, n.ID())
}

2
p2p/discover/table_reval.go
View file

@ -111,7 +111,7 @@ func (tr *tableRevalidation) startRequest(tab *Table, n *tableNode) {
func (tab *Table) doRevalidate(resp revalidationResponse, node *enode.Node) { func (tab *Table) doRevalidate(resp revalidationResponse, node *enode.Node) {
// Ping the selected node and wait for a pong response. // Ping the selected node and wait for a pong response.
remoteSeq, err := tab.net.Ping(node) remoteSeq, err := tab.net.ping(node)
resp.didRespond = err == nil resp.didRespond = err == nil
// Also fetch record if the node replied and returned a higher sequence number. // Also fetch record if the node replied and returned a higher sequence number.

2
p2p/discover/table_reval_test.go
View file

@ -63,7 +63,7 @@ func TestRevalidation_nodeRemoved(t *testing.T) {
tr.handleResponse(tab, resp) tr.handleResponse(tab, resp)
// Ensure the node was not re-added to the table. // Ensure the node was not re-added to the table.
if tab.GetNode(node.ID()) != nil { if tab.getNode(node.ID()) != nil {
t.Fatal("node was re-added to Table") t.Fatal("node was re-added to Table")
} }
if tr.fast.contains(node.ID()) || tr.slow.contains(node.ID()) { if tr.fast.contains(node.ID()) || tr.slow.contains(node.ID()) {

70
p2p/discover/table_test.go
View file

@ -59,7 +59,7 @@ func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding
Log: testlog.Logger(t, log.LevelTrace), Log: testlog.Logger(t, log.LevelTrace),
}) })
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
<-tab.initDone <-tab.initDone
@ -79,7 +79,7 @@ func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding
transport.dead[replacementNode.ID()] = !newNodeIsResponding transport.dead[replacementNode.ID()] = !newNodeIsResponding
// Add replacement node to table. // Add replacement node to table.
tab.AddFoundNode(replacementNode, false) tab.addFoundNode(replacementNode, false)
t.Log("last:", last.ID()) t.Log("last:", last.ID())
t.Log("replacement:", replacementNode.ID()) t.Log("replacement:", replacementNode.ID())
@ -108,7 +108,7 @@ func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding
// Check bucket content. // Check bucket content.
tab.mutex.Lock() tab.mutex.Lock()
defer tab.mutex.Unlock() defer tab.mutex.Unlock()
wantSize := BucketSize wantSize := bucketSize
if !lastInBucketIsResponding && !newNodeIsResponding { if !lastInBucketIsResponding && !newNodeIsResponding {
wantSize-- wantSize--
} }
@ -150,11 +150,11 @@ func TestTable_IPLimit(t *testing.T) {
transport := newPingRecorder() transport := newPingRecorder()
tab, db := newTestTable(transport, Config{}) tab, db := newTestTable(transport, Config{})
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
for i := 0; i < tableIPLimit+1; i++ { for i := 0; i < tableIPLimit+1; i++ {
n := nodeAtDistance(tab.self().ID(), i, net.IP{172, 0, 1, byte(i)}) n := nodeAtDistance(tab.self().ID(), i, net.IP{172, 0, 1, byte(i)})
tab.AddFoundNode(n, false) tab.addFoundNode(n, false)
} }
if tab.len() > tableIPLimit { if tab.len() > tableIPLimit {
t.Errorf("too many nodes in table") t.Errorf("too many nodes in table")
@ -167,12 +167,12 @@ func TestTable_BucketIPLimit(t *testing.T) {
transport := newPingRecorder() transport := newPingRecorder()
tab, db := newTestTable(transport, Config{}) tab, db := newTestTable(transport, Config{})
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
d := 3 d := 3
for i := 0; i < bucketIPLimit+1; i++ { for i := 0; i < bucketIPLimit+1; i++ {
n := nodeAtDistance(tab.self().ID(), d, net.IP{172, 0, 1, byte(i)}) n := nodeAtDistance(tab.self().ID(), d, net.IP{172, 0, 1, byte(i)})
tab.AddFoundNode(n, false) tab.addFoundNode(n, false)
} }
if tab.len() > bucketIPLimit { if tab.len() > bucketIPLimit {
t.Errorf("too many nodes in table") t.Errorf("too many nodes in table")
@ -204,11 +204,11 @@ func TestTable_findnodeByID(t *testing.T) {
transport := newPingRecorder() transport := newPingRecorder()
tab, db := newTestTable(transport, Config{}) tab, db := newTestTable(transport, Config{})
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
fillTable(tab, test.All, true) fillTable(tab, test.All, true)
// check that closest(Target, N) returns nodes // check that closest(Target, N) returns nodes
result := tab.FindnodeByID(test.Target, test.N, false).Entries result := tab.findnodeByID(test.Target, test.N, false).entries
if hasDuplicates(result) { if hasDuplicates(result) {
t.Errorf("result contains duplicates") t.Errorf("result contains duplicates")
return false return false
@ -264,7 +264,7 @@ func (*closeTest) Generate(rand *rand.Rand, size int) reflect.Value {
t := &closeTest{ t := &closeTest{
Self: gen(enode.ID{}, rand).(enode.ID), Self: gen(enode.ID{}, rand).(enode.ID),
Target: gen(enode.ID{}, rand).(enode.ID), Target: gen(enode.ID{}, rand).(enode.ID),
N: rand.Intn(BucketSize), N: rand.Intn(bucketSize),
} }
for _, id := range gen([]enode.ID{}, rand).([]enode.ID) { for _, id := range gen([]enode.ID{}, rand).([]enode.ID) {
r := new(enr.Record) r := new(enr.Record)
@ -279,20 +279,20 @@ func TestTable_addInboundNode(t *testing.T) {
tab, db := newTestTable(newPingRecorder(), Config{}) tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone <-tab.initDone
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
// Insert two nodes. // Insert two nodes.
n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1}) n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2}) n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
tab.AddFoundNode(n1, false) tab.addFoundNode(n1, false)
tab.AddFoundNode(n2, false) tab.addFoundNode(n2, false)
checkBucketContent(t, tab, []*enode.Node{n1, n2}) checkBucketContent(t, tab, []*enode.Node{n1, n2})
// Add a changed version of n2. The bucket should be updated. // Add a changed version of n2. The bucket should be updated.
newrec := n2.Record() newrec := n2.Record()
newrec.Set(enr.IP{99, 99, 99, 99}) newrec.Set(enr.IP{99, 99, 99, 99})
n2v2 := enode.SignNull(newrec, n2.ID()) n2v2 := enode.SignNull(newrec, n2.ID())
tab.AddInboundNode(n2v2) tab.addInboundNode(n2v2)
checkBucketContent(t, tab, []*enode.Node{n1, n2v2}) checkBucketContent(t, tab, []*enode.Node{n1, n2v2})
// Try updating n2 without sequence number change. The update is accepted // Try updating n2 without sequence number change. The update is accepted
@ -301,7 +301,7 @@ func TestTable_addInboundNode(t *testing.T) {
newrec.Set(enr.IP{100, 100, 100, 100}) newrec.Set(enr.IP{100, 100, 100, 100})
newrec.SetSeq(n2.Seq()) newrec.SetSeq(n2.Seq())
n2v3 := enode.SignNull(newrec, n2.ID()) n2v3 := enode.SignNull(newrec, n2.ID())
tab.AddInboundNode(n2v3) tab.addInboundNode(n2v3)
checkBucketContent(t, tab, []*enode.Node{n1, n2v3}) checkBucketContent(t, tab, []*enode.Node{n1, n2v3})
} }
@ -309,20 +309,20 @@ func TestTable_addFoundNode(t *testing.T) {
tab, db := newTestTable(newPingRecorder(), Config{}) tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone <-tab.initDone
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
// Insert two nodes. // Insert two nodes.
n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1}) n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2}) n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
tab.AddFoundNode(n1, false) tab.addFoundNode(n1, false)
tab.AddFoundNode(n2, false) tab.addFoundNode(n2, false)
checkBucketContent(t, tab, []*enode.Node{n1, n2}) checkBucketContent(t, tab, []*enode.Node{n1, n2})
// Add a changed version of n2. The bucket should be updated. // Add a changed version of n2. The bucket should be updated.
newrec := n2.Record() newrec := n2.Record()
newrec.Set(enr.IP{99, 99, 99, 99}) newrec.Set(enr.IP{99, 99, 99, 99})
n2v2 := enode.SignNull(newrec, n2.ID()) n2v2 := enode.SignNull(newrec, n2.ID())
tab.AddFoundNode(n2v2, false) tab.addFoundNode(n2v2, false)
checkBucketContent(t, tab, []*enode.Node{n1, n2v2}) checkBucketContent(t, tab, []*enode.Node{n1, n2v2})
// Try updating n2 without a sequence number change. // Try updating n2 without a sequence number change.
@ -331,7 +331,7 @@ func TestTable_addFoundNode(t *testing.T) {
newrec.Set(enr.IP{100, 100, 100, 100}) newrec.Set(enr.IP{100, 100, 100, 100})
newrec.SetSeq(n2.Seq()) newrec.SetSeq(n2.Seq())
n2v3 := enode.SignNull(newrec, n2.ID()) n2v3 := enode.SignNull(newrec, n2.ID())
tab.AddFoundNode(n2v3, false) tab.addFoundNode(n2v3, false)
checkBucketContent(t, tab, []*enode.Node{n1, n2v2}) checkBucketContent(t, tab, []*enode.Node{n1, n2v2})
} }
@ -340,18 +340,18 @@ func TestTable_addInboundNodeUpdateV4Accept(t *testing.T) {
tab, db := newTestTable(newPingRecorder(), Config{}) tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone <-tab.initDone
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
// Add a v4 node. // Add a v4 node.
key, _ := crypto.HexToECDSA("dd3757a8075e88d0f2b1431e7d3c5b1562e1c0aab9643707e8cbfcc8dae5cfe3") key, _ := crypto.HexToECDSA("dd3757a8075e88d0f2b1431e7d3c5b1562e1c0aab9643707e8cbfcc8dae5cfe3")
n1 := enode.NewV4(&key.PublicKey, net.IP{88, 77, 66, 1}, 9000, 9000) n1 := enode.NewV4(&key.PublicKey, net.IP{88, 77, 66, 1}, 9000, 9000)
tab.AddInboundNode(n1) tab.addInboundNode(n1)
checkBucketContent(t, tab, []*enode.Node{n1}) checkBucketContent(t, tab, []*enode.Node{n1})
// Add an updated version with changed IP. // Add an updated version with changed IP.
// The update will be accepted because it is inbound. // The update will be accepted because it is inbound.
n1v2 := enode.NewV4(&key.PublicKey, net.IP{99, 99, 99, 99}, 9000, 9000) n1v2 := enode.NewV4(&key.PublicKey, net.IP{99, 99, 99, 99}, 9000, 9000)
tab.AddInboundNode(n1v2) tab.addInboundNode(n1v2)
checkBucketContent(t, tab, []*enode.Node{n1v2}) checkBucketContent(t, tab, []*enode.Node{n1v2})
} }
@ -361,18 +361,18 @@ func TestTable_addFoundNodeV4UpdateReject(t *testing.T) {
tab, db := newTestTable(newPingRecorder(), Config{}) tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone <-tab.initDone
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
// Add a v4 node. // Add a v4 node.
key, _ := crypto.HexToECDSA("dd3757a8075e88d0f2b1431e7d3c5b1562e1c0aab9643707e8cbfcc8dae5cfe3") key, _ := crypto.HexToECDSA("dd3757a8075e88d0f2b1431e7d3c5b1562e1c0aab9643707e8cbfcc8dae5cfe3")
n1 := enode.NewV4(&key.PublicKey, net.IP{88, 77, 66, 1}, 9000, 9000) n1 := enode.NewV4(&key.PublicKey, net.IP{88, 77, 66, 1}, 9000, 9000)
tab.AddFoundNode(n1, false) tab.addFoundNode(n1, false)
checkBucketContent(t, tab, []*enode.Node{n1}) checkBucketContent(t, tab, []*enode.Node{n1})
// Add an updated version with changed IP. // Add an updated version with changed IP.
// The update won't be accepted because it isn't inbound. // The update won't be accepted because it isn't inbound.
n1v2 := enode.NewV4(&key.PublicKey, net.IP{99, 99, 99, 99}, 9000, 9000) n1v2 := enode.NewV4(&key.PublicKey, net.IP{99, 99, 99, 99}, 9000, 9000)
tab.AddFoundNode(n1v2, false) tab.addFoundNode(n1v2, false)
checkBucketContent(t, tab, []*enode.Node{n1}) checkBucketContent(t, tab, []*enode.Node{n1})
} }
@ -407,14 +407,14 @@ func TestTable_revalidateSyncRecord(t *testing.T) {
}) })
<-tab.initDone <-tab.initDone
defer db.Close() defer db.Close()
defer tab.Close() defer tab.close()
// Insert a node. // Insert a node.
var r enr.Record var r enr.Record
r.Set(enr.IP(net.IP{127, 0, 0, 1})) r.Set(enr.IP(net.IP{127, 0, 0, 1}))
id := enode.ID{1} id := enode.ID{1}
n1 := enode.SignNull(&r, id) n1 := enode.SignNull(&r, id)
tab.AddFoundNode(n1, false) tab.addFoundNode(n1, false)
// Update the node record. // Update the node record.
r.Set(enr.WithEntry("foo", "bar")) r.Set(enr.WithEntry("foo", "bar"))
@ -426,7 +426,7 @@ func TestTable_revalidateSyncRecord(t *testing.T) {
waitForRevalidationPing(t, transport, tab, n2.ID()) waitForRevalidationPing(t, transport, tab, n2.ID())
waitForRevalidationPing(t, transport, tab, n2.ID()) waitForRevalidationPing(t, transport, tab, n2.ID())
intable := tab.GetNode(id) intable := tab.getNode(id)
if !reflect.DeepEqual(intable, n2) { if !reflect.DeepEqual(intable, n2) {
t.Fatalf("table contains old record with seq %d, want seq %d", intable.Seq(), n2.Seq()) t.Fatalf("table contains old record with seq %d, want seq %d", intable.Seq(), n2.Seq())
} }
@ -448,22 +448,22 @@ func TestNodesPush(t *testing.T) {
// Insert all permutations into lists with size limit 3. // Insert all permutations into lists with size limit 3.
for _, nodes := range perm { for _, nodes := range perm {
list := NodesByDistance{Target: target} list := nodesByDistance{target: target}
for _, n := range nodes { for _, n := range nodes {
list.Push(n, 3) list.push(n, 3)
} }
if !slices.EqualFunc(list.Entries, perm[0], nodeIDEqual) { if !slices.EqualFunc(list.entries, perm[0], nodeIDEqual) {
t.Fatal("not equal") t.Fatal("not equal")
} }
} }
// Insert all permutations into lists with size limit 2. // Insert all permutations into lists with size limit 2.
for _, nodes := range perm { for _, nodes := range perm {
list := NodesByDistance{Target: target} list := nodesByDistance{target: target}
for _, n := range nodes { for _, n := range nodes {
list.Push(n, 2) list.push(n, 2)
} }
if !slices.EqualFunc(list.Entries, perm[0][:2], nodeIDEqual) { if !slices.EqualFunc(list.entries, perm[0][:2], nodeIDEqual) {
t.Fatal("not equal") t.Fatal("not equal")
} }
} }

20
p2p/discover/table_util_test.go
View file

@ -45,14 +45,14 @@ func init() {
func newTestTable(t transport, cfg Config) (*Table, *enode.DB) { func newTestTable(t transport, cfg Config) (*Table, *enode.DB) {
tab, db := newInactiveTestTable(t, cfg) tab, db := newInactiveTestTable(t, cfg)
go tab.Loop() go tab.loop()
return tab, db return tab, db
} }
// newInactiveTestTable creates a Table without running the main loop. // newInactiveTestTable creates a Table without running the main loop.
func newInactiveTestTable(t transport, cfg Config) (*Table, *enode.DB) { func newInactiveTestTable(t transport, cfg Config) (*Table, *enode.DB) {
db, _ := enode.OpenDB("") db, _ := enode.OpenDB("")
tab, _ := NewTable(t, db, cfg) tab, _ := newTable(t, db, cfg)
return tab, db return tab, db
} }
@ -64,7 +64,7 @@ func nodeAtDistance(base enode.ID, ld int, ip net.IP) *enode.Node {
return enode.SignNull(&r, idAtDistance(base, ld)) return enode.SignNull(&r, idAtDistance(base, ld))
} }
// nodesAtDistance creates n nodes for which enode.LogDist(base, Node.ID()) == ld. // nodesAtDistance creates n nodes for which enode.LogDist(base, node.ID()) == ld.
func nodesAtDistance(base enode.ID, ld int, n int) []*enode.Node { func nodesAtDistance(base enode.ID, ld int, n int) []*enode.Node {
results := make([]*enode.Node, n) results := make([]*enode.Node, n)
for i := range results { for i := range results {
@ -110,20 +110,20 @@ func intIP(i int) net.IP {
func fillBucket(tab *Table, id enode.ID) (last *tableNode) { func fillBucket(tab *Table, id enode.ID) (last *tableNode) {
ld := enode.LogDist(tab.self().ID(), id) ld := enode.LogDist(tab.self().ID(), id)
b := tab.bucket(id) b := tab.bucket(id)
for len(b.entries) < BucketSize { for len(b.entries) < bucketSize {
node := nodeAtDistance(tab.self().ID(), ld, intIP(ld)) node := nodeAtDistance(tab.self().ID(), ld, intIP(ld))
if !tab.AddFoundNode(node, false) { if !tab.addFoundNode(node, false) {
panic("node not added") panic("node not added")
} }
} }
return b.entries[BucketSize-1] return b.entries[bucketSize-1]
} }
// fillTable adds nodes the table to the end of their corresponding bucket // fillTable adds nodes the table to the end of their corresponding bucket
// if the bucket is not full. The caller must not hold tab.mutex. // if the bucket is not full. The caller must not hold tab.mutex.
func fillTable(tab *Table, nodes []*enode.Node, setLive bool) { func fillTable(tab *Table, nodes []*enode.Node, setLive bool) {
for _, n := range nodes { for _, n := range nodes {
tab.AddFoundNode(n, setLive) tab.addFoundNode(n, setLive)
} }
} }
@ -160,8 +160,8 @@ func (t *pingRecorder) updateRecord(n *enode.Node) {
// Stubs to satisfy the transport interface. // Stubs to satisfy the transport interface.
func (t *pingRecorder) Self() *enode.Node { return nullNode } func (t *pingRecorder) Self() *enode.Node { return nullNode }
func (t *pingRecorder) LookupSelf() []*enode.Node { return nil } func (t *pingRecorder) lookupSelf() []*enode.Node { return nil }
func (t *pingRecorder) LookupRandom() []*enode.Node { return nil } func (t *pingRecorder) lookupRandom() []*enode.Node { return nil }
func (t *pingRecorder) waitPing(timeout time.Duration) *enode.Node { func (t *pingRecorder) waitPing(timeout time.Duration) *enode.Node {
t.mu.Lock() t.mu.Lock()
@ -190,7 +190,7 @@ func (t *pingRecorder) waitPing(timeout time.Duration) *enode.Node {
} }
// ping simulates a ping request. // ping simulates a ping request.
func (t *pingRecorder) Ping(n *enode.Node) (seq uint64, err error) { func (t *pingRecorder) ping(n *enode.Node) (seq uint64, err error) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()

4
p2p/discover/v4_lookup_test.go
View file

@ -59,8 +59,8 @@ func TestUDPv4_Lookup(t *testing.T) {
for _, e := range results { for _, e := range results {
t.Logf(" ld=%d, %x", enode.LogDist(lookupTestnet.target.ID(), e.ID()), e.ID().Bytes()) t.Logf(" ld=%d, %x", enode.LogDist(lookupTestnet.target.ID(), e.ID()), e.ID().Bytes())
} }
if len(results) != BucketSize { if len(results) != bucketSize {
t.Errorf("wrong number of results: got %d, want %d", len(results), BucketSize) t.Errorf("wrong number of results: got %d, want %d", len(results), bucketSize)
} }
checkLookupResults(t, lookupTestnet, results) checkLookupResults(t, lookupTestnet, results)
} }

48
p2p/discover/v4_udp.go
View file

@ -43,8 +43,8 @@ var (
errUnknownNode = errors.New("unknown node") errUnknownNode = errors.New("unknown node")
errTimeout = errors.New("RPC timeout") errTimeout = errors.New("RPC timeout")
errClockWarp = errors.New("reply deadline too far in the future") errClockWarp = errors.New("reply deadline too far in the future")
ErrClosed = errors.New("socket closed") errClosed = errors.New("socket closed")
ErrLowPort = errors.New("low port") errLowPort = errors.New("low port")
errNoUDPEndpoint = errors.New("node has no UDP endpoint") errNoUDPEndpoint = errors.New("node has no UDP endpoint")
) )
@ -143,12 +143,12 @@ func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
log: cfg.Log, log: cfg.Log,
} }
tab, err := NewTable(t, ln.Database(), cfg) tab, err := newTable(t, ln.Database(), cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
t.tab = tab t.tab = tab
go tab.Loop() go tab.loop()
t.wg.Add(2) t.wg.Add(2)
go t.loop() go t.loop()
@ -167,7 +167,7 @@ func (t *UDPv4) Close() {
t.cancelCloseCtx() t.cancelCloseCtx()
t.conn.Close() t.conn.Close()
t.wg.Wait() t.wg.Wait()
t.tab.Close() t.tab.close()
}) })
} }
@ -179,7 +179,7 @@ func (t *UDPv4) Resolve(n *enode.Node) *enode.Node {
return rn return rn
} }
// Check table for the ID, we might have a newer version there. // Check table for the ID, we might have a newer version there.
if intable := t.tab.GetNode(n.ID()); intable != nil && intable.Seq() > n.Seq() { if intable := t.tab.getNode(n.ID()); intable != nil && intable.Seq() > n.Seq() {
n = intable n = intable
if rn, err := t.RequestENR(n); err == nil { if rn, err := t.RequestENR(n); err == nil {
return rn return rn
@ -212,12 +212,12 @@ func (t *UDPv4) ourEndpoint() v4wire.Endpoint {
// PingWithoutResp sends a ping message to the given node. // PingWithoutResp sends a ping message to the given node.
func (t *UDPv4) PingWithoutResp(n *enode.Node) error { func (t *UDPv4) PingWithoutResp(n *enode.Node) error {
_, err := t.Ping(n) _, err := t.ping(n)
return err return err
} }
// ping sends a ping message to the given node and waits for a reply. // ping sends a ping message to the given node and waits for a reply.
func (t *UDPv4) Ping(n *enode.Node) (seq uint64, err error) { func (t *UDPv4) ping(n *enode.Node) (seq uint64, err error) {
addr, ok := n.UDPEndpoint() addr, ok := n.UDPEndpoint()
if !ok { if !ok {
return 0, errNoUDPEndpoint return 0, errNoUDPEndpoint
@ -271,7 +271,7 @@ func (t *UDPv4) LookupPubkey(key *ecdsa.PublicKey) []*enode.Node {
// case and run the bootstrapping logic. // case and run the bootstrapping logic.
<-t.tab.refresh() <-t.tab.refresh()
} }
return t.newLookup(t.closeCtx, v4wire.EncodePubkey(key)).Run() return t.newLookup(t.closeCtx, v4wire.EncodePubkey(key)).run()
} }
// RandomNodes is an iterator yielding nodes from a random walk of the DHT. // RandomNodes is an iterator yielding nodes from a random walk of the DHT.
@ -280,25 +280,25 @@ func (t *UDPv4) RandomNodes() enode.Iterator {
} }
// lookupRandom implements transport. // lookupRandom implements transport.
func (t *UDPv4) LookupRandom() []*enode.Node { func (t *UDPv4) lookupRandom() []*enode.Node {
return t.newRandomLookup(t.closeCtx).Run() return t.newRandomLookup(t.closeCtx).run()
} }
// lookupSelf implements transport. // lookupSelf implements transport.
func (t *UDPv4) LookupSelf() []*enode.Node { func (t *UDPv4) lookupSelf() []*enode.Node {
pubkey := v4wire.EncodePubkey(&t.priv.PublicKey) pubkey := v4wire.EncodePubkey(&t.priv.PublicKey)
return t.newLookup(t.closeCtx, pubkey).Run() return t.newLookup(t.closeCtx, pubkey).run()
} }
func (t *UDPv4) newRandomLookup(ctx context.Context) *Lookup { func (t *UDPv4) newRandomLookup(ctx context.Context) *lookup {
var target v4wire.Pubkey var target v4wire.Pubkey
crand.Read(target[:]) crand.Read(target[:])
return t.newLookup(ctx, target) return t.newLookup(ctx, target)
} }
func (t *UDPv4) newLookup(ctx context.Context, targetKey v4wire.Pubkey) *Lookup { func (t *UDPv4) newLookup(ctx context.Context, targetKey v4wire.Pubkey) *lookup {
target := enode.ID(crypto.Keccak256Hash(targetKey[:])) target := enode.ID(crypto.Keccak256Hash(targetKey[:]))
it := NewLookup(ctx, t.tab, target, func(n *enode.Node) ([]*enode.Node, error) { it := newLookup(ctx, t.tab, target, func(n *enode.Node) ([]*enode.Node, error) {
addr, ok := n.UDPEndpoint() addr, ok := n.UDPEndpoint()
if !ok { if !ok {
return nil, errNoUDPEndpoint return nil, errNoUDPEndpoint
@ -315,7 +315,7 @@ func (t *UDPv4) findnode(toid enode.ID, toAddrPort netip.AddrPort, target v4wire
// Add a matcher for 'neighbours' replies to the pending reply queue. The matcher is // Add a matcher for 'neighbours' replies to the pending reply queue. The matcher is
// active until enough nodes have been received. // active until enough nodes have been received.
nodes := make([]*enode.Node, 0, BucketSize) nodes := make([]*enode.Node, 0, bucketSize)
nreceived := 0 nreceived := 0
rm := t.pending(toid, toAddrPort.Addr(), v4wire.NeighborsPacket, func(r v4wire.Packet) (matched bool, requestDone bool) { rm := t.pending(toid, toAddrPort.Addr(), v4wire.NeighborsPacket, func(r v4wire.Packet) (matched bool, requestDone bool) {
reply := r.(*v4wire.Neighbors) reply := r.(*v4wire.Neighbors)
@ -328,7 +328,7 @@ func (t *UDPv4) findnode(toid enode.ID, toAddrPort netip.AddrPort, target v4wire
} }
nodes = append(nodes, n) nodes = append(nodes, n)
} }
return true, nreceived >= BucketSize return true, nreceived >= bucketSize
}) })
t.send(toAddrPort, toid, &v4wire.Findnode{ t.send(toAddrPort, toid, &v4wire.Findnode{
Target: target, Target: target,
@ -400,7 +400,7 @@ func (t *UDPv4) pending(id enode.ID, ip netip.Addr, ptype byte, callback replyMa
case t.addReplyMatcher <- p: case t.addReplyMatcher <- p:
// loop will handle it // loop will handle it
case <-t.closeCtx.Done(): case <-t.closeCtx.Done():
ch <- ErrClosed ch <- errClosed
} }
return p return p
} }
@ -461,7 +461,7 @@ func (t *UDPv4) loop() {
select { select {
case <-t.closeCtx.Done(): case <-t.closeCtx.Done():
for el := plist.Front(); el != nil; el = el.Next() { for el := plist.Front(); el != nil; el = el.Next() {
el.Value.(*replyMatcher).errc <- ErrClosed el.Value.(*replyMatcher).errc <- errClosed
} }
return return
@ -599,7 +599,7 @@ func (t *UDPv4) ensureBond(toid enode.ID, toaddr netip.AddrPort) {
func (t *UDPv4) nodeFromRPC(sender netip.AddrPort, rn v4wire.Node) (*enode.Node, error) { func (t *UDPv4) nodeFromRPC(sender netip.AddrPort, rn v4wire.Node) (*enode.Node, error) {
if rn.UDP <= 1024 { if rn.UDP <= 1024 {
return nil, ErrLowPort return nil, errLowPort
} }
if err := netutil.CheckRelayIP(sender.Addr().AsSlice(), rn.IP); err != nil { if err := netutil.CheckRelayIP(sender.Addr().AsSlice(), rn.IP); err != nil {
return nil, err return nil, err
@ -692,10 +692,10 @@ func (t *UDPv4) handlePing(h *packetHandlerV4, from netip.AddrPort, fromID enode
n := enode.NewV4(h.senderKey, fromIP, int(req.From.TCP), int(from.Port())) n := enode.NewV4(h.senderKey, fromIP, int(req.From.TCP), int(from.Port()))
if time.Since(t.db.LastPongReceived(n.ID(), from.Addr())) > bondExpiration { if time.Since(t.db.LastPongReceived(n.ID(), from.Addr())) > bondExpiration {
t.sendPing(fromID, from, func() { t.sendPing(fromID, from, func() {
t.tab.AddInboundNode(n) t.tab.addInboundNode(n)
}) })
} else { } else {
t.tab.AddInboundNode(n) t.tab.addInboundNode(n)
} }
// Update node database and endpoint predictor. // Update node database and endpoint predictor.
@ -747,7 +747,7 @@ func (t *UDPv4) handleFindnode(h *packetHandlerV4, from netip.AddrPort, fromID e
// Determine closest nodes. // Determine closest nodes.
target := enode.ID(crypto.Keccak256Hash(req.Target[:])) target := enode.ID(crypto.Keccak256Hash(req.Target[:]))
preferLive := !t.tab.cfg.NoFindnodeLivenessCheck preferLive := !t.tab.cfg.NoFindnodeLivenessCheck
closest := t.tab.FindnodeByID(target, BucketSize, preferLive).Entries closest := t.tab.findnodeByID(target, bucketSize, preferLive).entries
// Send neighbors in chunks with at most maxNeighbors per packet // Send neighbors in chunks with at most maxNeighbors per packet
// to stay below the packet size limit. // to stay below the packet size limit.

20
p2p/discover/v4_udp_test.go
View file

@ -112,7 +112,7 @@ func (test *udpTest) waitPacketOut(validate interface{}) (closed bool) {
test.t.Helper() test.t.Helper()
dgram, err := test.pipe.receive() dgram, err := test.pipe.receive()
if err == ErrClosed { if err == errClosed {
return true return true
} else if err != nil { } else if err != nil {
test.t.Error("packet receive error:", err) test.t.Error("packet receive error:", err)
@ -151,7 +151,7 @@ func TestUDPv4_pingTimeout(t *testing.T) {
key := newkey() key := newkey()
toaddr := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 2222} toaddr := &net.UDPAddr{IP: net.ParseIP("1.2.3.4"), Port: 2222}
node := enode.NewV4(&key.PublicKey, toaddr.IP, 0, toaddr.Port) node := enode.NewV4(&key.PublicKey, toaddr.IP, 0, toaddr.Port)
if _, err := test.udp.Ping(node); err != errTimeout { if _, err := test.udp.ping(node); err != errTimeout {
t.Error("expected timeout error, got", err) t.Error("expected timeout error, got", err)
} }
} }
@ -256,9 +256,9 @@ func TestUDPv4_findnode(t *testing.T) {
// put a few nodes into the table. their exact // put a few nodes into the table. their exact
// distribution shouldn't matter much, although we need to // distribution shouldn't matter much, although we need to
// take care not to overflow any bucket. // take care not to overflow any bucket.
nodes := &NodesByDistance{Target: testTarget.ID()} nodes := &nodesByDistance{target: testTarget.ID()}
live := make(map[enode.ID]bool) live := make(map[enode.ID]bool)
numCandidates := 2 * BucketSize numCandidates := 2 * bucketSize
for i := 0; i < numCandidates; i++ { for i := 0; i < numCandidates; i++ {
key := newkey() key := newkey()
ip := net.IP{10, 13, 0, byte(i)} ip := net.IP{10, 13, 0, byte(i)}
@ -267,8 +267,8 @@ func TestUDPv4_findnode(t *testing.T) {
if i > numCandidates/2 { if i > numCandidates/2 {
live[n.ID()] = true live[n.ID()] = true
} }
test.table.AddFoundNode(n, live[n.ID()]) test.table.addFoundNode(n, live[n.ID()])
nodes.Push(n, numCandidates) nodes.push(n, numCandidates)
} }
// ensure there's a bond with the test node, // ensure there's a bond with the test node,
@ -277,7 +277,7 @@ func TestUDPv4_findnode(t *testing.T) {
test.table.db.UpdateLastPongReceived(remoteID, test.remoteaddr.Addr(), time.Now()) test.table.db.UpdateLastPongReceived(remoteID, test.remoteaddr.Addr(), time.Now())
// check that closest neighbors are returned. // check that closest neighbors are returned.
expected := test.table.FindnodeByID(testTarget.ID(), BucketSize, true) expected := test.table.findnodeByID(testTarget.ID(), bucketSize, true)
test.packetIn(nil, &v4wire.Findnode{Target: testTarget, Expiration: futureExp}) test.packetIn(nil, &v4wire.Findnode{Target: testTarget, Expiration: futureExp})
waitNeighbors := func(want []*enode.Node) { waitNeighbors := func(want []*enode.Node) {
test.waitPacketOut(func(p *v4wire.Neighbors, to netip.AddrPort, hash []byte) { test.waitPacketOut(func(p *v4wire.Neighbors, to netip.AddrPort, hash []byte) {
@ -287,7 +287,7 @@ func TestUDPv4_findnode(t *testing.T) {
} }
for i, n := range p.Nodes { for i, n := range p.Nodes {
if n.ID.ID() != want[i].ID() { if n.ID.ID() != want[i].ID() {
t.Errorf("result mismatch at %d:\n got: %v\n want: %v", i, n, expected.Entries[i]) t.Errorf("result mismatch at %d:\n got: %v\n want: %v", i, n, expected.entries[i])
} }
if !live[n.ID.ID()] { if !live[n.ID.ID()] {
t.Errorf("result includes dead node %v", n.ID.ID()) t.Errorf("result includes dead node %v", n.ID.ID())
@ -296,7 +296,7 @@ func TestUDPv4_findnode(t *testing.T) {
}) })
} }
// Receive replies. // Receive replies.
want := expected.Entries want := expected.entries
if len(want) > v4wire.MaxNeighbors { if len(want) > v4wire.MaxNeighbors {
waitNeighbors(want[:v4wire.MaxNeighbors]) waitNeighbors(want[:v4wire.MaxNeighbors])
want = want[v4wire.MaxNeighbors:] want = want[v4wire.MaxNeighbors:]
@ -644,7 +644,7 @@ func (c *dgramPipe) receive() (dgram, error) {
c.cond.Wait() c.cond.Wait()
} }
if c.closed { if c.closed {
return dgram{}, ErrClosed return dgram{}, errClosed
} }
if timedOut { if timedOut {
return dgram{}, errTimeout return dgram{}, errTimeout

67
p2p/discover/v5_udp.go
View file

@ -141,7 +141,7 @@ func ListenV5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
go t.tab.Loop() go t.tab.loop()
t.wg.Add(2) t.wg.Add(2)
go t.readLoop() go t.readLoop()
go t.dispatch() go t.dispatch()
@ -181,7 +181,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
cancelCloseCtx: cancelCloseCtx, cancelCloseCtx: cancelCloseCtx,
} }
t.talk = newTalkSystem(t) t.talk = newTalkSystem(t)
tab, err := NewTable(t, t.db, cfg) tab, err := newTable(t, t.db, cfg)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -201,20 +201,20 @@ func (t *UDPv5) Close() {
t.conn.Close() t.conn.Close()
t.talk.wait() t.talk.wait()
t.wg.Wait() t.wg.Wait()
t.tab.Close() t.tab.close()
}) })
} }
// PingWithoutResp sends a ping message to the given node. // PingWithoutResp sends a ping message to the given node.
func (t *UDPv5) PingWithoutResp(n *enode.Node) error { func (t *UDPv5) PingWithoutResp(n *enode.Node) error {
_, err := t.Ping(n) _, err := t.ping(n)
return err return err
} }
// Resolve searches for a specific node with the given ID and tries to get the most recent // Resolve searches for a specific node with the given ID and tries to get the most recent
// version of the node record for it. It returns n if the node could not be resolved. // version of the node record for it. It returns n if the node could not be resolved.
func (t *UDPv5) Resolve(n *enode.Node) *enode.Node { func (t *UDPv5) Resolve(n *enode.Node) *enode.Node {
if intable := t.tab.GetNode(n.ID()); intable != nil && intable.Seq() > n.Seq() { if intable := t.tab.getNode(n.ID()); intable != nil && intable.Seq() > n.Seq() {
n = intable n = intable
} }
// Try asking directly. This works if the node is still responding on the endpoint we have. // Try asking directly. This works if the node is still responding on the endpoint we have.
@ -238,7 +238,7 @@ func (t *UDPv5) ResolveNodeId(id enode.ID) *enode.Node {
return t.Self() return t.Self()
} }
n := t.tab.GetNode(id) n := t.tab.getNode(id)
if n != nil { if n != nil {
// Try asking directly. This works if the Node is still responding on the endpoint we have. // Try asking directly. This works if the Node is still responding on the endpoint we have.
if resp, err := t.RequestENR(n); err == nil { if resp, err := t.RequestENR(n); err == nil {
@ -263,11 +263,22 @@ func (t *UDPv5) ResolveNodeId(id enode.ID) *enode.Node {
// AllNodes returns all the nodes stored in the local table. // AllNodes returns all the nodes stored in the local table.
func (t *UDPv5) AllNodes() []*enode.Node { func (t *UDPv5) AllNodes() []*enode.Node {
return t.tab.NodeList() return t.tab.nodeList()
} }
// RoutingTableInfo returns the routing table information. Used for Portal discv5 RoutingTableInfo API.
func (t *UDPv5) RoutingTableInfo() [][]string { func (t *UDPv5) RoutingTableInfo() [][]string {
return t.tab.NodeIds() return t.tab.nodeIds()
}
// AddKnownNode adds a node to the routing table. Used for Portal discv5 AddEnr API.
func (t *UDPv5) AddKnownNode(n *enode.Node) bool {
return t.tab.addFoundNode(n, true)
}
// DeleteNode removes a node from the routing table. Used for Portal discv5 DeleteEnr API.
func (t *UDPv5) DeleteNode(n *enode.Node) {
t.tab.deleteNode(n)
} }
// LocalNode returns the current local Node running the // LocalNode returns the current local Node running the
@ -323,29 +334,29 @@ func (t *UDPv5) RandomNodes() enode.Iterator {
// Lookup performs a recursive lookup for the given target. // Lookup performs a recursive lookup for the given target.
// It returns the closest nodes to target. // It returns the closest nodes to target.
func (t *UDPv5) Lookup(target enode.ID) []*enode.Node { func (t *UDPv5) Lookup(target enode.ID) []*enode.Node {
return t.newLookup(t.closeCtx, target).Run() return t.newLookup(t.closeCtx, target).run()
} }
// lookupRandom looks up a random target. // lookupRandom looks up a random target.
// This is needed to satisfy the transport interface. // This is needed to satisfy the transport interface.
func (t *UDPv5) LookupRandom() []*enode.Node { func (t *UDPv5) lookupRandom() []*enode.Node {
return t.newRandomLookup(t.closeCtx).Run() return t.newRandomLookup(t.closeCtx).run()
} }
// lookupSelf looks up our own node ID. // lookupSelf looks up our own node ID.
// This is needed to satisfy the transport interface. // This is needed to satisfy the transport interface.
func (t *UDPv5) LookupSelf() []*enode.Node { func (t *UDPv5) lookupSelf() []*enode.Node {
return t.newLookup(t.closeCtx, t.Self().ID()).Run() return t.newLookup(t.closeCtx, t.Self().ID()).run()
} }
func (t *UDPv5) newRandomLookup(ctx context.Context) *Lookup { func (t *UDPv5) newRandomLookup(ctx context.Context) *lookup {
var target enode.ID var target enode.ID
crand.Read(target[:]) crand.Read(target[:])
return t.newLookup(ctx, target) return t.newLookup(ctx, target)
} }
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *Lookup { func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
return NewLookup(ctx, t.tab, target, func(n *enode.Node) ([]*enode.Node, error) { return newLookup(ctx, t.tab, target, func(n *enode.Node) ([]*enode.Node, error) {
return t.lookupWorker(n, target) return t.lookupWorker(n, target)
}) })
} }
@ -354,20 +365,20 @@ func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *Lookup {
func (t *UDPv5) lookupWorker(destNode *enode.Node, target enode.ID) ([]*enode.Node, error) { func (t *UDPv5) lookupWorker(destNode *enode.Node, target enode.ID) ([]*enode.Node, error) {
var ( var (
dists = LookupDistances(target, destNode.ID()) dists = LookupDistances(target, destNode.ID())
nodes = NodesByDistance{Target: target} nodes = nodesByDistance{target: target}
err error err error
) )
var r []*enode.Node var r []*enode.Node
r, err = t.Findnode(destNode, dists) r, err = t.Findnode(destNode, dists)
if errors.Is(err, ErrClosed) { if errors.Is(err, errClosed) {
return nil, err return nil, err
} }
for _, n := range r { for _, n := range r {
if n.ID() != t.Self().ID() { if n.ID() != t.Self().ID() {
nodes.Push(n, findnodeResultLimit) nodes.push(n, findnodeResultLimit)
} }
} }
return nodes.Entries, err return nodes.entries, err
} }
// LookupDistances computes the distance parameter for FINDNODE calls to dest. // LookupDistances computes the distance parameter for FINDNODE calls to dest.
@ -388,7 +399,7 @@ func LookupDistances(target, dest enode.ID) (dists []uint) {
} }
// ping calls PING on a node and waits for a PONG response. // ping calls PING on a node and waits for a PONG response.
func (t *UDPv5) Ping(n *enode.Node) (uint64, error) { func (t *UDPv5) ping(n *enode.Node) (uint64, error) {
pong, err := t.PingWithResp(n) pong, err := t.PingWithResp(n)
if err != nil { if err != nil {
return 0, err return 0, err
@ -475,7 +486,7 @@ func (t *UDPv5) verifyResponseNode(c *callV5, r *enr.Record, distances []uint, s
return nil, errors.New("not contained in netrestrict list") return nil, errors.New("not contained in netrestrict list")
} }
if node.UDP() <= 1024 { if node.UDP() <= 1024 {
return nil, ErrLowPort return nil, errLowPort
} }
if distances != nil { if distances != nil {
nd := enode.LogDist(c.id, node.ID()) nd := enode.LogDist(c.id, node.ID())
@ -519,7 +530,7 @@ func (t *UDPv5) initCall(c *callV5, responseType byte, packet v5wire.Packet) {
select { select {
case t.callCh <- c: case t.callCh <- c:
case <-t.closeCtx.Done(): case <-t.closeCtx.Done():
c.err <- ErrClosed c.err <- errClosed
} }
} }
@ -612,12 +623,12 @@ func (t *UDPv5) dispatch() {
close(t.readNextCh) close(t.readNextCh)
for id, queue := range t.callQueue { for id, queue := range t.callQueue {
for _, c := range queue { for _, c := range queue {
c.err <- ErrClosed c.err <- errClosed
} }
delete(t.callQueue, id) delete(t.callQueue, id)
} }
for id, c := range t.activeCallByNode { for id, c := range t.activeCallByNode {
c.err <- ErrClosed c.err <- errClosed
delete(t.activeCallByNode, id) delete(t.activeCallByNode, id)
delete(t.activeCallByAuth, c.nonce) delete(t.activeCallByAuth, c.nonce)
} }
@ -774,7 +785,7 @@ func (t *UDPv5) handlePacket(rawpacket []byte, fromAddr netip.AddrPort) error {
} }
if fromNode != nil { if fromNode != nil {
// Handshake succeeded, add to table. // Handshake succeeded, add to table.
t.tab.AddInboundNode(fromNode) t.tab.addInboundNode(fromNode)
t.putCache(fromAddr.String(), fromNode) t.putCache(fromAddr.String(), fromNode)
} }
if packet.Kind() != v5wire.WhoareyouPacket { if packet.Kind() != v5wire.WhoareyouPacket {
@ -809,7 +820,7 @@ func (t *UDPv5) handleCallResponse(fromID enode.ID, fromAddr netip.AddrPort, p v
// GetNode looks for a node record in table and database. // GetNode looks for a node record in table and database.
func (t *UDPv5) GetNode(id enode.ID) *enode.Node { func (t *UDPv5) GetNode(id enode.ID) *enode.Node {
if n := t.tab.GetNode(id); n != nil { if n := t.tab.getNode(id); n != nil {
return n return n
} }
if n := t.localNode.Database().Node(id); n != nil { if n := t.localNode.Database().Node(id); n != nil {
@ -934,7 +945,7 @@ func (t *UDPv5) collectTableNodes(rip netip.Addr, distances []uint, limit int) [
processed[dist] = struct{}{} processed[dist] = struct{}{}
checkLive := !t.tab.cfg.NoFindnodeLivenessCheck checkLive := !t.tab.cfg.NoFindnodeLivenessCheck
for _, n := range t.tab.AppendBucketNodes(dist, bn[:0], checkLive) { for _, n := range t.tab.appendBucketNodes(dist, bn[:0], checkLive) {
// Apply some pre-checks to avoid sending invalid nodes. // Apply some pre-checks to avoid sending invalid nodes.
// Note liveness is checked by appendLiveNodes. // Note liveness is checked by appendLiveNodes.
if netutil.CheckRelayAddr(rip, n.IPAddr()) != nil { if netutil.CheckRelayAddr(rip, n.IPAddr()) != nil {

16
p2p/discover/v5_udp_test.go
View file

@ -143,7 +143,7 @@ func TestUDPv5_unknownPacket(t *testing.T) {
// Make Node known. // Make Node known.
n := test.getNode(test.remotekey, test.remoteaddr).Node() n := test.getNode(test.remotekey, test.remoteaddr).Node()
test.table.AddFoundNode(n, false) test.table.addFoundNode(n, false)
test.packetIn(&v5wire.Unknown{Nonce: nonce}) test.packetIn(&v5wire.Unknown{Nonce: nonce})
test.waitPacketOut(func(p *v5wire.Whoareyou, addr netip.AddrPort, _ v5wire.Nonce) { test.waitPacketOut(func(p *v5wire.Whoareyou, addr netip.AddrPort, _ v5wire.Nonce) {
@ -237,7 +237,7 @@ func TestUDPv5_pingCall(t *testing.T) {
// This ping times out. // This ping times out.
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) {}) test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) {})
@ -247,7 +247,7 @@ func TestUDPv5_pingCall(t *testing.T) {
// This ping works. // This ping works.
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) { test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) {
@ -259,7 +259,7 @@ func TestUDPv5_pingCall(t *testing.T) {
// This ping gets a reply from the wrong endpoint. // This ping gets a reply from the wrong endpoint.
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) { test.waitPacketOut(func(p *v5wire.Ping, addr netip.AddrPort, _ v5wire.Nonce) {
@ -330,11 +330,11 @@ func TestUDPv5_callResend(t *testing.T) {
remote := test.getNode(test.remotekey, test.remoteaddr).Node() remote := test.getNode(test.remotekey, test.remoteaddr).Node()
done := make(chan error, 2) done := make(chan error, 2)
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
@ -367,7 +367,7 @@ func TestUDPv5_multipleHandshakeRounds(t *testing.T) {
remote := test.getNode(test.remotekey, test.remoteaddr).Node() remote := test.getNode(test.remotekey, test.remoteaddr).Node()
done := make(chan error, 1) done := make(chan error, 1)
go func() { go func() {
_, err := test.udp.Ping(remote) _, err := test.udp.ping(remote)
done <- err done <- err
}() }()
@ -817,7 +817,7 @@ func (test *udpV5Test) waitPacketOut(validate interface{}) (closed bool) {
exptype := fn.Type().In(0) exptype := fn.Type().In(0)
dgram, err := test.pipe.receive() dgram, err := test.pipe.receive()
if err == ErrClosed { if err == errClosed {
return true return true
} }
if err == errTimeout { if err == errTimeout {