all: use WaigGroup.Go() to simplify code

Author: gzliudan

accounts/keystore/keystore_test.go

@@ -406,14 +406,12 @@ func TestImportRace(t *testing.T) {
 	_, ks2 := tmpKeyStore(t)
 	var atom atomic.Uint32
 	var wg sync.WaitGroup
-	wg.Add(2)
 	for i := 0; i < 2; i++ {
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			if _, err := ks2.Import(json, "new", "new"); err != nil {
 				atom.Add(1)
 			}
-		}()
+		})
 	}
 	wg.Wait()
 	if atom.Load() != 1 {

bmt/bmt_test.go

@@ -263,23 +263,21 @@ func TestHasherConcurrency(t *testing.T) {
 	defer pool.Drain(0)
 	wg := sync.WaitGroup{}
 	cycles := 100
-	wg.Add(maxproccnt * cycles)
 	errc := make(chan error)
 
 	for p := 0; p < maxproccnt; p++ {
 		for i := 0; i < cycles; i++ {
-			go func() {
+			wg.Go(func() {
 				bmt := New(pool)
 				n := rand.Intn(4096)
 				tdata := testDataReader(n)
 				data := make([]byte, n)
 				tdata.Read(data)
 				err := testHasherCorrectness(bmt, hasher, data, n, 128)
-				wg.Done()
 				if err != nil {
 					errc <- err
 				}
-			}()
+			})
 		}
 	}
 	go func() {
@@ -386,18 +384,16 @@ func benchmarkBMTBaseline(n int, t *testing.B) {
 	for i := 0; i < t.N; i++ {
 		count := int32((n-1)/hasher().Size() + 1)
 		wg := sync.WaitGroup{}
-		wg.Add(maxproccnt)
 		var i int32
 		for j := 0; j < maxproccnt; j++ {
-			go func() {
-				defer wg.Done()
+			wg.Go(func() {
 				h := hasher()
 				for atomic.AddInt32(&i, 1) < count {
 					h.Reset()
 					h.Write(data)
 					h.Sum(nil)
 				}
-			}()
+			})
 		}
 		wg.Wait()
 	}
@@ -437,15 +433,13 @@ func benchmarkHasherReuse(poolsize, n int, t *testing.B) {
 	t.ResetTimer()
 	for i := 0; i < t.N; i++ {
 		wg := sync.WaitGroup{}
-		wg.Add(cycles)
 		for j := 0; j < cycles; j++ {
-			bmt := New(pool)
-			go func() {
-				defer wg.Done()
+			wg.Go(func() {
+				bmt := New(pool)
 				bmt.Reset()
 				bmt.Write(data)
 				bmt.Sum(nil)
-			}()
+			})
 		}
 		wg.Wait()
 	}

common/prque/lazyqueue_test.go

@@ -79,9 +79,7 @@ func TestLazyQueue(t *testing.T) {
 		stopCh = make(chan chan struct{})
 	)
 	defer wg.Wait()
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		for {
 			select {
 			case <-clock.After(testQueueRefresh):
@@ -92,7 +90,7 @@ func TestLazyQueue(t *testing.T) {
 				return
 			}
 		}
-	}()
+	})
 
 	for c := 0; c < testSteps; c++ {
 		i := rand.Intn(testItems)

consensus/XDPoS/engines/engine_v2/engine.go

@@ -757,16 +757,14 @@ func (x *XDPoS_v2) verifyQC(blockChainReader consensus.ChainReader, quorumCert *
 	start := time.Now()
 
 	var wg sync.WaitGroup
-	wg.Add(len(signatures))
 	sigErrChan := make(chan error, len(signatures))
 
 	for _, signature := range signatures {
-		go func(sig types.Signature) {
-			defer wg.Done()
+		wg.Go(func() {
 			verified, _, err := x.verifyMsgSignature(types.VoteSigHash(&types.VoteForSign{
 				ProposedBlockInfo: quorumCert.ProposedBlockInfo,
 				GapNumber:         quorumCert.GapNumber,
-			}), sig, epochInfo.Masternodes)
+			}), signature, epochInfo.Masternodes)
 			if err != nil {
 				log.Error("[verifyQC] Error while verfying QC message signatures", "Error", err)
 				sigErrChan <- errors.New("error while verfying QC message signatures")
@@ -777,7 +775,7 @@ func (x *XDPoS_v2) verifyQC(blockChainReader consensus.ChainReader, quorumCert *
 				sigErrChan <- errors.New("fail to verify QC due to signature mis-match")
 				return
 			}
-		}(signature)
+		})
 	}
 	wg.Wait()
 	elapsed := time.Since(start)

consensus/XDPoS/engines/engine_v2/syncInfo.go

@@ -128,13 +128,11 @@ func (x *XDPoS_v2) processSyncInfoPool(chain consensus.ChainReader) {
 
 func (x *XDPoS_v2) verifySignatures(messageHash common.Hash, signatures []types.Signature, candidates []common.Address) error {
 	var wg sync.WaitGroup
-	wg.Add(len(signatures))
 	var haveError error
 
 	for _, signature := range signatures {
-		go func(sig types.Signature) {
-			defer wg.Done()
-			verified, _, err := x.verifyMsgSignature(messageHash, sig, candidates)
+		wg.Go(func() {
+			verified, _, err := x.verifyMsgSignature(messageHash, signature, candidates)
 			if err != nil {
 				log.Error("[verifySignatures] Error while verfying QC message signatures", "error", err)
 				haveError = errors.New("error while verfying QC message signatures")
@@ -145,7 +143,7 @@ func (x *XDPoS_v2) verifySignatures(messageHash common.Hash, signatures []types.
 				haveError = errors.New("fail to verify QC due to signature mismatch")
 				return
 			}
-		}(signature)
+		})
 	}
 	wg.Wait()
 	if haveError != nil {

consensus/XDPoS/engines/engine_v2/timeout.go

@@ -185,8 +185,6 @@ func (x *XDPoS_v2) verifyTC(chain consensus.ChainReader, timeoutCert *types.Time
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(len(signatures))
-
 	var mutex sync.Mutex
 	var haveError error
 
@@ -196,11 +194,10 @@ func (x *XDPoS_v2) verifyTC(chain consensus.ChainReader, timeoutCert *types.Time
 	})
 
 	for _, signature := range signatures {
-		go func(sig types.Signature) {
-			defer wg.Done()
-			verified, _, err := x.verifyMsgSignature(signedTimeoutObj, sig, snap.NextEpochCandidates)
+		wg.Go(func() {
+			verified, _, err := x.verifyMsgSignature(signedTimeoutObj, signature, snap.NextEpochCandidates)
 			if err != nil || !verified {
-				log.Error("[verifyTC] Error or verification failure", "signature", sig, "error", err)
+				log.Error("[verifyTC] Error or verification failure", "signature", signature, "error", err)
 				mutex.Lock() // Lock before accessing haveError
 				if haveError == nil {
 					if err != nil {
@@ -213,7 +210,7 @@ func (x *XDPoS_v2) verifyTC(chain consensus.ChainReader, timeoutCert *types.Time
 				}
 				mutex.Unlock() // Unlock after modifying haveError
 			}
-		}(signature)
+		})
 	}
 	wg.Wait()
 	if haveError != nil {

consensus/XDPoS/engines/engine_v2/vote.go

@@ -164,10 +164,9 @@ func (x *XDPoS_v2) verifyVotes(chain consensus.ChainReader, votes map[common.Has
 	emptySigner := common.Address{}
 	// Filter out non-Master nodes signatures
 	var wg sync.WaitGroup
-	wg.Add(len(votes))
-	for h, vote := range votes {
-		go func(hash common.Hash, v *types.Vote) {
-			defer wg.Done()
+	for _, vote := range votes {
+		wg.Go(func() {
+			v := vote.(*types.Vote)
 			signerAddress := v.GetSigner()
 			if signerAddress != emptySigner {
 				// verify that signer belongs to the final masternodes, we have not do so in previous steps
@@ -199,7 +198,7 @@ func (x *XDPoS_v2) verifyVotes(chain consensus.ChainReader, votes map[common.Has
 				return
 			}
 			v.SetSigner(masterNode)
-		}(h, vote.(*types.Vote))
+		})
 	}
 	wg.Wait()
 	elapsed := time.Since(start)

console/console.go

@@ -119,8 +119,7 @@ func New(config Config) (*Console, error) {
 		return nil, err
 	}
 
-	console.wg.Add(1)
-	go console.interruptHandler()
+	console.wg.Go(console.interruptHandler)
 
 	return console, nil
 }
@@ -341,8 +340,6 @@ func (c *Console) Evaluate(statement string) {
 // interruptHandler runs in its own goroutine and waits for signals.
 // When a signal is received, it interrupts the JS interpreter.
 func (c *Console) interruptHandler() {
-	defer c.wg.Done()
-
 	// During Interactive, liner inhibits the signal while it is prompting for
 	// input. However, the signal will be received while evaluating JS.
 	//

core/blockchain.go

@@ -327,8 +327,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	}
 
 	// Start future block processor.
-	bc.wg.Add(1)
-	go bc.futureBlocksLoop()
+	bc.wg.Go(bc.futureBlocksLoop)
 
 	return bc, nil
 }
@@ -2572,8 +2571,6 @@ func (bc *BlockChain) PostChainEvents(events []interface{}, logs []*types.Log) {
 
 // futureBlocksLoop processes the 'future block' queue.
 func (bc *BlockChain) futureBlocksLoop() {
-	defer bc.wg.Done()
-
 	futureTimer := time.NewTicker(100 * time.Millisecond)
 	defer futureTimer.Stop()
 

core/bloombits/scheduler.go

@@ -62,9 +62,8 @@ func (s *scheduler) run(sections chan uint64, dist chan *request, done chan []by
 	pend := make(chan uint64, cap(dist))
 
 	// Start the pipeline schedulers to forward between user -> distributor -> user
-	wg.Add(2)
-	go s.scheduleRequests(sections, dist, pend, quit, wg)
-	go s.scheduleDeliveries(pend, done, quit, wg)
+	wg.Go(func() { s.scheduleRequests(sections, dist, pend, quit) })
+	wg.Go(func() { s.scheduleDeliveries(pend, done, quit) })
 }
 
 // reset cleans up any leftovers from previous runs. This is required before a
@@ -84,9 +83,8 @@ func (s *scheduler) reset() {
 // scheduleRequests reads section retrieval requests from the input channel,
 // deduplicates the stream and pushes unique retrieval tasks into the distribution
 // channel for a database or network layer to honour.
-func (s *scheduler) scheduleRequests(reqs chan uint64, dist chan *request, pend chan uint64, quit chan struct{}, wg *sync.WaitGroup) {
+func (s *scheduler) scheduleRequests(reqs chan uint64, dist chan *request, pend chan uint64, quit chan struct{}) {
 	// Clean up the goroutine and pipeline when done
-	defer wg.Done()
 	defer close(pend)
 
 	// Keep reading and scheduling section requests
@@ -131,9 +129,8 @@ func (s *scheduler) scheduleRequests(reqs chan uint64, dist chan *request, pend
 
 // scheduleDeliveries reads section acceptance notifications and waits for them
 // to be delivered, pushing them into the output data buffer.
-func (s *scheduler) scheduleDeliveries(pend chan uint64, done chan []byte, quit chan struct{}, wg *sync.WaitGroup) {
+func (s *scheduler) scheduleDeliveries(pend chan uint64, done chan []byte, quit chan struct{}) {
 	// Clean up the goroutine and pipeline when done
-	defer wg.Done()
 	defer close(done)
 
 	// Keep reading notifications and scheduling deliveries