Skip to content

Commit

Permalink
multi: Reduce done goroutines.
Browse files Browse the repository at this point in the history
This modifies the logic in several Run methods to block until the
context is cancelled directly in the Run method instead of launching a
separate goroutine for that purpose.

This approach is preferred since it provides the same functionality
without additional long-running goroutines.
  • Loading branch information
davecgh committed Oct 25, 2023
1 parent 672c345 commit ae64e51
Show file tree
Hide file tree
Showing 6 changed files with 49 additions and 70 deletions.
26 changes: 10 additions & 16 deletions connmgr/connmanager.go
Original file line number Diff line number Diff line change
Expand Up @@ -719,22 +719,16 @@ func (cm *ConnManager) Run(ctx context.Context) {
}
}

// Shutdown the connection manager when the context is canceled.
cm.wg.Add(1)
go func(ctx context.Context, listeners []net.Listener) {
<-ctx.Done()
close(cm.quit)

// Stop all the listeners on shutdown. There will not be any
// listeners if listening is disabled.
for _, listener := range listeners {
// Ignore the error since this is shutdown and there is no way
// to recover anyways.
_ = listener.Close()
}

cm.wg.Done()
}(ctx, listeners)
// Stop all the listeners and shutdown the connection manager when the
// context is cancelled. There will not be any listeners if listening is
// disabled.
<-ctx.Done()
close(cm.quit)
for _, listener := range listeners {
// Ignore the error since this is shutdown and there is no way
// to recover anyways.
_ = listener.Close()
}

cm.wg.Wait()
log.Trace("Connection manager stopped")
Expand Down
32 changes: 15 additions & 17 deletions internal/blockchain/indexers/indexsubscriber.go
Original file line number Diff line number Diff line change
Expand Up @@ -367,22 +367,7 @@ func (s *IndexSubscriber) handleIndexUpdates(ctx context.Context) {
for {
select {
case <-ctx.Done():
close(s.quit)

// Stop all updates to subscribed indexes and terminate their
// processes.
s.mtx.Lock()
for _, sub := range s.subscriptions {
err := sub.stop()
if err != nil {
log.Error("unable to stop index subscription: %v", err)
}
}
s.mtx.Unlock()

s.cancel()
s.wg.Done()

return

case ntfn := <-s.c:
Expand Down Expand Up @@ -412,7 +397,20 @@ func (s *IndexSubscriber) Run(ctx context.Context) {
s.wg.Add(2)
go s.handleIndexUpdates(ctx)
go s.handleSyncSubscribers(ctx)
s.wg.Wait()

log.Infof("Index subscriber shutting down")
// Stop all the subscriptions and shutdown the subscriber when the context
// is cancelled.
<-ctx.Done()
close(s.quit)
s.mtx.Lock()
for _, sub := range s.subscriptions {
err := sub.stop()
if err != nil {
log.Errorf("unable to stop index subscription: %v", err)
}
}
s.mtx.Unlock()
s.cancel()
s.wg.Wait()
log.Tracef("Index subscriber stopped")
}
11 changes: 5 additions & 6 deletions internal/mining/bgblktmplgenerator.go
Original file line number Diff line number Diff line change
Expand Up @@ -1515,15 +1515,14 @@ func (g *BgBlkTmplGenerator) initialStartupHandler(ctx context.Context) {
// necessary for it to function properly and blocks until the provided context
// is cancelled.
func (g *BgBlkTmplGenerator) Run(ctx context.Context) {
g.wg.Add(5)
g.wg.Add(4)
go g.regenQueueHandler(ctx)
go g.regenHandler(ctx)
go g.notifySubscribersHandler(ctx)
go g.initialStartupHandler(ctx)
go func() {
<-ctx.Done()
close(g.quit)
g.wg.Done()
}()

// Shutdown the generator when the context is cancelled.
<-ctx.Done()
close(g.quit)
g.wg.Wait()
}
10 changes: 4 additions & 6 deletions internal/mining/cpuminer/cpuminer.go
Original file line number Diff line number Diff line change
Expand Up @@ -617,15 +617,13 @@ out:
func (m *CPUMiner) Run(ctx context.Context) {
log.Trace("Starting CPU miner in idle state")

m.wg.Add(3)
m.wg.Add(2)
go m.speedMonitor(ctx)
go m.miningWorkerController(ctx)
go func(ctx context.Context) {
<-ctx.Done()
close(m.quit)
m.wg.Done()
}(ctx)

// Shutdown the miner when the context is cancelled.
<-ctx.Done()
close(m.quit)
m.wg.Wait()
log.Trace("CPU miner stopped")
}
Expand Down
9 changes: 2 additions & 7 deletions internal/netsync/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -1796,13 +1796,8 @@ func (m *SyncManager) Run(ctx context.Context) {
go m.eventHandler(ctx)

// Shutdown the sync manager when the context is cancelled.
m.wg.Add(1)
go func(ctx context.Context) {
<-ctx.Done()
close(m.quit)
m.wg.Done()
}(ctx)

<-ctx.Done()
close(m.quit)
m.wg.Wait()
log.Trace("Sync manager stopped")
}
Expand Down
31 changes: 13 additions & 18 deletions internal/rpcserver/rpcwebsocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -1204,14 +1204,13 @@ func (m *wsNotificationManager) RemoveClient(wsc *wsClient) {
// websocket client notifications. It blocks until the provided context is
// cancelled.
func (m *wsNotificationManager) Run(ctx context.Context) {
m.wg.Add(3)
m.wg.Add(2)
go m.queueHandler(ctx)
go m.notificationHandler(ctx)
go func(ctx context.Context) {
<-ctx.Done()
close(m.quit)
m.wg.Done()
}(ctx)

// Shutdown the notification manager when the context is cancelled.
<-ctx.Done()
close(m.quit)
m.wg.Wait()
}

Expand Down Expand Up @@ -1929,18 +1928,14 @@ func (c *wsClient) Run(ctx context.Context) {
// Forcibly disconnect the websocket client when the context is cancelled
// which also closes the quit channel and thus ensures all of the above
// goroutines are shutdown.
c.wg.Add(1)
go func(ctx context.Context) {
// Select across the quit channel as well since the context is not
// cancelled when the connection is closed due to websocket connection
// hijacking.
select {
case <-ctx.Done():
c.Disconnect()
case <-c.quit:
}
c.wg.Done()
}(ctx)
//
// Select across the quit channel as well since the context is not cancelled
// when the connection is closed due to websocket connection hijacking.
select {
case <-ctx.Done():
c.Disconnect()
case <-c.quit:
}

c.wg.Wait()
}
Expand Down

0 comments on commit ae64e51

Please sign in to comment.