pool: Keep hub waitgroup local.

The waitgroup for the hub is currently being passed down into all
various configuration structs for each of the goroutines it runs which
requires a lot of plumbing and is quite error prone.

This addresses that by modifying the logic to keep the hub waitgroup
local to the run method and removes the aforementioned unnecessary
plumbing.

pool/chainstate.go

@@ -45,8 +45,6 @@ type ChainStateConfig struct {
 	Cancel context.CancelFunc
 	// SignalCache sends the provided cache update event to the gui cache.
 	SignalCache func(event CacheUpdateEvent)
-	// HubWg represents the hub's waitgroup.
-	HubWg *sync.WaitGroup
 }
 
 // blockNotification wraps a block header notification and a done channel.
@@ -201,7 +199,6 @@ func (cs *ChainState) handleChainUpdates(ctx context.Context) {
 		case <-ctx.Done():
 			close(cs.discCh)
 			close(cs.connCh)
-			cs.cfg.HubWg.Done()
 			return
 
 		case msg := <-cs.connCh:

pool/endpoint.go

@@ -36,8 +36,6 @@ type EndpointConfig struct {
 	MaxConnectionsPerHost uint32
 	// MaxGenTime represents the share creation target time for the pool.
 	MaxGenTime time.Duration
-	// HubWg represents the hub's waitgroup.
-	HubWg *sync.WaitGroup
 	// FetchMinerDifficulty returns the difficulty information for the
 	// provided miner if it exists.
 	FetchMinerDifficulty func(string) (*DifficultyInfo, error)
@@ -215,7 +213,6 @@ func (e *Endpoint) disconnect(ctx context.Context) {
 			e.clientsMtx.Unlock()
 
 			e.wg.Done()
-			e.cfg.HubWg.Done()
 			return
 
 		case <-e.discCh:

pool/hub.go

@@ -194,7 +194,7 @@ type Hub struct {
 	connectionsMtx sync.RWMutex
 	endpoint       *Endpoint
 	blake256Pad    []byte
-	wg             *sync.WaitGroup
+	wg             sync.WaitGroup
 	cacheCh        chan CacheUpdateEvent
 }
 
@@ -228,7 +228,6 @@ func NewHub(cancel context.CancelFunc, hcfg *HubConfig) (*Hub, error) {
 	h := &Hub{
 		cfg:         hcfg,
 		limiter:     NewRateLimiter(),
-		wg:          new(sync.WaitGroup),
 		connections: make(map[string]uint32),
 		cacheCh:     make(chan CacheUpdateEvent, bufferSize),
 	}
@@ -259,7 +258,6 @@ func NewHub(cancel context.CancelFunc, hcfg *HubConfig) (*Hub, error) {
 		FetchTxBroadcaster:    func() TxBroadcaster { return h.walletConn },
 		CoinbaseConfTimeout:   h.cfg.CoinbaseConfTimeout,
 		SignalCache:           h.SignalCache,
-		HubWg:                 h.wg,
 	}
 
 	var err error
@@ -277,7 +275,6 @@ func NewHub(cancel context.CancelFunc, hcfg *HubConfig) (*Hub, error) {
 		GetBlockConfirmations: h.getBlockConfirmations,
 		Cancel:                cancel,
 		SignalCache:           h.SignalCache,
-		HubWg:                 h.wg,
 	}
 	h.chainState = NewChainState(sCfg)
 
@@ -320,7 +317,6 @@ func NewHub(cancel context.CancelFunc, hcfg *HubConfig) (*Hub, error) {
 		Blake256Pad:           h.blake256Pad,
 		NonceIterations:       h.cfg.NonceIterations,
 		MaxConnectionsPerHost: h.cfg.MaxConnectionsPerHost,
-		HubWg:                 h.wg,
 		FetchMinerDifficulty:  h.poolDiffs.fetchMinerDifficulty,
 		SubmitWork:            h.submitWork,
 		FetchCurrentWork:      h.chainState.fetchCurrentWork,
@@ -632,9 +628,18 @@ func (h *Hub) shutdown() {
 // Run handles the process lifecycles of the pool hub.
 func (h *Hub) Run(ctx context.Context) {
 	h.wg.Add(3)
-	go h.endpoint.run(ctx)
-	go h.chainState.handleChainUpdates(ctx)
-	go h.paymentMgr.handlePayments(ctx)
+	go func(ctx context.Context) {
+		h.endpoint.run(ctx)
+		h.wg.Done()
+	}(ctx)
+	go func(ctx context.Context) {
+		h.chainState.handleChainUpdates(ctx)
+		h.wg.Done()
+	}(ctx)
+	go func(ctx context.Context) {
+		h.paymentMgr.handlePayments(ctx)
+		h.wg.Done()
+	}(ctx)
 
 	// Wait until all hub processes have terminated, and then shutdown.
 	h.wg.Wait()

pool/paymentmgr.go

@@ -117,8 +117,6 @@ type PaymentMgrConfig struct {
 	CoinbaseConfTimeout time.Duration
 	// SignalCache sends the provided cache update event to the gui cache.
 	SignalCache func(event CacheUpdateEvent)
-	// HubWg represents the hub's waitgroup.
-	HubWg *sync.WaitGroup
 }
 
 // paymentMsg represents a payment processing signal.
@@ -942,11 +940,11 @@ func (pm *PaymentMgr) processPayments(msg *paymentMsg) {
 }
 
 // handlePayments processes dividend payment signals.
+// This MUST be run as a goroutine.
 func (pm *PaymentMgr) handlePayments(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			pm.cfg.HubWg.Done()
 			return
 
 		case msg := <-pm.paymentCh: