fix region worker run once

pingcap · Mar 5, 2021 · c40f154 · c40f154
1 parent be8fa25
commit c40f154
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 15 deletions.
diff --git a/cdc/kv/client_bench_test.go b/cdc/kv/client_bench_test.go
@@ -394,6 +394,11 @@ func BenchmarkResolvedTsClientV1(b *testing.B) {
 }
 
 func BenchmarkResolvedTsClientV2(b *testing.B) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		RunWorkerPool(ctx) //nolint:errcheck
+	}()
 	benchmarkSingleWorkerResolvedTs(b, true /* clientV2 */)
 }
 
@@ -494,5 +499,10 @@ func BenchmarkMultiStoreResolvedTsClientV1(b *testing.B) {
 }
 
 func BenchmarkMultiStoreResolvedTsClientV2(b *testing.B) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		RunWorkerPool(ctx) //nolint:errcheck
+	}()
 	benchmarkMultipleStoreResolvedTs(b, true /* clientV2 */)
 }
diff --git a/cdc/kv/client_test.go b/cdc/kv/client_test.go
@@ -43,14 +43,10 @@ import (
 	"google.golang.org/grpc"
 )
 
-var kvClientTestOnce sync.Once
-
 func Test(t *testing.T) {
-	kvClientTestOnce.Do(func() {
-		go func() {
-			RunWorkerPool(context.Background()) //nolint:errcheck
-		}()
-	})
+	go func() {
+		RunWorkerPool(context.Background()) //nolint:errcheck
+	}()
 	check.TestingT(t)
 }
 

diff --git a/cdc/kv/region_worker.go b/cdc/kv/region_worker.go
@@ -36,7 +36,10 @@ import (
 	"golang.org/x/time/rate"
 )
 
-var regionWorkerPool workerpool.WorkerPool
+var (
+	regionWorkerPool workerpool.WorkerPool
+	workerPoolOnce   sync.Once
+)
 
 const (
 	minRegionStateBucket = 4
@@ -563,12 +566,16 @@ func RunWorkerPool(ctx context.Context) error {
 	if !enableKVClientV2 {
 		return nil
 	}
-	// TODO: make the pool size configurable
-	size := runtime.NumCPU()
-	regionWorkerPool = workerpool.NewDefaultWorkerPool(size)
-	errg, ctx := errgroup.WithContext(ctx)
-	errg.Go(func() error {
-		return errors.Trace(regionWorkerPool.Run(ctx))
+	var err error
+	workerPoolOnce.Do(func() {
+		// TODO: make the pool size configurable
+		size := runtime.NumCPU()
+		regionWorkerPool = workerpool.NewDefaultWorkerPool(size)
+		errg, ctx := errgroup.WithContext(ctx)
+		errg.Go(func() error {
+			return errors.Trace(regionWorkerPool.Run(ctx))
+		})
+		err = errg.Wait()
 	})
-	return errors.Trace(errg.Wait())
+	return err
 }