forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

p2p/discover: put back findnode request tracking

Browse source
This commit is contained in:
Felix Lange 2024-04-25 17:37:53 +02:00
parent 2c2c7f7a07
commit ea9f0a4b03
2 changed files with 60 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
p2p/discover/lookup.go
View file

@ -140,32 +140,12 @@ func (it *lookup) slowdown() {
}
func (it *lookup) query(n *node, reply chan<- []*node) {
fails := it.tab.db.FindFails(n.ID(), n.IP())
r, err := it.queryfunc(n)
if errors.Is(err, errClosed) {
if !errors.Is(err, errClosed) {
// Avoid recording failures on shutdown.
reply <- nil
return
} else if len(r) == 0 {
fails++
it.tab.db.UpdateFindFails(n.ID(), n.IP(), fails)
// Remove the node from the local table if it fails to return anything useful too
// many times, but only if there are enough other nodes in the bucket.
dropped := false
if fails >= maxFindnodeFailures {
dropped = true
it.tab.trackFindFailure(n)
}
it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "failcount", fails, "dropped", dropped, "err", err)
} else if fails > 0 {
// Reset failure counter because it counts _consecutive_ failures.
it.tab.db.UpdateFindFails(n.ID(), n.IP(), 0)
}
// Grab as many nodes as possible. Some of them might not be alive anymore, but we'll
// just remove those again during revalidation.
for _, n := range r {
it.tab.addFoundNode(n)
success := len(r) > 0
it.tab.trackRequest(n, success, r)
it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "err", err)
}
reply <- r
}

77
p2p/discover/table.go
View file

@ -78,9 +78,9 @@ type Table struct {
// loop channels
refreshReq chan chan struct{}
revalResponseCh chan revalidationResponse
addNodeCh chan addNodeRequest
addNodeCh chan addNodeOp
addNodeHandled chan struct{}
findFailureCh chan *node
trackRequestCh chan trackRequestOp
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
@ -107,11 +107,17 @@ type bucket struct {
index int
}
type addNodeRequest struct {
type addNodeOp struct {
node *node
isInbound bool
}
type trackRequestOp struct {
node *node
foundNodes []*node
success bool
}
func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
cfg = cfg.withDefaults()
tab := &Table{
@ -121,9 +127,9 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
log: cfg.Log,
refreshReq: make(chan chan struct{}),
revalResponseCh: make(chan revalidationResponse),
addNodeCh: make(chan addNodeRequest),
addNodeCh: make(chan addNodeOp),
addNodeHandled: make(chan struct{}),
findFailureCh: make(chan *node),
trackRequestCh: make(chan trackRequestOp),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
closed: make(chan struct{}),
@ -147,13 +153,6 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
return tab, nil
}
func (tab *Table) trackFindFailure(n *node) {
select {
case tab.findFailureCh <- n:
case <-tab.closed:
}
}
// Nodes returns all nodes contained in the table.
func (tab *Table) Nodes() [][]BucketNode {
tab.mutex.Lock()
@ -306,9 +305,9 @@ func (tab *Table) len() (n int) {
//
// The caller must not hold tab.mutex.
func (tab *Table) addFoundNode(n *node) {
req := addNodeRequest{node: n, isInbound: false}
op := addNodeOp{node: n, isInbound: false}
select {
case tab.addNodeCh <- req:
case tab.addNodeCh <- op:
<-tab.addNodeHandled
case <-tab.closeReq:
}
@ -323,14 +322,22 @@ func (tab *Table) addFoundNode(n *node) {
//
// The caller must not hold tab.mutex.
func (tab *Table) addInboundNode(n *node) {
req := addNodeRequest{node: n, isInbound: true}
op := addNodeOp{node: n, isInbound: true}
select {
case tab.addNodeCh <- req:
case tab.addNodeCh <- op:
<-tab.addNodeHandled
case <-tab.closeReq:
}
}
func (tab *Table) trackRequest(n *node, success bool, foundNodes []*node) {
op := trackRequestOp{n, foundNodes, success}
select {
case tab.trackRequestCh <- op:
case <-tab.closeReq:
}
}
// loop is the main loop of Table.
func (tab *Table) loop() {
var (
@ -361,11 +368,12 @@ loop:
case r := <-tab.revalResponseCh:
tab.revalidation.handleResponse(tab, r)
case addreq := <-tab.addNodeCh:
tab.handleAddNode(addreq)
case op := <-tab.addNodeCh:
tab.handleAddNode(op)
tab.addNodeHandled <- struct{}{}
case <-tab.findFailureCh:
case op := <-tab.trackRequestCh:
tab.handleTrackRequest(op)
// TODO: handle failure by potentially dropping node
case <-refresh.C:
@ -435,7 +443,7 @@ func (tab *Table) loadSeedNodes() {
age := time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP()))
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
}
tab.handleAddNode(addNodeRequest{node: seed, isInbound: true})
tab.handleAddNode(addNodeOp{node: seed, isInbound: true})
}
}
@ -484,7 +492,7 @@ func (tab *Table) removeIP(b *bucket, ip net.IP) {
b.ips.Remove(ip)
}
func (tab *Table) handleAddNode(req addNodeRequest) {
func (tab *Table) handleAddNode(req addNodeOp) {
if req.node.ID() == tab.self().ID() {
return
}
@ -610,6 +618,33 @@ func (tab *Table) bumpInBucket(b *bucket, newRecord *enode.Node) bool {
return true
}
func (tab *Table) handleTrackRequest(op trackRequestOp) {
var fails int
if op.success {
// Reset failure counter because it counts _consecutive_ failures.
tab.db.UpdateFindFails(op.node.ID(), op.node.IP(), 0)
} else {
fails = tab.db.FindFails(op.node.ID(), op.node.IP())
fails++
tab.db.UpdateFindFails(op.node.ID(), op.node.IP(), fails)
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(op.node.ID())
// Remove the node from the local table if it fails to return anything useful too
// many times, but only if there are enough other nodes in the bucket.
if fails >= maxFindnodeFailures && len(b.entries) >= bucketSize/2 {
tab.deleteInBucket(b, op.node.ID())
}
// Add found nodes.
for _, n := range op.foundNodes {
tab.handleAddNode(addNodeOp{n, false})
}
}
func contains(ns []*node, id enode.ID) bool {
for _, n := range ns {
if n.ID() == id {