From 0ab3af0eb208804f81a16443933716b9cee73f32 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 19:03:40 +0800 Subject: [PATCH 1/4] logpuller: make paused event pushing interruptible --- logservice/logpuller/region_event_sink.go | 112 ++++++++++++------ .../logpuller/region_event_sink_test.go | 66 +++++++++-- .../logpuller/subscription_client_test.go | 5 +- 3 files changed, 137 insertions(+), 46 deletions(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 3fb1771461..cf0a5f7666 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -26,16 +26,19 @@ import ( // regionEventSink delivers region events to dynstream and owns push-side flow control. type regionEventSink struct { - ctx context.Context - ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] - // the following three fields are used to manage feedback from ds and notify other goroutines - mu sync.Mutex - cond *sync.Cond - paused atomic.Bool + ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] + // the following fields are used to coordinate pause/resume feedback with event producers. + mu sync.Mutex + resumeCh chan struct{} + stopCh chan struct{} + stopOnce sync.Once + paused atomic.Bool } -func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandler) *regionEventSink { - sink := ®ionEventSink{ctx: ctx} +func newRegionEventSink(_ context.Context, failureHandler *regionFailureHandler) *regionEventSink { + sink := ®ionEventSink{ + stopCh: make(chan struct{}), + } option := dynstream.NewOption() // Note: it is max batch size of the kv sent from tikv(not committed rows) @@ -51,7 +54,6 @@ func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandle ) ds.Start() sink.ds = ds - sink.cond = sync.NewCond(&sink.mu) return sink } @@ -73,44 +75,44 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { } func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { - // fast path - if !s.paused.Load() { - s.ds.Push(subID, event) - return - } - - // slow path: wait until paused is false - s.mu.Lock() - for s.paused.Load() { + for { + // fast path select { - case <-s.ctx.Done(): - s.mu.Unlock() + case <-s.stopCh: return default: - s.cond.Wait() + } + if !s.paused.Load() { + s.ds.Push(subID, event) + return + } + + resumeCh := s.getResumeCh() + if resumeCh == nil { + continue + } + + select { + case <-s.stopCh: + return + case <-resumeCh: } } - s.mu.Unlock() - s.ds.Push(subID, event) } func (s *regionEventSink) Run(ctx context.Context) error { for { select { case <-ctx.Done(): + s.stopWaiting() return nil case feedback := <-s.ds.Feedback(): switch feedback.FeedbackType { case dynstream.PauseArea: - s.mu.Lock() - s.paused.Store(true) - s.mu.Unlock() + s.pause() log.Info("subscription client pause push region event") case dynstream.ResumeArea: - s.mu.Lock() - s.paused.Store(false) - s.cond.Broadcast() - s.mu.Unlock() + s.resume() log.Info("subscription client resume push region event") case dynstream.ReleasePath, dynstream.ResumePath: // Ignore it, because it is no need to pause and resume a path in puller. @@ -148,9 +150,53 @@ func (s *regionEventSink) UpdateMetrics() { } func (s *regionEventSink) Close() { + s.stopWaiting() + s.ds.Close() +} + +func (s *regionEventSink) getResumeCh() <-chan struct{} { + s.mu.Lock() + defer s.mu.Unlock() + if !s.paused.Load() { + return nil + } + return s.resumeCh +} + +func (s *regionEventSink) pause() { + s.mu.Lock() + defer s.mu.Unlock() + if s.paused.Load() { + return + } + s.paused.Store(true) + s.resumeCh = make(chan struct{}) +} + +func (s *regionEventSink) resume() { s.mu.Lock() + defer s.mu.Unlock() + s.resumeLocked() +} + +func (s *regionEventSink) resumeLocked() { + if !s.paused.Load() { + return + } s.paused.Store(false) - s.cond.Broadcast() - s.mu.Unlock() - s.ds.Close() + if s.resumeCh != nil { + close(s.resumeCh) + s.resumeCh = nil + } +} + +func (s *regionEventSink) stopWaiting() { + s.stopOnce.Do(func() { + s.mu.Lock() + s.resumeLocked() + s.mu.Unlock() + if s.stopCh != nil { + close(s.stopCh) + } + }) } diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go index 5928a1525a..4969b8cf61 100644 --- a/logservice/logpuller/region_event_sink_test.go +++ b/logservice/logpuller/region_event_sink_test.go @@ -15,7 +15,6 @@ package logpuller import ( "context" - "sync" "sync/atomic" "testing" "time" @@ -77,10 +76,9 @@ func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { ds := newMockRegionEventSinkStream() sink := ®ionEventSink{ - ctx: ctx, - ds: ds, + ds: ds, + stopCh: make(chan struct{}), } - sink.cond = sync.NewCond(&sink.mu) runErrCh := make(chan error, 1) go func() { @@ -130,6 +128,7 @@ func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { } } +<<<<<<< HEAD func TestRegionEventSinkUpdateMetrics(t *testing.T) { t.Run("empty area metrics returns after queue gauges", func(t *testing.T) { ds := newMockRegionEventSinkStream() @@ -152,10 +151,9 @@ func TestRegionEventSinkUpdateMetrics(t *testing.T) { ).Set(456) sink := ®ionEventSink{ - ctx: context.Background(), - ds: ds, + ds: ds, + stopCh: make(chan struct{}), } - sink.cond = sync.NewCond(&sink.mu) sink.UpdateMetrics() require.Equal(t, float64(11), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) @@ -191,10 +189,9 @@ func TestRegionEventSinkUpdateMetrics(t *testing.T) { } sink := ®ionEventSink{ - ctx: context.Background(), - ds: ds, + ds: ds, + stopCh: make(chan struct{}), } - sink.cond = sync.NewCond(&sink.mu) sink.UpdateMetrics() require.Equal(t, float64(33), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) @@ -213,3 +210,52 @@ func TestRegionEventSinkUpdateMetrics(t *testing.T) { ))) }) } + +func TestRegionEventSinkRunCancelUnblocksPush(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ds := newMockRegionEventSinkStream() + sink := ®ionEventSink{ + ds: ds, + stopCh: make(chan struct{}), + } + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- sink.Run(ctx) + }() + + ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ + FeedbackType: dynstream.PauseArea, + } + require.Eventually(t, sink.paused.Load, time.Second, 10*time.Millisecond) + + pushDone := make(chan struct{}) + go func() { + sink.Push(SubscriptionID(1), regionEvent{resolvedTs: 100}) + close(pushDone) + }() + + select { + case <-pushDone: + t.Fatal("Push should block while the sink is paused") + case <-time.After(100 * time.Millisecond): + } + require.Equal(t, int32(0), ds.pushCount.Load()) + + cancel() + + select { + case <-pushDone: + case <-time.After(time.Second): + t.Fatal("Push should be unblocked by Run context cancellation") + } + require.Equal(t, int32(0), ds.pushCount.Load()) + + select { + case err := <-runErrCh: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Run should exit after context cancellation") + } +} diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 95d0921f4d..d77e5c8200 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -418,10 +418,9 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { sink := ®ionEventSink{ - ctx: context.Background(), - ds: &mockDynamicStream{}, + ds: &mockDynamicStream{}, + stopCh: make(chan struct{}), } - sink.cond = sync.NewCond(&sink.mu) client := &subscriptionClient{ eventSink: sink, regionTaskQueue: priorityqueue.New[PriorityTask](), From 36a675057066159777609fd890214059f637487c Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 20:45:28 +0800 Subject: [PATCH 2/4] fix --- logservice/logpuller/region_event_sink.go | 94 +++++++------------ .../logpuller/region_event_sink_test.go | 19 ++-- .../logpuller/region_request_worker_test.go | 2 +- logservice/logpuller/subscription_client.go | 2 +- .../logpuller/subscription_client_test.go | 7 +- 5 files changed, 50 insertions(+), 74 deletions(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index cf0a5f7666..1343f72076 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -28,17 +28,15 @@ import ( type regionEventSink struct { ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] // the following fields are used to coordinate pause/resume feedback with event producers. - mu sync.Mutex - resumeCh chan struct{} - stopCh chan struct{} - stopOnce sync.Once - paused atomic.Bool + mu sync.Mutex + cond *sync.Cond + paused atomic.Bool + stopped atomic.Bool } -func newRegionEventSink(_ context.Context, failureHandler *regionFailureHandler) *regionEventSink { - sink := ®ionEventSink{ - stopCh: make(chan struct{}), - } +func newRegionEventSink(failureHandler *regionFailureHandler) *regionEventSink { + sink := ®ionEventSink{} + sink.cond = sync.NewCond(&sink.mu) option := dynstream.NewOption() // Note: it is max batch size of the kv sent from tikv(not committed rows) @@ -75,36 +73,32 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { } func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { - for { - // fast path - select { - case <-s.stopCh: - return - default: - } - if !s.paused.Load() { - s.ds.Push(subID, event) - return - } + if s.stopped.Load() { + return + } + if !s.paused.Load() { + s.ds.Push(subID, event) + return + } - resumeCh := s.getResumeCh() - if resumeCh == nil { - continue - } + s.mu.Lock() + for s.paused.Load() && !s.stopped.Load() { + s.cond.Wait() + } + stopped := s.stopped.Load() + s.mu.Unlock() - select { - case <-s.stopCh: - return - case <-resumeCh: - } + if stopped { + return } + s.ds.Push(subID, event) } func (s *regionEventSink) Run(ctx context.Context) error { for { select { case <-ctx.Done(): - s.stopWaiting() + s.stop() return nil case feedback := <-s.ds.Feedback(): switch feedback.FeedbackType { @@ -150,53 +144,35 @@ func (s *regionEventSink) UpdateMetrics() { } func (s *regionEventSink) Close() { - s.stopWaiting() + s.stop() s.ds.Close() } -func (s *regionEventSink) getResumeCh() <-chan struct{} { - s.mu.Lock() - defer s.mu.Unlock() - if !s.paused.Load() { - return nil - } - return s.resumeCh -} - func (s *regionEventSink) pause() { s.mu.Lock() defer s.mu.Unlock() - if s.paused.Load() { + if s.stopped.Load() || s.paused.Load() { return } s.paused.Store(true) - s.resumeCh = make(chan struct{}) } func (s *regionEventSink) resume() { s.mu.Lock() defer s.mu.Unlock() - s.resumeLocked() -} - -func (s *regionEventSink) resumeLocked() { if !s.paused.Load() { return } s.paused.Store(false) - if s.resumeCh != nil { - close(s.resumeCh) - s.resumeCh = nil - } + s.cond.Broadcast() } -func (s *regionEventSink) stopWaiting() { - s.stopOnce.Do(func() { - s.mu.Lock() - s.resumeLocked() - s.mu.Unlock() - if s.stopCh != nil { - close(s.stopCh) - } - }) +func (s *regionEventSink) stop() { + if !s.stopped.CompareAndSwap(false, true) { + return + } + s.mu.Lock() + s.paused.Store(false) + s.cond.Broadcast() + s.mu.Unlock() } diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go index 4969b8cf61..98cf7c683b 100644 --- a/logservice/logpuller/region_event_sink_test.go +++ b/logservice/logpuller/region_event_sink_test.go @@ -15,6 +15,7 @@ package logpuller import ( "context" + "sync" "sync/atomic" "testing" "time" @@ -70,15 +71,20 @@ func (s *mockRegionEventSinkStream) GetMetrics() dynstream.Metrics[int, Subscrip return s.metrics } +func newTestRegionEventSink( + ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler], +) *regionEventSink { + sink := ®ionEventSink{ds: ds} + sink.cond = sync.NewCond(&sink.mu) + return sink +} + func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() ds := newMockRegionEventSinkStream() - sink := ®ionEventSink{ - ds: ds, - stopCh: make(chan struct{}), - } + sink := newTestRegionEventSink(ds) runErrCh := make(chan error, 1) go func() { @@ -215,10 +221,7 @@ func TestRegionEventSinkRunCancelUnblocksPush(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) ds := newMockRegionEventSinkStream() - sink := ®ionEventSink{ - ds: ds, - stopCh: make(chan struct{}), - } + sink := newTestRegionEventSink(ds) runErrCh := make(chan error, 1) go func() { diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 268045c492..97e631f71e 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -157,7 +157,7 @@ func newDispatchResolvedTsTestWorker(regionCount int) (*regionRequestWorker, *mo metrics: sharedClientMetrics{ batchResolvedSize: prometheus.ObserverFunc(func(float64) {}), }, - eventSink: ®ionEventSink{ds: ds}, + eventSink: newTestRegionEventSink(ds), }, } worker.requestedRegions.subscriptions = map[SubscriptionID]regionFeedStates{ diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 15d5002108..047f53c0fa 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -180,7 +180,7 @@ func NewSubscriptionClient( } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) subClient.failureHandler = newRegionFailureHandler(subClient) - subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) + subClient.eventSink = newRegionEventSink(subClient.failureHandler) subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) subClient.initMetrics() diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index d77e5c8200..95cf2f990d 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -343,7 +343,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { client := &subscriptionClient{ - eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + eventSink: newTestRegionEventSink(&mockDynamicStream{}), } client.spanRegistry = newSpanRegistry(nil, nil) client.failureHandler = newRegionFailureHandler(client) @@ -417,10 +417,7 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] } func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { - sink := ®ionEventSink{ - ds: &mockDynamicStream{}, - stopCh: make(chan struct{}), - } + sink := newTestRegionEventSink(&mockDynamicStream{}) client := &subscriptionClient{ eventSink: sink, regionTaskQueue: priorityqueue.New[PriorityTask](), From 52d0c39f0120c4d36680e59e746e164893bf6780 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 21:04:34 +0800 Subject: [PATCH 3/4] fix --- logservice/logpuller/region_event_sink.go | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 1343f72076..a89a0555bc 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -26,12 +26,16 @@ import ( // regionEventSink delivers region events to dynstream and owns push-side flow control. type regionEventSink struct { - ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] - // the following fields are used to coordinate pause/resume feedback with event producers. - mu sync.Mutex - cond *sync.Cond - paused atomic.Bool + // mu/cond coordinate the paused push path with pause/resume and shutdown signals. + mu sync.Mutex + cond *sync.Cond + // paused tracks whether region event pushing is temporarily held back by feedback. + paused atomic.Bool + // stopped marks the sink as shutting down so blocked pushers can exit instead of waiting for resume. stopped atomic.Bool + + // ds owns the dynstream used to deliver region events and receive flow-control feedback. + ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] } func newRegionEventSink(failureHandler *regionFailureHandler) *regionEventSink { @@ -76,11 +80,13 @@ func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { if s.stopped.Load() { return } + // fast path if !s.paused.Load() { s.ds.Push(subID, event) return } + // slow path: wait until paused is false s.mu.Lock() for s.paused.Load() && !s.stopped.Load() { s.cond.Wait() From 9ba978b2e8d50dfc3814e027871ab7d0f2a588e9 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 10 Jul 2026 18:11:22 +0800 Subject: [PATCH 4/4] fix --- logservice/logpuller/region_event_sink_test.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go index 98cf7c683b..0698d3fb4a 100644 --- a/logservice/logpuller/region_event_sink_test.go +++ b/logservice/logpuller/region_event_sink_test.go @@ -134,7 +134,6 @@ func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { } } -<<<<<<< HEAD func TestRegionEventSinkUpdateMetrics(t *testing.T) { t.Run("empty area metrics returns after queue gauges", func(t *testing.T) { ds := newMockRegionEventSinkStream() @@ -157,8 +156,7 @@ func TestRegionEventSinkUpdateMetrics(t *testing.T) { ).Set(456) sink := ®ionEventSink{ - ds: ds, - stopCh: make(chan struct{}), + ds: ds, } sink.UpdateMetrics() @@ -195,8 +193,7 @@ func TestRegionEventSinkUpdateMetrics(t *testing.T) { } sink := ®ionEventSink{ - ds: ds, - stopCh: make(chan struct{}), + ds: ds, } sink.UpdateMetrics()