p2p/discover: decouple nodeFeed from Table mutex in waitForNodes

This commit is contained in:
Felix Lange 2026-05-07 12:39:02 +02:00
parent 11cb8b55a6
commit 85e81382ad

View file

@ -753,6 +753,36 @@ func (tab *Table) deleteNode(n *enode.Node) {
// waitForNodes blocks until the table contains at least n nodes. // waitForNodes blocks until the table contains at least n nodes.
func (tab *Table) waitForNodes(ctx context.Context, n int) error { func (tab *Table) waitForNodes(ctx context.Context, n int) error {
// Set up a notification channel that gets unblocked when there was any activity on
// the table. Ultimately this reads from the table's nodeFeed, but can't use the feed
// directly on the same goroutine that takes Table.mutex, it would deadlock.
var notify chan struct{}
var notifyErr error
initsub := func() event.Subscription {
notify = make(chan struct{}, 1)
newnode := make(chan *enode.Node, 1)
sub := tab.nodeFeed.Subscribe(newnode)
go func() {
defer close(notify)
for {
select {
case <-newnode:
select {
case notify <- struct{}{}:
default:
}
case <-ctx.Done():
notifyErr = ctx.Err()
return
case <-tab.closeReq:
notifyErr = errClosed
return
}
}
}()
return sub
}
getlength := func() (count int) { getlength := func() (count int) {
for _, b := range &tab.buckets { for _, b := range &tab.buckets {
count += len(b.entries) count += len(b.entries)
@ -760,28 +790,25 @@ func (tab *Table) waitForNodes(ctx context.Context, n int) error {
return count return count
} }
var ch chan *enode.Node
for { for {
tab.mutex.Lock() tab.mutex.Lock()
if getlength() >= n { if getlength() >= n {
tab.mutex.Unlock() tab.mutex.Unlock()
return nil return nil
} }
if ch == nil { if notify == nil {
// Init subscription. // Lazily init the subscription.
ch = make(chan *enode.Node) sub := initsub()
sub := tab.nodeFeed.Subscribe(ch)
defer sub.Unsubscribe() defer sub.Unsubscribe()
} }
tab.mutex.Unlock() tab.mutex.Unlock()
// Wait for a node add event. // Wait for table event.
select { _, ok := <-notify
case <-ch: if !ok {
case <-ctx.Done(): break
return ctx.Err()
case <-tab.closeReq:
return errClosed
} }
} }
return notifyErr
} }