all: use WaigGroup.Go() to simplify code (XinFinOrg#1699)

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/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

@@ -334,8 +334,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
 }
@@ -2580,8 +2579,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

core/state/statedb_test.go

@@ -182,16 +182,10 @@ func TestCopy(t *testing.T) {
 	}
 
 	// Finalise the changes on all concurrently
-	finalise := func(wg *sync.WaitGroup, db *StateDB) {
-		defer wg.Done()
-		db.Finalise(true)
-	}
-
 	var wg sync.WaitGroup
-	wg.Add(3)
-	go finalise(&wg, orig)
-	go finalise(&wg, copy)
-	go finalise(&wg, ccopy)
+	wg.Go(func() { orig.Finalise(true) })
+	wg.Go(func() { copy.Finalise(true) })
+	wg.Go(func() { ccopy.Finalise(true) })
 	wg.Wait()
 
 	// Verify that the three states have been updated independently

core/state_processor.go

@@ -585,19 +585,17 @@ func InitSignerInTransactions(config *params.ChainConfig, header *types.Header,
 		chunkSize++
 	}
 	wg := sync.WaitGroup{}
-	wg.Add(nWorker)
 	for i := 0; i < nWorker; i++ {
 		from := i * chunkSize
 		to := from + chunkSize
 		if to > txs.Len() {
 			to = txs.Len()
 		}
-		go func(from int, to int) {
+		wg.Go(func() {
 			for j := from; j < to; j++ {
 				types.CacheSigner(signer, txs[j])
 			}
-			wg.Done()
-		}(from, to)
+		})
 	}
 	wg.Wait()
 }

core/txpool/lending_pool.go

@@ -182,8 +182,7 @@ func NewLendingPool(chainconfig *params.ChainConfig, chain blockChainLending) *L
 	pool.chainHeadSub = pool.chain.SubscribeChainHeadEvent(pool.chainHeadCh)
 
 	// Start the event loop and return
-	pool.wg.Add(1)
-	go pool.loop()
+	pool.wg.Go(pool.loop)
 
 	return pool
 }
@@ -192,8 +191,6 @@ func NewLendingPool(chainconfig *params.ChainConfig, chain blockChainLending) *L
 // outside blockchain events as well as for various reporting and transaction
 // eviction events.
 func (pool *LendingPool) loop() {
-	defer pool.wg.Done()
-
 	// Start the stats reporting and transaction eviction tickers
 	var prevPending, prevQueued int