*: use (*sync.WaitGroup).Go

Update uses of WaitGroup to use the new (Go 1.25+) Go method to prevent
accidental misuse and simplify call sites.

Author: jbowens

checkpoint_test.go

@@ -291,35 +291,31 @@ func TestCheckpointCompaction(t *testing.T) {
 	d, err := Open("", &Options{FS: fs, Logger: testutils.Logger{T: t}})
 	require.NoError(t, err)
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(t.Context())
 
 	var wg sync.WaitGroup
-	wg.Add(4)
-	go func() {
+	wg.Go(func() {
 		defer cancel()
-		defer wg.Done()
 		for i := 0; ctx.Err() == nil; i++ {
 			if err := d.Set([]byte(fmt.Sprintf("key%06d", i)), nil, nil); err != nil {
 				t.Error(err)
 				return
 			}
 		}
-	}()
-	go func() {
+	})
+	wg.Go(func() {
 		defer cancel()
-		defer wg.Done()
 		for ctx.Err() == nil {
-			if err := d.Compact(context.Background(), []byte("key"), []byte("key999999"), false); err != nil {
+			if err := d.Compact(t.Context(), []byte("key"), []byte("key999999"), false); err != nil {
 				t.Error(err)
 				return
 			}
 		}
-	}()
+	})
 	check := make(chan string, 100)
-	go func() {
+	wg.Go(func() {
 		defer cancel()
 		defer close(check)
-		defer wg.Done()
 		for i := 0; ctx.Err() == nil && i < 200; i++ {
 			dir := fmt.Sprintf("checkpoint%06d", i)
 			if err := d.Checkpoint(dir); err != nil {
@@ -332,11 +328,10 @@ func TestCheckpointCompaction(t *testing.T) {
 			case check <- dir:
 			}
 		}
-	}()
-	go func() {
+	})
+	wg.Go(func() {
 		opts := &Options{FS: fs, Logger: testutils.Logger{T: t}}
 		defer cancel()
-		defer wg.Done()
 		for dir := range check {
 			d2, err := Open(dir, opts)
 			if err != nil {
@@ -362,7 +357,7 @@ func TestCheckpointCompaction(t *testing.T) {
 				return
 			}
 		}
-	}()
+	})
 	<-ctx.Done()
 	wg.Wait()
 	require.NoError(t, d.Close())

cmd/pebble/fsbench.go

@@ -612,13 +612,10 @@ func (bench *fsBench) init(wg *sync.WaitGroup) {
 	fmt.Println("Running benchmark:", bench.name)
 	fmt.Println("Description:", bench.description)
 
-	wg.Add(1)
-	go bench.execute(wg)
+	wg.Go(bench.execute)
 }
 
-func (bench *fsBench) execute(wg *sync.WaitGroup) {
-	defer wg.Done()
-
+func (bench *fsBench) execute() {
 	latencyHist := bench.reg.Register(bench.name)
 
 	for {

cmd/pebble/queue.go

@@ -62,10 +62,7 @@ func queueTest() (test, *atomic.Int64) {
 			}
 
 			limiter := maxOpsPerSec.newRateLimiter()
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-
+			wg.Go(func() {
 				for i := queueConfig.size; ; i++ {
 					idx := i % queueConfig.size
 
@@ -94,7 +91,7 @@ func queueTest() (test, *atomic.Int64) {
 					wait(limiter)
 					ops.Add(1)
 				}
-			}()
+			})
 		},
 		tick: func(elapsed time.Duration, i int) {
 			if i%20 == 0 {

cmd/pebble/scan.go

@@ -90,11 +90,8 @@ func runScan(cmd *cobra.Command, args []string) {
 
 			limiter := maxOpsPerSec.newRateLimiter()
 
-			wg.Add(concurrency)
-			for i := 0; i < concurrency; i++ {
-				go func(i int) {
-					defer wg.Done()
-
+			for i := range concurrency {
+				wg.Go(func() {
 					rng := rand.New(rand.NewPCG(0, uint64(i)))
 					startKeyBuf := append(make([]byte, 0, 64), []byte("key-")...)
 					endKeyBuf := append(make([]byte, 0, 64), []byte("key-")...)
@@ -123,7 +120,7 @@ func runScan(cmd *cobra.Command, args []string) {
 						bytes.Add(nbytes)
 						scanned.Add(int64(count))
 					}
-				}(i)
+				})
 			}
 		},
 

cmd/pebble/sync.go

@@ -61,12 +61,9 @@ func runSync(cmd *cobra.Command, args []string) {
 		init: func(d DB, wg *sync.WaitGroup) {
 			limiter := maxOpsPerSec.newRateLimiter()
 
-			wg.Add(concurrency)
-			for i := 0; i < concurrency; i++ {
+			for range concurrency {
 				latency := reg.Register("ops")
-				go func() {
-					defer wg.Done()
-
+				wg.Go(func() {
 					rand := rand.New(rand.NewPCG(0, uint64(time.Now().UnixNano())))
 					var raw []byte
 					var buf []byte
@@ -78,7 +75,7 @@ func runSync(cmd *cobra.Command, args []string) {
 						b := d.NewBatch()
 						var n uint64
 						count := int(batchDist.Uint64(rand))
-						for j := 0; j < count; j++ {
+						for range count {
 							block = syncConfig.values.Bytes(rand, block)
 
 							if syncConfig.walOnly {
@@ -101,7 +98,7 @@ func runSync(cmd *cobra.Command, args []string) {
 						latency.Record(time.Since(start))
 						bytes.Add(n)
 					}
-				}()
+				})
 			}
 		},
 

cmd/pebble/write_bench.go

@@ -224,11 +224,7 @@ func runWriteBenchmark(_ *cobra.Command, args []string) error {
 			// take a material amount of time. Instead, pause the
 			// writers in parallel in the background, and wait for all
 			// to complete before continuing.
-			wg.Add(1)
-			go func(writer *pauseWriter) {
-				writer.pause()
-				wg.Done()
-			}(w)
+			wg.Go(func() { w.pause() })
 		}
 		wg.Wait()
 

cmd/pebble/ycsb.go

@@ -365,23 +365,18 @@ func (y *ycsb) init(db DB, wg *sync.WaitGroup) {
 
 	y.limiter = maxOpsPerSec.newRateLimiter()
 
-	wg.Add(concurrency)
-
 	// If this workload doesn't produce reads, sample the worst case read-amp
 	// from Metrics() periodically.
 	if y.weights.get(ycsbRead) == 0 && y.weights.get(ycsbScan) == 0 && y.weights.get(ycsbReverseScan) == 0 {
-		wg.Add(1)
-		go y.sampleReadAmp(db, wg)
+		wg.Go(func() { y.sampleReadAmp(db) })
 	}
 
-	for i := 0; i < concurrency; i++ {
-		go y.run(db, wg)
+	for range concurrency {
+		wg.Go(func() { y.run(db) })
 	}
 }
 
-func (y *ycsb) run(db DB, wg *sync.WaitGroup) {
-	defer wg.Done()
-
+func (y *ycsb) run(db DB) {
 	var latency [ycsbNumOps]*namedHistogram
 	for name, op := range y.opsMap {
 		latency[op] = y.reg.Register(name)
@@ -418,9 +413,7 @@ func (y *ycsb) run(db DB, wg *sync.WaitGroup) {
 	}
 }
 
-func (y *ycsb) sampleReadAmp(db DB, wg *sync.WaitGroup) {
-	defer wg.Done()
-
+func (y *ycsb) sampleReadAmp(db DB) {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 	for range ticker.C {

commit_test.go

@@ -98,14 +98,12 @@ func TestCommitPipeline(t *testing.T) {
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(n)
-	for i := 0; i < n; i++ {
-		go func(i int) {
-			defer wg.Done()
+	for i := range n {
+		wg.Go(func() {
 			var b Batch
 			_ = b.Set([]byte(fmt.Sprint(i)), nil, nil)
 			_ = p.Commit(&b, false, false)
-		}(i)
+		})
 	}
 	wg.Wait()
 
@@ -141,17 +139,15 @@ func TestCommitPipelineSync(t *testing.T) {
 			e.queueSemChan = p.logSyncQSem
 
 			var wg sync.WaitGroup
-			wg.Add(n)
-			for i := 0; i < n; i++ {
-				go func(i int) {
-					defer wg.Done()
+			for i := range n {
+				wg.Go(func() {
 					var b Batch
 					require.NoError(t, b.Set([]byte(fmt.Sprint(i)), nil, nil))
 					require.NoError(t, p.Commit(&b, true, noSyncWait))
 					if noSyncWait {
 						require.NoError(t, b.SyncWait())
 					}
-				}(i)
+				})
 			}
 			wg.Wait()
 			if s := e.writeCount.Load(); uint64(n) != s {
@@ -340,19 +336,16 @@ func TestCommitPipelineLogDataSeqNum(t *testing.T) {
 	p := newCommitPipeline(testEnv)
 
 	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		b := &Batch{}
 		require.NoError(t, b.Set([]byte("foo"), []byte("bar"), nil))
 		require.NoError(t, p.Commit(b, false /* sync */, false))
-	}()
-	go func() {
-		defer wg.Done()
+	})
+	wg.Go(func() {
 		b := &Batch{}
 		require.NoError(t, b.LogData([]byte("foo"), nil))
 		require.NoError(t, p.Commit(b, false /* sync */, false))
-	}()
+	})
 	wg.Wait()
 }
 

compaction_test.go

@@ -3412,9 +3412,7 @@ func TestCompactionCorruption(t *testing.T) {
 	startWorkload := func(minKey, maxKey byte) (stop func()) {
 		var shouldStop atomic.Bool
 		var wg sync.WaitGroup
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			var valSeed [32]byte
 			for i := range valSeed {
 				valSeed[i] = byte(rand.Uint32())
@@ -3443,7 +3441,7 @@ func TestCompactionCorruption(t *testing.T) {
 					panic(err)
 				}
 			}
-		}()
+		})
 		return func() {
 			shouldStop.Store(true)
 			wg.Wait()

db_test.go

@@ -1161,10 +1161,8 @@ func TestDBConcurrentCommitCompactFlush(t *testing.T) {
 	// those operations.
 	const n = 1000
 	var wg sync.WaitGroup
-	wg.Add(n)
-	for i := 0; i < n; i++ {
-		go func(i int) {
-			defer wg.Done()
+	for i := range n {
+		wg.Go(func() {
 			_ = d.Set([]byte(fmt.Sprint(i)), nil, nil)
 			var err error
 			switch i % 3 {
@@ -1176,7 +1174,7 @@ func TestDBConcurrentCommitCompactFlush(t *testing.T) {
 				_, err = d.AsyncFlush()
 			}
 			require.NoError(t, err)
-		}(i)
+		})
 	}
 	wg.Wait()
 
@@ -2495,12 +2493,10 @@ type parallel []orderingNode
 
 func (p parallel) visit(fn func(int)) {
 	var wg sync.WaitGroup
-	wg.Add(len(p))
 	for i := range p {
-		go func(i int) {
-			defer wg.Done()
+		wg.Go(func() {
 			p[i].visit(fn)
-		}(i)
+		})
 	}
 	wg.Wait()
 }
@@ -2617,7 +2613,7 @@ func TestLoadBlockSema(t *testing.T) {
 	}
 
 	// Read all regions to warm up the file cache.
-	for i := 0; i < numRegions; i++ {
+	for i := range numRegions {
 		val, closer, err := db.Get(key(i, 1))
 		require.NoError(t, err)
 		require.Equal(t, []byte("value"), val)
@@ -2633,12 +2629,10 @@ func TestLoadBlockSema(t *testing.T) {
 			var wg sync.WaitGroup
 			// Spin up workers that perform random reads.
 			const numWorkers = 20
-			for i := 0; i < numWorkers; i++ {
-				wg.Add(1)
-				go func() {
-					defer wg.Done()
+			for range numWorkers {
+				wg.Go(func() {
 					const numQueries = 100
-					for i := 0; i < numQueries; i++ {
+					for range numQueries {
 						val, closer, err := db.Get(key(rand.IntN(numRegions), rand.IntN(numKeys)))
 						require.NoError(t, err)
 						require.Equal(t, []byte("value"), val)
@@ -2647,7 +2641,7 @@ func TestLoadBlockSema(t *testing.T) {
 						}
 						runtime.Gosched()
 					}
-				}()
+				})
 			}
 			wg.Wait()
 			// Verify the maximum read count did not exceed the limit.