Skip to content

Commit

Permalink
kvclient(ticdc): fix the workerpool data race (#10196) (#10201)
Browse files Browse the repository at this point in the history
ref #10095
  • Loading branch information
ti-chi-bot authored Feb 19, 2024
1 parent 5f05520 commit 2d9a053
Show file tree
Hide file tree
Showing 2 changed files with 9 additions and 4 deletions.
5 changes: 5 additions & 0 deletions cdc/kv/region_worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import (
var (
regionWorkerPool workerpool.WorkerPool
workerPoolOnce sync.Once
workerPoolLock sync.Mutex
// The magic number here is keep the same with some magic numbers in some
// other components in TiCDC, including worker pool task chan size, mounter
// chan size etc.
Expand Down Expand Up @@ -412,6 +413,8 @@ func (w *regionWorker) processEvent(ctx context.Context, event *regionStatefulEv

func (w *regionWorker) initPoolHandles() {
handles := make([]workerpool.EventHandle, 0, w.concurrency)
workerPoolLock.Lock()
defer workerPoolLock.Unlock()
for i := 0; i < w.concurrency; i++ {
poolHandle := regionWorkerPool.RegisterEvent(func(ctx context.Context, eventI interface{}) error {
event := eventI.(*regionStatefulEvent)
Expand Down Expand Up @@ -867,6 +870,8 @@ func getWorkerPoolSize() (size int) {
func InitWorkerPool() {
workerPoolOnce.Do(func() {
size := getWorkerPoolSize()
workerPoolLock.Lock()
defer workerPoolLock.Unlock()
regionWorkerPool = workerpool.NewDefaultWorkerPool(size)
})
}
Expand Down
8 changes: 4 additions & 4 deletions cdc/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -342,10 +342,6 @@ func (s *server) run(ctx context.Context) (err error) {

eg, egCtx := errgroup.WithContext(ctx)

eg.Go(func() error {
return s.capture.Run(egCtx)
})

eg.Go(func() error {
return s.upstreamPDHealthChecker(egCtx)
})
Expand All @@ -371,6 +367,10 @@ func (s *server) run(ctx context.Context) (err error) {
return nil
})

eg.Go(func() error {
return s.capture.Run(egCtx)
})

return eg.Wait()
}

Expand Down

0 comments on commit 2d9a053

Please sign in to comment.