all: use sync.WaitGroup.Go

Drops the Add/Done bookkeeping, which in batcher.go was spread across
three functions.
This commit is contained in:
Kristoffer Dalby
2026-08-25 09:43:47 +00:00
committed by Kristoffer Dalby
parent 245b81401c
commit 373c60efe3
9 changed files with 46 additions and 106 deletions
+2 -10
View File
@@ -403,9 +403,7 @@ func (b *Batcher) Start() {
return return
} }
b.wg.Add(1) b.wg.Go(b.doWork)
go b.doWork()
} }
func (b *Batcher) Close() { func (b *Batcher) Close() {
@@ -439,12 +437,8 @@ func (b *Batcher) Close() {
} }
func (b *Batcher) doWork() { func (b *Batcher) doWork() {
defer b.wg.Done()
for i := range b.workers { for i := range b.workers {
b.wg.Add(1) b.wg.Go(func() { b.worker(i + 1) })
go b.worker(i + 1)
} }
// Create a cleanup ticker for removing truly disconnected nodes // Create a cleanup ticker for removing truly disconnected nodes
@@ -467,8 +461,6 @@ func (b *Batcher) doWork() {
} }
func (b *Batcher) worker(workerID int) { func (b *Batcher) worker(workerID int) {
defer b.wg.Done()
wlog := log.With().Int(zf.WorkerID, workerID).Logger() wlog := log.With().Int(zf.WorkerID, workerID).Logger()
for { for {
+4 -7
View File
@@ -142,19 +142,16 @@ func runConcurrently(t *testing.T, n int, fn func(i int)) int {
) )
for i := range n { for i := range n {
wg.Add(1) wg.Go(func() {
go func(idx int) {
defer wg.Done()
defer func() { defer func() {
if r := recover(); r != nil { if r := recover(); r != nil {
panics.Add(1) panics.Add(1)
t.Logf("panic in goroutine %d: %v", idx, r) t.Logf("panic in goroutine %d: %v", i, r)
} }
}() }()
fn(idx) fn(i)
}(i) })
} }
wg.Wait() wg.Wait()
+4 -10
View File
@@ -96,22 +96,16 @@ func TestConnectDisconnectRace(t *testing.T) {
start := make(chan struct{}) start := make(chan struct{})
wg.Add(2) wg.Go(func() {
go func() {
defer wg.Done()
<-start <-start
_, _ = srv.State().Disconnect(r2ID, gen) _, _ = srv.State().Disconnect(r2ID, gen)
}() })
go func() { wg.Go(func() {
defer wg.Done()
<-start <-start
_, _ = srv.State().Connect(r2ID) _, _ = srv.State().Connect(r2ID)
}() })
close(start) close(start)
wg.Wait() wg.Wait()
+2 -6
View File
@@ -494,16 +494,12 @@ func TestRaceBatcherContention(t *testing.T) {
srv.App.Change(c) srv.App.Change(c)
}) })
wg.Add(1)
var c2 *servertest.TestClient var c2 *servertest.TestClient
go func() { wg.Go(func() {
defer wg.Done()
c2 = servertest.NewClient(t, srv, "rtcon-node2", c2 = servertest.NewClient(t, srv, "rtcon-node2",
servertest.WithUser(user)) servertest.WithUser(user))
}() })
wg.Wait() wg.Wait()
+3 -7
View File
@@ -104,17 +104,13 @@ func TestConcurrentPutNodeSameGivenNameAllUnique(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
results := make(chan string, N) results := make(chan string, N)
for i := range N { for id := range N {
wg.Add(1) wg.Go(func() {
go func(id int) {
defer wg.Done()
n := createTestNode(types.NodeID(id+1), 1, "alice", "laptop") //nolint:gosec // test ids n := createTestNode(types.NodeID(id+1), 1, "alice", "laptop") //nolint:gosec // test ids
view := store.PutNode(n) view := store.PutNode(n)
results <- view.GivenName() results <- view.GivenName()
}(i) })
} }
wg.Wait() wg.Wait()
+17 -37
View File
@@ -901,17 +901,13 @@ func TestNodeStoreConcurrentPutNode(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
results := make(chan bool, concurrentOps) results := make(chan bool, concurrentOps)
for i := range concurrentOps { for nodeID := 1; nodeID <= concurrentOps; nodeID++ {
wg.Add(1) wg.Go(func() {
go func(nodeID int) {
defer wg.Done()
node := createConcurrentTestNode(types.NodeID(nodeID), "concurrent-node") //nolint:gosec // safe conversion in test node := createConcurrentTestNode(types.NodeID(nodeID), "concurrent-node") //nolint:gosec // safe conversion in test
resultNode := store.PutNode(node) resultNode := store.PutNode(node)
results <- resultNode.Valid() results <- resultNode.Valid()
}(i + 1) })
} }
wg.Wait() wg.Wait()
@@ -940,17 +936,13 @@ func TestNodeStoreBatchingEfficiency(t *testing.T) {
var wg sync.WaitGroup var wg sync.WaitGroup
results := make(chan bool, ops) results := make(chan bool, ops)
for i := range ops { for nodeID := 1; nodeID <= ops; nodeID++ {
wg.Add(1) wg.Go(func() {
go func(nodeID int) {
defer wg.Done()
node := createConcurrentTestNode(types.NodeID(nodeID), "batch-node") //nolint:gosec // test code with small integers node := createConcurrentTestNode(types.NodeID(nodeID), "batch-node") //nolint:gosec // test code with small integers
resultNode := store.PutNode(node) resultNode := store.PutNode(node)
results <- resultNode.Valid() results <- resultNode.Valid()
}(i + 1) })
} }
wg.Wait() wg.Wait()
@@ -988,12 +980,8 @@ func TestNodeStoreRaceConditions(t *testing.T) {
errors := make(chan error, numGoroutines*opsPerGoroutine) errors := make(chan error, numGoroutines*opsPerGoroutine)
for i := range numGoroutines { for gid := range numGoroutines {
wg.Add(1) wg.Go(func() {
go func(gid int) {
defer wg.Done()
for j := range opsPerGoroutine { for j := range opsPerGoroutine {
switch j % 3 { switch j % 3 {
case 0: case 0:
@@ -1017,7 +1005,7 @@ func TestNodeStoreRaceConditions(t *testing.T) {
} }
} }
} }
}(i) })
} }
wg.Wait() wg.Wait()
@@ -1097,14 +1085,10 @@ func TestNodeStoreOperationTimeout(t *testing.T) {
updateResults := make([]error, ops) updateResults := make([]error, ops)
// Launch all PutNode operations concurrently // Launch all PutNode operations concurrently
for i := 1; i <= ops; i++ { for idx := 1; idx <= ops; idx++ {
nodeID := types.NodeID(i) //nolint:gosec // test code with small integers id := types.NodeID(idx) //nolint:gosec // test code with small integers
wg.Add(1)
go func(idx int, id types.NodeID) {
defer wg.Done()
wg.Go(func() {
startPut := time.Now() startPut := time.Now()
fmt.Printf("[TestNodeStoreOperationTimeout] %s: PutNode(%d) starting\n", startPut.Format("15:04:05.000"), id) fmt.Printf("[TestNodeStoreOperationTimeout] %s: PutNode(%d) starting\n", startPut.Format("15:04:05.000"), id)
node := createConcurrentTestNode(id, "timeout-node") node := createConcurrentTestNode(id, "timeout-node")
@@ -1115,7 +1099,7 @@ func TestNodeStoreOperationTimeout(t *testing.T) {
if !resultNode.Valid() { if !resultNode.Valid() {
putResults[idx-1] = fmt.Errorf("PutNode failed for node %d", id) //nolint:err113 putResults[idx-1] = fmt.Errorf("PutNode failed for node %d", id) //nolint:err113
} }
}(i, nodeID) })
} }
wg.Wait() wg.Wait()
@@ -1123,14 +1107,10 @@ func TestNodeStoreOperationTimeout(t *testing.T) {
// Launch all UpdateNode operations concurrently // Launch all UpdateNode operations concurrently
wg = sync.WaitGroup{} wg = sync.WaitGroup{}
for i := 1; i <= ops; i++ { for idx := 1; idx <= ops; idx++ {
nodeID := types.NodeID(i) //nolint:gosec // test code with small integers id := types.NodeID(idx) //nolint:gosec // test code with small integers
wg.Add(1)
go func(idx int, id types.NodeID) {
defer wg.Done()
wg.Go(func() {
startUpdate := time.Now() startUpdate := time.Now()
fmt.Printf("[TestNodeStoreOperationTimeout] %s: UpdateNode(%d) starting\n", startUpdate.Format("15:04:05.000"), id) fmt.Printf("[TestNodeStoreOperationTimeout] %s: UpdateNode(%d) starting\n", startUpdate.Format("15:04:05.000"), id)
resultNode, ok := store.UpdateNode(id, func(n *types.Node) { resultNode, ok := store.UpdateNode(id, func(n *types.Node) {
@@ -1142,7 +1122,7 @@ func TestNodeStoreOperationTimeout(t *testing.T) {
if !ok || !resultNode.Valid() { if !ok || !resultNode.Valid() {
updateResults[idx-1] = fmt.Errorf("UpdateNode failed for node %d", id) //nolint:err113 updateResults[idx-1] = fmt.Errorf("UpdateNode failed for node %d", id) //nolint:err113
} }
}(i, nodeID) })
} }
done := make(chan struct{}) done := make(chan struct{})
+5 -8
View File
@@ -1,6 +1,7 @@
package state package state
import ( import (
"slices"
"sync" "sync"
"testing" "testing"
"time" "time"
@@ -79,14 +80,10 @@ func TestPingTracker_ConcurrentDifferentIDs(t *testing.T) {
// Complete in reverse order concurrently. // Complete in reverse order concurrently.
var wg sync.WaitGroup var wg sync.WaitGroup
for i := count - 1; i >= 0; i-- { for _, id := range slices.Backward(ids) {
wg.Add(1) wg.Go(func() {
assert.True(t, pt.complete(id))
go func(idx int) { })
defer wg.Done()
assert.True(t, pt.complete(ids[idx]))
}(i)
} }
// All channels should receive. // All channels should receive.
+4 -10
View File
@@ -85,19 +85,13 @@ func TestSSHCheckAuthConcurrent(t *testing.T) {
wg.Wait() wg.Wait()
// Clear concurrently with reads // Clear concurrently with reads
wg.Add(2) wg.Go(func() {
go func() {
defer wg.Done()
s.ClearSSHCheckAuth() s.ClearSSHCheckAuth()
}() })
go func() {
defer wg.Done()
wg.Go(func() {
s.GetLastSSHAuth(types.NodeID(1), types.NodeID(2)) s.GetLastSSHAuth(types.NodeID(1), types.NodeID(2))
}() })
wg.Wait() wg.Wait()
} }
+5 -11
View File
@@ -735,17 +735,11 @@ func assertClientsState(t *testing.T, clients []TailscaleClient) {
var wg sync.WaitGroup var wg sync.WaitGroup
for _, client := range clients { for _, client := range clients {
wg.Add(1) wg.Go(func() {
assertValidStatus(t, client)
c := client // Avoid loop pointer assertValidNetcheck(t, client)
assertValidNetmap(t, client)
go func() { })
defer wg.Done()
assertValidStatus(t, c)
assertValidNetcheck(t, c)
assertValidNetmap(t, c)
}()
} }
t.Logf("waiting for client state checks to finish") t.Logf("waiting for client state checks to finish")