Browse Source

pex: improve goroutine lifecycle (#7343)

I saw a race detected in a test here that I think would be better
handled by just wiring up these threads.
pull/7349/head
Sam Kleinman 3 years ago
committed by GitHub
parent
commit
070445bc10
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 10 deletions
  1. +13
    -10
      internal/p2p/pex/reactor.go

+ 13
- 10
internal/p2p/pex/reactor.go View File

@ -141,8 +141,8 @@ func NewReactor(
// messages on that p2p channel accordingly. The caller must be sure to execute // messages on that p2p channel accordingly. The caller must be sure to execute
// OnStop to ensure the outbound p2p Channels are closed. // OnStop to ensure the outbound p2p Channels are closed.
func (r *Reactor) OnStart(ctx context.Context) error { func (r *Reactor) OnStart(ctx context.Context) error {
go r.processPexCh()
go r.processPeerUpdates()
go r.processPexCh(ctx)
go r.processPeerUpdates(ctx)
return nil return nil
} }
@ -162,17 +162,22 @@ func (r *Reactor) OnStop() {
// processPexCh implements a blocking event loop where we listen for p2p // processPexCh implements a blocking event loop where we listen for p2p
// Envelope messages from the pexCh. // Envelope messages from the pexCh.
func (r *Reactor) processPexCh() {
func (r *Reactor) processPexCh(ctx context.Context) {
defer r.pexCh.Close() defer r.pexCh.Close()
timer := time.NewTimer(0)
defer timer.Stop()
for { for {
timer.Reset(time.Until(r.nextRequestTime))
select { select {
case <-ctx.Done():
return
case <-r.closeCh: case <-r.closeCh:
r.Logger.Debug("stopped listening on PEX channel; closing...") r.Logger.Debug("stopped listening on PEX channel; closing...")
return return
// outbound requests for new peers // outbound requests for new peers
case <-r.waitUntilNextRequest():
case <-timer.C:
r.sendRequestForPeers() r.sendRequestForPeers()
// inbound requests for new peers or responses to requests sent by this // inbound requests for new peers or responses to requests sent by this
@ -192,11 +197,13 @@ func (r *Reactor) processPexCh() {
// processPeerUpdates initiates a blocking process where we listen for and handle // processPeerUpdates initiates a blocking process where we listen for and handle
// PeerUpdate messages. When the reactor is stopped, we will catch the signal and // PeerUpdate messages. When the reactor is stopped, we will catch the signal and
// close the p2p PeerUpdatesCh gracefully. // close the p2p PeerUpdatesCh gracefully.
func (r *Reactor) processPeerUpdates() {
func (r *Reactor) processPeerUpdates(ctx context.Context) {
defer r.peerUpdates.Close() defer r.peerUpdates.Close()
for { for {
select { select {
case <-ctx.Done():
return
case peerUpdate := <-r.peerUpdates.Updates(): case peerUpdate := <-r.peerUpdates.Updates():
r.processPeerUpdate(peerUpdate) r.processPeerUpdate(peerUpdate)
@ -317,10 +324,6 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
} }
} }
func (r *Reactor) waitUntilNextRequest() <-chan time.Time {
return time.After(time.Until(r.nextRequestTime))
}
// sendRequestForPeers pops the first peerID off the list and sends the // sendRequestForPeers pops the first peerID off the list and sends the
// peer a request for more peer addresses. The function then moves the // peer a request for more peer addresses. The function then moves the
// peer into the requestsSent bucket and calculates when the next request // peer into the requestsSent bucket and calculates when the next request


Loading…
Cancel
Save