From aee059dc178287e0b0b195624100b7b9ff86d0da Mon Sep 17 00:00:00 2001 From: lidezhu Date: Sun, 5 Jul 2026 13:00:53 +0800 Subject: [PATCH 1/5] alpha version --- logservice/eventstore/event_store.go | 12 +- logservice/eventstore/event_store_test.go | 2 + logservice/logpuller/memory_quota.go | 354 ++++++++++++++++++ logservice/logpuller/memory_quota_test.go | 117 ++++++ logservice/logpuller/region_event_handler.go | 31 ++ logservice/logpuller/region_event_sink.go | 25 +- .../logpuller/region_request_scheduler.go | 23 +- logservice/logpuller/region_request_worker.go | 3 + logservice/logpuller/requested_store.go | 1 + logservice/logpuller/span_registry.go | 3 + logservice/logpuller/subscription_client.go | 47 ++- .../logpuller/subscription_client_test.go | 11 +- logservice/schemastore/ddl_job_fetcher.go | 10 +- pkg/config/debug.go | 3 + server/server.go | 1 + 15 files changed, 609 insertions(+), 34 deletions(-) create mode 100644 logservice/logpuller/memory_quota.go create mode 100644 logservice/logpuller/memory_quota_test.go diff --git a/logservice/eventstore/event_store.go b/logservice/eventstore/event_store.go index f5b3e78986..61adf3068c 100644 --- a/logservice/eventstore/event_store.go +++ b/logservice/eventstore/event_store.go @@ -460,7 +460,7 @@ func (e *eventStore) Close(_ context.Context) error { } func (e *eventStore) RegisterDispatcher( - _ common.ChangeFeedID, + changefeedID common.ChangeFeedID, dispatcherID common.DispatcherID, dispatcherSpan *heartbeatpb.TableSpan, startTs uint64, @@ -678,7 +678,15 @@ func (e *eventStore) RegisterDispatcher( serverConfig := config.GetGlobalServerConfig() resolvedTsAdvanceInterval := int64(serverConfig.KVClient.AdvanceIntervalInMs) // Note: don't hold any lock when call Subscribe - e.subClient.Subscribe(subStat.subID, *dispatcherSpan, startTs, consumeKVEvents, advanceResolvedTs, resolvedTsAdvanceInterval, bdrMode) + e.subClient.Subscribe( + subStat.subID, + logpuller.NewChangefeedSubscriptionMeta(changefeedID), + *dispatcherSpan, + startTs, + consumeKVEvents, + advanceResolvedTs, + resolvedTsAdvanceInterval, + bdrMode) log.Info("new subscription created", zap.Stringer("dispatcherID", dispatcherID), zap.Uint64("startTs", startTs), diff --git a/logservice/eventstore/event_store_test.go b/logservice/eventstore/event_store_test.go index 4ac22e504d..b6e8645cc1 100644 --- a/logservice/eventstore/event_store_test.go +++ b/logservice/eventstore/event_store_test.go @@ -121,6 +121,7 @@ func (s *mockSubscriptionClient) AllocSubscriptionID() logpuller.SubscriptionID func (s *mockSubscriptionClient) Subscribe( subID logpuller.SubscriptionID, + meta logpuller.SubscriptionMeta, span heartbeatpb.TableSpan, startTs uint64, consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, @@ -128,6 +129,7 @@ func (s *mockSubscriptionClient) Subscribe( advanceInterval int64, bdrMode bool, ) { + _ = meta s.mu.Lock() defer s.mu.Unlock() s.subscriptions[subID] = &mockSubscriptionStat{ diff --git a/logservice/logpuller/memory_quota.go b/logservice/logpuller/memory_quota.go new file mode 100644 index 0000000000..cdd58e0511 --- /dev/null +++ b/logservice/logpuller/memory_quota.go @@ -0,0 +1,354 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "sync" + "sync/atomic" + + "github.com/pingcap/ticdc/pkg/common" +) + +const ( + defaultLogPullerMemoryQuota uint64 = 1024 * 1024 * 1024 + + defaultPauseWarmingRatio = 0.2 + defaultResumeWarmingRatio = 0.1 + defaultFreezeAllRatio = 0.9 + defaultResumeAllRatio = 0.7 +) + +type subscriptionKind uint8 + +const ( + subscriptionKindChangefeed subscriptionKind = iota + subscriptionKindSystem +) + +// SubscriptionMeta carries subscription ownership information for log puller +// local resource control. +type SubscriptionMeta struct { + ChangefeedID common.ChangeFeedID + Kind subscriptionKind +} + +func NewChangefeedSubscriptionMeta(changefeedID common.ChangeFeedID) SubscriptionMeta { + return SubscriptionMeta{ + ChangefeedID: changefeedID, + Kind: subscriptionKindChangefeed, + } +} + +func NewSystemSubscriptionMeta() SubscriptionMeta { + return SubscriptionMeta{Kind: subscriptionKindSystem} +} + +func (m SubscriptionMeta) isSystem() bool { + return m.Kind == subscriptionKindSystem +} + +type admissionLevel uint8 + +const ( + admissionNormal admissionLevel = iota + admissionPauseWarming + admissionFreezeAllNewScans +) + +type changefeedPhase uint8 + +const ( + changefeedPhaseWarming changefeedPhase = iota + changefeedPhaseNormal +) + +type memoryQuotaLease struct { + once sync.Once + release func() +} + +func (l *memoryQuotaLease) Release() { + if l == nil { + return + } + l.once.Do(l.release) +} + +type changefeedQuotaState struct { + id common.ChangeFeedID + + mu sync.Mutex + phase changefeedPhase + spans map[SubscriptionID]*subscribedSpan + leases map[*memoryQuotaLease]SubscriptionID + memUsed uint64 +} + +func newChangefeedQuotaState(id common.ChangeFeedID) *changefeedQuotaState { + return &changefeedQuotaState{ + id: id, + phase: changefeedPhaseWarming, + spans: make(map[SubscriptionID]*subscribedSpan), + leases: make(map[*memoryQuotaLease]SubscriptionID), + } +} + +type memoryQuotaController struct { + mu sync.Mutex + cond *sync.Cond + + capacity uint64 + used uint64 + level admissionLevel + + pauseWarmingRatio float64 + resumeWarmingRatio float64 + freezeAllRatio float64 + resumeAllRatio float64 + + changefeeds map[common.ChangeFeedID]*changefeedQuotaState + + onAvailable atomic.Value // func() +} + +func newMemoryQuotaController(capacity uint64) *memoryQuotaController { + if capacity == 0 { + capacity = defaultLogPullerMemoryQuota + } + c := &memoryQuotaController{ + capacity: capacity, + level: admissionNormal, + pauseWarmingRatio: defaultPauseWarmingRatio, + resumeWarmingRatio: defaultResumeWarmingRatio, + freezeAllRatio: defaultFreezeAllRatio, + resumeAllRatio: defaultResumeAllRatio, + changefeeds: make(map[common.ChangeFeedID]*changefeedQuotaState), + } + c.cond = sync.NewCond(&c.mu) + return c +} + +func (c *memoryQuotaController) SetOnAvailable(fn func()) { + c.onAvailable.Store(fn) +} + +func (c *memoryQuotaController) WakeAll() { + c.mu.Lock() + c.cond.Broadcast() + c.mu.Unlock() +} + +func (c *memoryQuotaController) onMemoryAvailable() { + if fn, ok := c.onAvailable.Load().(func()); ok && fn != nil { + fn() + } +} + +func (c *memoryQuotaController) Snapshot() (used uint64, capacity uint64, level admissionLevel) { + c.mu.Lock() + defer c.mu.Unlock() + return c.used, c.capacity, c.level +} + +func (c *memoryQuotaController) addSubscription(span *subscribedSpan) { + if span.meta.isSystem() { + return + } + c.mu.Lock() + defer c.mu.Unlock() + + state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) + state.mu.Lock() + state.spans[span.subID] = span + state.mu.Unlock() +} + +func (c *memoryQuotaController) removeSubscription(span *subscribedSpan) { + if span == nil || span.meta.isSystem() { + return + } + + var leases []*memoryQuotaLease + c.mu.Lock() + state := c.changefeeds[span.meta.ChangefeedID] + if state != nil { + state.mu.Lock() + delete(state.spans, span.subID) + for lease, subID := range state.leases { + if subID == span.subID { + leases = append(leases, lease) + } + } + if len(state.spans) == 0 { + delete(c.changefeeds, span.meta.ChangefeedID) + } + state.mu.Unlock() + } + c.mu.Unlock() + + for _, lease := range leases { + lease.Release() + } +} + +func (c *memoryQuotaController) markSubscriptionInitialized(span *subscribedSpan) { + if span == nil || span.meta.isSystem() { + return + } + + c.mu.Lock() + state := c.changefeeds[span.meta.ChangefeedID] + c.mu.Unlock() + if state == nil { + return + } + + state.mu.Lock() + defer state.mu.Unlock() + if state.phase == changefeedPhaseNormal { + return + } + for _, subSpan := range state.spans { + if !subSpan.initialized.Load() { + return + } + } + state.phase = changefeedPhaseNormal + c.onMemoryAvailable() +} + +func (c *memoryQuotaController) allowNewScan(span *subscribedSpan) (bool, string) { + if span == nil || span.meta.isSystem() { + return true, "" + } + + c.mu.Lock() + c.refreshLevelLocked() + level := c.level + state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) + c.mu.Unlock() + + if level == admissionFreezeAllNewScans { + return false, deferReasonMemoryFreeze + } + if level == admissionPauseWarming { + state.mu.Lock() + phase := state.phase + state.mu.Unlock() + if phase == changefeedPhaseWarming { + return false, deferReasonMemoryWarming + } + } + return true, "" +} + +func (c *memoryQuotaController) acquireEvent( + ctx context.Context, + span *subscribedSpan, + bytes uint64, +) *memoryQuotaLease { + if span == nil || span.meta.isSystem() || bytes == 0 { + return nil + } + + c.mu.Lock() + for c.used+bytes > c.capacity && c.used > 0 { + if ctx.Err() != nil { + c.mu.Unlock() + return nil + } + c.cond.Wait() + } + c.used += bytes + c.refreshLevelLocked() + state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) + + lease := &memoryQuotaLease{} + lease.release = func() { + c.mu.Lock() + if c.used >= bytes { + c.used -= bytes + } else { + c.used = 0 + } + c.refreshLevelLocked() + c.cond.Broadcast() + c.mu.Unlock() + + state.mu.Lock() + delete(state.leases, lease) + if state.memUsed >= bytes { + state.memUsed -= bytes + } else { + state.memUsed = 0 + } + state.mu.Unlock() + + c.onMemoryAvailable() + } + + state.mu.Lock() + state.leases[lease] = span.subID + state.memUsed += bytes + state.mu.Unlock() + c.mu.Unlock() + return lease +} + +func (c *memoryQuotaController) getOrCreateChangefeedStateLocked( + changefeedID common.ChangeFeedID, +) *changefeedQuotaState { + state := c.changefeeds[changefeedID] + if state != nil { + return state + } + state = newChangefeedQuotaState(changefeedID) + c.changefeeds[changefeedID] = state + return state +} + +func (c *memoryQuotaController) refreshLevelLocked() { + if c.capacity == 0 { + c.level = admissionNormal + return + } + usage := float64(c.used) / float64(c.capacity) + switch c.level { + case admissionFreezeAllNewScans: + if usage <= c.resumeAllRatio { + if usage >= c.pauseWarmingRatio { + c.level = admissionPauseWarming + } else { + c.level = admissionNormal + } + } + case admissionPauseWarming: + switch { + case usage >= c.freezeAllRatio: + c.level = admissionFreezeAllNewScans + case usage <= c.resumeWarmingRatio: + c.level = admissionNormal + } + default: + switch { + case usage >= c.freezeAllRatio: + c.level = admissionFreezeAllNewScans + case usage >= c.pauseWarmingRatio: + c.level = admissionPauseWarming + default: + c.level = admissionNormal + } + } +} diff --git a/logservice/logpuller/memory_quota_test.go b/logservice/logpuller/memory_quota_test.go new file mode 100644 index 0000000000..29cbdeb53c --- /dev/null +++ b/logservice/logpuller/memory_quota_test.go @@ -0,0 +1,117 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "testing" + "time" + + "github.com/pingcap/ticdc/pkg/common" + "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/oracle" +) + +func newTestQuotaSpan(subID SubscriptionID, changefeedID common.ChangeFeedID) *subscribedSpan { + span := &subscribedSpan{ + subID: subID, + meta: NewChangefeedSubscriptionMeta(changefeedID), + } + span.resolvedTs.Store(oracle.GoTimeToTS(time.Now())) + return span +} + +func TestMemoryQuotaAdmissionLevels(t *testing.T) { + controller := newMemoryQuotaController(100) + + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) + normalSpan := newTestQuotaSpan(2, common.NewChangeFeedIDWithName("normal", common.DefaultKeyspaceName)) + normalSpan.initialized.Store(true) + + controller.addSubscription(warmingSpan) + controller.addSubscription(normalSpan) + controller.markSubscriptionInitialized(normalSpan) + + ok, reason := controller.allowNewScan(warmingSpan) + require.True(t, ok) + require.Empty(t, reason) + + softLease := controller.acquireEvent(context.Background(), normalSpan, 20) + t.Cleanup(softLease.Release) + ok, reason = controller.allowNewScan(warmingSpan) + require.False(t, ok) + require.Equal(t, deferReasonMemoryWarming, reason) + ok, reason = controller.allowNewScan(normalSpan) + require.True(t, ok) + require.Empty(t, reason) + + hardLease := controller.acquireEvent(context.Background(), normalSpan, 70) + t.Cleanup(hardLease.Release) + ok, reason = controller.allowNewScan(normalSpan) + require.False(t, ok) + require.Equal(t, deferReasonMemoryFreeze, reason) + + hardLease.Release() + _, _, level := controller.Snapshot() + require.Equal(t, admissionPauseWarming, level) + + softLease.Release() + _, _, level = controller.Snapshot() + require.Equal(t, admissionNormal, level) +} + +func TestWarmingHighPriorityTaskBlockedByMemoryGate(t *testing.T) { + controller := newMemoryQuotaController(100) + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) + controller.addSubscription(warmingSpan) + lease := controller.acquireEvent(context.Background(), warmingSpan, 20) + t.Cleanup(lease.Release) + + scheduler := ®ionRequestScheduler{memoryQuota: controller} + region := regionInfo{subscribedSpan: warmingSpan} + task := newRegionPriorityTask(TaskHighPrior, region, oracle.GoTimeToTS(time.Now()), 1) + + ok, reason, err := scheduler.tryAdmitTask(context.Background(), &requestedStore{}, task, region) + require.NoError(t, err) + require.False(t, ok) + require.Equal(t, deferReasonMemoryWarming, reason) +} + +func TestRemoveSubscriptionReleasesOnlyItsOutstandingMemory(t *testing.T) { + controller := newMemoryQuotaController(100) + changefeedID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) + span1 := newTestQuotaSpan(1, changefeedID) + span2 := newTestQuotaSpan(2, changefeedID) + controller.addSubscription(span1) + controller.addSubscription(span2) + + lease1 := controller.acquireEvent(context.Background(), span1, 30) + lease2 := controller.acquireEvent(context.Background(), span2, 40) + t.Cleanup(lease2.Release) + + used, _, _ := controller.Snapshot() + require.Equal(t, uint64(70), used) + + controller.removeSubscription(span1) + used, _, _ = controller.Snapshot() + require.Equal(t, uint64(40), used) + + lease1.Release() + used, _, _ = controller.Snapshot() + require.Equal(t, uint64(40), used) + + lease2.Release() + used, _, _ = controller.Snapshot() + require.Equal(t, uint64(0), used) +} diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 91c4466bd0..073b49046f 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -56,6 +56,20 @@ type regionEvent struct { entries *cdcpb.Event_Entries_ resolvedTs uint64 + + memoryQuota *memoryQuotaLease +} + +func (event *regionEvent) needMemoryQuota() bool { + return event != nil && event.entries != nil +} + +func (event *regionEvent) releaseMemoryQuota() { + if event == nil || event.memoryQuota == nil { + return + } + event.memoryQuota.Release() + event.memoryQuota = nil } func (event *regionEvent) getSize() int { @@ -122,7 +136,12 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) } newResolvedTs := uint64(0) + quotaEvents := make([]*regionEvent, 0, len(events)) for _, event := range events { + event := event + if event.memoryQuota != nil { + quotaEvents = append(quotaEvents, &event) + } if len(event.states) == 1 && event.states[0].isStale() { hasError = true h.handleRegionError(event.states[0]) @@ -148,9 +167,15 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) span.advanceResolvedTs(newResolvedTs) } } + releaseMemoryQuota := func() { + for _, event := range quotaEvents { + event.releaseMemoryQuota() + } + } if len(span.kvEventsCache) > 0 { metricsEventCount.Add(float64(len(span.kvEventsCache))) await := span.consumeKVEvents(span.kvEventsCache, func() { + defer releaseMemoryQuota() start := time.Now() span.clearKVEventsCache() metricConsumeKVEventsCallbackDurationClearCache.Observe(time.Since(start).Seconds()) @@ -167,10 +192,12 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) if !await { span.clearKVEventsCache() tryAdvanceResolvedTs() + releaseMemoryQuota() } return await } else { tryAdvanceResolvedTs() + releaseMemoryQuota() } return false } @@ -227,6 +254,7 @@ func (h *regionEventHandler) GetType(event regionEvent) dynstream.EventType { } func (h *regionEventHandler) OnDrop(event regionEvent) interface{} { + event.releaseMemoryQuota() // TODO: Distinguish between drop events caused by "path not found" errors and memory control. state := event.mustFirstState() fields := []zap.Field{ @@ -398,6 +426,9 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u zap.Uint64("subscriptionID", uint64(span.subID)), zap.Uint64("regionID", regionID), zap.Uint64("resolvedTs", ts)) + if state.worker != nil && state.worker.memoryQuota != nil { + state.worker.memoryQuota.markSubscriptionInitialized(span) + } } lastResolvedTs := span.resolvedTs.Load() nextResolvedPhyTs := oracle.ExtractPhysical(ts) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 0a90527d4b..0414599093 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -27,13 +27,19 @@ type regionEventSink struct { ctx context.Context ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] + memoryQuota *memoryQuotaController + mu sync.Mutex cond *sync.Cond paused atomic.Bool } -func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandler) *regionEventSink { - sink := ®ionEventSink{ctx: ctx} +func newRegionEventSink( + ctx context.Context, + failureHandler *regionFailureHandler, + memoryQuota *memoryQuotaController, +) *regionEventSink { + sink := ®ionEventSink{ctx: ctx, memoryQuota: memoryQuota} sink.cond = sync.NewCond(&sink.mu) option := dynstream.NewOption() @@ -42,7 +48,7 @@ func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandle // TODO: Set `UseBuffer` to true until we refactor the `regionEventHandler.Handle` method so that it doesn't call any method of the dynamic stream. Currently, if `UseBuffer` is set to false, there will be a deadlock: // ds.handleLoop fetch events from `ch` -> regionEventHandler.Handle -> ds.RemovePath -> send event to `ch` option.UseBuffer = true - option.EnableMemoryControl = true + option.EnableMemoryControl = false ds := dynstream.NewParallelDynamicStream( "log-puller", ®ionEventHandler{eventSink: sink, failureHandler: failureHandler}, @@ -54,8 +60,7 @@ func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandle } func (s *regionEventSink) AddPath(rt *subscribedSpan) { - areaSetting := dynstream.NewAreaSettingsWithMaxPendingSize(1*1024*1024*1024, dynstream.MemoryControlForPuller, "logPuller") // 1GB - if err := s.ds.AddPath(rt.subID, rt, areaSetting); err != nil { + if err := s.ds.AddPath(rt.subID, rt); err != nil { log.Warn("subscription client add path failed", zap.Uint64("subscriptionID", uint64(rt.subID)), zap.Error(err)) @@ -71,6 +76,13 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { } func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { + if event.needMemoryQuota() && s.memoryQuota != nil { + span := event.mustFirstState().region.subscribedSpan + event.memoryQuota = s.memoryQuota.acquireEvent(s.ctx, span, uint64(event.getSize())) + if event.memoryQuota == nil && s.ctx.Err() != nil { + return + } + } // fast path if !s.paused.Load() { s.ds.Push(subID, event) @@ -125,6 +137,9 @@ func (s *regionEventSink) Metrics() dynstream.Metrics[int, SubscriptionID] { } func (s *regionEventSink) Close() { + if s.memoryQuota != nil { + s.memoryQuota.WakeAll() + } s.mu.Lock() s.paused.Store(false) s.cond.Broadcast() diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index 21c2020fe7..13ea4683b3 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -31,9 +31,11 @@ import ( ) const ( - deferReasonStorePending = "store_pending" - deferReasonStoreQuota = "store_quota" - deferReasonWorkerCache = "worker_cache" + deferReasonStorePending = "store_pending" + deferReasonStoreQuota = "store_quota" + deferReasonWorkerCache = "worker_cache" + deferReasonMemoryWarming = "memory_warming" + deferReasonMemoryFreeze = "memory_freeze" ) // regionRequestScheduler owns region request admission from the global @@ -44,6 +46,7 @@ type regionRequestScheduler struct { eventSink *regionEventSink failureHandler *regionFailureHandler + memoryQuota *memoryQuotaController // queue stores newly submitted tasks before they are routed to a TiKV store. queue *priorityqueue.PriorityQueue[*regionPriorityTask] @@ -68,6 +71,7 @@ func newRegionRequestScheduler(client *subscriptionClient) *regionRequestSchedul upstream: client.upstream, eventSink: client.eventSink, failureHandler: client.failureHandler, + memoryQuota: client.memoryQuota, queue: priorityqueue.New[*regionPriorityTask](), schedulerNotify: make(chan struct{}, 1), storeAvailable: chann.NewUnlimitedChannelDefault[*requestedStore](), @@ -183,6 +187,14 @@ func (s *regionRequestScheduler) Close() { }) } +func (s *regionRequestScheduler) NotifyAvailable() { + s.stores.Range(func(_, value any) bool { + value.(*requestedStore).NotifyAvailable() + return true + }) + s.notifyScheduler() +} + func (s *regionRequestScheduler) notifyScheduler() { select { case s.schedulerNotify <- struct{}{}: @@ -280,6 +292,11 @@ func (s *regionRequestScheduler) tryAdmitTask( region regionInfo, ) (bool, string, error) { force := task.Priority() <= forcedPriorityBase + if s.memoryQuota != nil { + if ok, reason := s.memoryQuota.allowNewScan(region.subscribedSpan); !ok { + return false, reason, nil + } + } acquiredQuota, ok := store.quota.TryAcquire() if !ok { return false, deferReasonStoreQuota, nil diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 747826a875..9f051bd907 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -86,6 +86,7 @@ type regionRequestWorker struct { upstream *upstreamHandle eventSink *regionEventSink failureHandler *regionFailureHandler + memoryQuota *memoryQuotaController // request cache with flow control requestCache *requestCache @@ -100,6 +101,7 @@ func newRegionRequestWorker( upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, + memoryQuota *memoryQuotaController, ) *regionRequestWorker { workerID := workerIDGen.Add(1) worker := ®ionRequestWorker{ @@ -108,6 +110,7 @@ func newRegionRequestWorker( upstream: upstream, eventSink: eventSink, failureHandler: failureHandler, + memoryQuota: memoryQuota, requestCache: requestCache, controlQueue: newControlQueue(), tracker: newRegionTracker(workerID), diff --git a/logservice/logpuller/requested_store.go b/logservice/logpuller/requested_store.go index 11a47a4448..7cf713f16a 100644 --- a/logservice/logpuller/requested_store.go +++ b/logservice/logpuller/requested_store.go @@ -127,6 +127,7 @@ func newRequestedStore(scheduler *regionRequestScheduler, storeAddr string) *req scheduler.upstream, scheduler.eventSink, scheduler.failureHandler, + scheduler.memoryQuota, ) rs.requestWorkers = append(rs.requestWorkers, requestWorker) } diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go index c4cc080892..b5cf01d018 100644 --- a/logservice/logpuller/span_registry.go +++ b/logservice/logpuller/span_registry.go @@ -35,6 +35,7 @@ const kvEventsCacheMaxSize = 32 // subscribedSpan is the local state for one subscribed table span. type subscribedSpan struct { subID SubscriptionID + meta SubscriptionMeta startTs uint64 // Whether to filter out the value written by TiCDC itself. // It should be `true` in BDR mode. @@ -86,6 +87,7 @@ func newSubscribedSpan( resolveLockRateLimiter *resolveLockRateLimiter, resolveLockTaskCh chan resolveLockTask, subID SubscriptionID, + meta SubscriptionMeta, span heartbeatpb.TableSpan, startTs uint64, consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, @@ -97,6 +99,7 @@ func newSubscribedSpan( rt := &subscribedSpan{ subID: subID, + meta: meta, span: span, startTs: startTs, filterLoop: filterLoop, diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index a5a0ee2b8f..2ad8622591 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -84,6 +84,8 @@ type SubscriptionClientConfig struct { RegionRequestWorkerPerStore uint // PendingRegionRequestQueueSize is the total pending region request quota for one TiKV store. PendingRegionRequestQueueSize int + // MemoryQuota is the log puller local memory quota in bytes. + MemoryQuota uint64 } type upstreamHandle struct { @@ -107,6 +109,7 @@ type SubscriptionClient interface { // subscribe a table span Subscribe( subID SubscriptionID, + meta SubscriptionMeta, span heartbeatpb.TableSpan, startTs uint64, consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, @@ -130,6 +133,7 @@ type subscriptionClient struct { failureHandler *regionFailureHandler spanRegistry *spanRegistry + memoryQuota *memoryQuotaController // rangeTaskCh is used to receive range tasks. // The tasks will be handled in `handleRangeTask` goroutine. @@ -166,6 +170,7 @@ func NewSubscriptionClient( } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) subClient.spanRegistry = newSpanRegistry(subClient.upstream) + subClient.memoryQuota = newMemoryQuotaController(config.MemoryQuota) subClient.failureHandler = newRegionFailureHandler( subClient.upstream, @@ -173,8 +178,9 @@ func NewSubscriptionClient( subClient.scheduleRegionRequest, subClient.scheduleRangeRequest, ) - subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) + subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler, subClient.memoryQuota) subClient.regionScheduler = newRegionRequestScheduler(subClient) + subClient.memoryQuota.SetOnAvailable(subClient.regionScheduler.NotifyAvailable) return subClient } @@ -198,24 +204,19 @@ func (s *subscriptionClient) runMetricsUpdater(ctx context.Context) error { dsMetrics := s.eventSink.Metrics() metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 1 { - log.Panic("subscription client should have only one area") - } - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 0 { - areaMetric := dsMetrics.MemoryControl.AreaMemoryMetrics[0] - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "max", - "default", - "default", - ).Set(float64(areaMetric.MaxMemory())) - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "used", - "default", - "default", - ).Set(float64(areaMetric.MemoryUsage())) - } + used, capacity, _ := s.memoryQuota.Snapshot() + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "max", + "default", + "default", + ).Set(float64(capacity)) + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "used", + "default", + "default", + ).Set(float64(used)) s.regionScheduler.UpdateMetrics() s.spanRegistry.UpdateMetrics() @@ -230,6 +231,7 @@ func (s *subscriptionClient) runMetricsUpdater(ctx context.Context) error { // The rangeTask will be handled in `handleRangeTasks` goroutine. func (s *subscriptionClient) Subscribe( subID SubscriptionID, + meta SubscriptionMeta, span heartbeatpb.TableSpan, startTs uint64, consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, @@ -247,6 +249,7 @@ func (s *subscriptionClient) Subscribe( s.resolveLockRateLimiter, s.resolveLockTaskCh, subID, + meta, span, startTs, consumeKVEvents, @@ -255,6 +258,9 @@ func (s *subscriptionClient) Subscribe( bdrMode, ) s.spanRegistry.Add(rt) + if s.memoryQuota != nil { + s.memoryQuota.addSubscription(rt) + } s.eventSink.AddPath(rt) @@ -335,6 +341,9 @@ func (s *subscriptionClient) onTableDrained(rt *subscribedSpan) { zap.Uint64("subscriptionID", uint64(rt.subID)), zap.Error(err)) } + if s.memoryQuota != nil { + s.memoryQuota.removeSubscription(rt) + } s.spanRegistry.Remove(rt.subID) } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 34dc76170b..9b4757f26e 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -74,6 +74,7 @@ func TestGenerateResolveLockTask(t *testing.T) { client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), + NewSystemSubscriptionMeta(), rawSpan, 100, consumeKVEvents, @@ -149,12 +150,12 @@ func TestResolveLockTaskDeduplicatedAcrossSubscribedSpans(t *testing.T) { consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span1 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), heartbeatpb.TableSpan{ + span1 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), NewSystemSubscriptionMeta(), heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte{'a'}, EndKey: []byte{'z'}, }, 100, consumeKVEvents, advanceResolvedTs, 0, false) - span2 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(2), heartbeatpb.TableSpan{ + span2 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(2), NewSystemSubscriptionMeta(), heartbeatpb.TableSpan{ TableID: 2, StartKey: []byte{'a'}, EndKey: []byte{'z'}, @@ -262,6 +263,7 @@ func TestResolveLockTaskDroppedWhenChannelFull(t *testing.T) { client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), + NewSystemSubscriptionMeta(), rawSpan, 100, consumeKVEvents, @@ -322,6 +324,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), + NewSystemSubscriptionMeta(), rawSpan, 100, consumeKVEvents, @@ -644,7 +647,7 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) { case tsCh <- ts: } } - client.Subscribe(subID, span, 1, consumeKVEvents, advanceResolvedTs, 0, false) + client.Subscribe(subID, NewSystemSubscriptionMeta(), span, 1, consumeKVEvents, advanceResolvedTs, 0, false) eventsCh1 <- mockInitializedEvent(11, uint64(subID)) targetTs := oracle.GoTimeToTS(pdClock.CurrentTime()) @@ -827,7 +830,7 @@ func TestGetResolvedTargetTs(t *testing.T) { consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), heartbeatpb.TableSpan{ + span := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), NewSystemSubscriptionMeta(), heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte{'a'}, EndKey: []byte{'z'}, diff --git a/logservice/schemastore/ddl_job_fetcher.go b/logservice/schemastore/ddl_job_fetcher.go index 16e079e290..570ada9b36 100644 --- a/logservice/schemastore/ddl_job_fetcher.go +++ b/logservice/schemastore/ddl_job_fetcher.go @@ -96,7 +96,15 @@ func (p *ddlJobFetcher) run(startTs uint64) error { advanceSubSpanResolvedTs := func(ts uint64) { p.tryAdvanceResolvedTs(subID, ts) } - p.subClient.Subscribe(subID, span, startTs, p.input, advanceSubSpanResolvedTs, 0, ddlPullerFilterLoop) + p.subClient.Subscribe( + subID, + logpuller.NewSystemSubscriptionMeta(), + span, + startTs, + p.input, + advanceSubSpanResolvedTs, + 0, + ddlPullerFilterLoop) } return nil } diff --git a/pkg/config/debug.go b/pkg/config/debug.go index 1aa6f9b924..e8d4eb482b 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -67,6 +67,8 @@ type PullerConfig struct { // For example, if PendingRegionRequestQueueSize is 32 and there are 8 workers connecting to the same store, // each worker's queue size will be 32 / 8 = 4. PendingRegionRequestQueueSize int `toml:"pending-region-request-queue-size" json:"pending_region_request_queue_size"` + // MemoryQuota is the log puller local memory quota in bytes. + MemoryQuota uint64 `toml:"memory-quota" json:"memory_quota"` } // NewDefaultPullerConfig return the default puller configuration @@ -76,6 +78,7 @@ func NewDefaultPullerConfig() *PullerConfig { ResolvedTsStuckInterval: TomlDuration(5 * time.Minute), LogRegionDetails: false, PendingRegionRequestQueueSize: 32, // This value is chosen to reduce the impact of new changefeeds on existing ones. + MemoryQuota: 1024 * 1024 * 1024, } } diff --git a/server/server.go b/server/server.go index 2e83359930..ab05f99bf7 100644 --- a/server/server.go +++ b/server/server.go @@ -201,6 +201,7 @@ func (c *server) initialize(ctx context.Context) error { &logpuller.SubscriptionClientConfig{ RegionRequestWorkerPerStore: 8, PendingRegionRequestQueueSize: conf.Debug.Puller.PendingRegionRequestQueueSize, + MemoryQuota: conf.Debug.Puller.MemoryQuota, }, c.pdClient, txnutil.NewLockerResolver(), c.security, From 282dca7802ffdbc7d5bc7041d6c1b0548e74bc41 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Mon, 6 Jul 2026 17:25:37 +0800 Subject: [PATCH 2/5] add memory quota metrics --- logservice/logpuller/subscription_client.go | 2 + metrics/grafana/ticdc_new_arch.json | 90 +++++++++++++++++++ .../ticdc_new_arch_next_gen.json | 90 +++++++++++++++++++ pkg/metrics/log_puller.go | 8 ++ 4 files changed, 190 insertions(+) diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 2ad8622591..ca52428cd6 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -205,6 +205,8 @@ func (s *subscriptionClient) runMetricsUpdater(ctx context.Context) error { metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) used, capacity, _ := s.memoryQuota.Snapshot() + metrics.LogPullerMemoryQuota.WithLabelValues("max").Set(float64(capacity)) + metrics.LogPullerMemoryQuota.WithLabelValues("used").Set(float64(used)) metrics.DynamicStreamMemoryUsage.WithLabelValues( "log-puller", "max", diff --git a/metrics/grafana/ticdc_new_arch.json b/metrics/grafana/ticdc_new_arch.json index 5b6a19db31..4b4cc1bdf6 100644 --- a/metrics/grafana/ticdc_new_arch.json +++ b/metrics/grafana/ticdc_new_arch.json @@ -8815,6 +8815,96 @@ "align": false, "alignLevel": null } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Log puller memory quota tracked by the local memory quota controller.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 0, + "fillGradient": 0, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 57 + }, + "hiddenSeries": false, + "id": 26001, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": true, + "min": false, + "show": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 2, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "exemplar": true, + "expr": "sum(ticdc_log_puller_memory_quota{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}) by (instance, type)", + "interval": "", + "legendFormat": "{{instance}}-{{type}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Controller", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "bytes", + "logBase": 1, + "min": "0", + "show": true + }, + { + "format": "short", + "logBase": 1, + "show": false + } + ], + "yaxis": { + "align": false + } } ], "title": "Log Puller", diff --git a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json index 77a3a6a48a..f774a96846 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json +++ b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json @@ -8815,6 +8815,96 @@ "align": false, "alignLevel": null } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Log puller memory quota tracked by the local memory quota controller.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 0, + "fillGradient": 0, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 57 + }, + "hiddenSeries": false, + "id": 26001, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": true, + "min": false, + "show": true, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "nullPointMode": "null", + "options": { + "alertThreshold": true + }, + "percentage": false, + "pluginVersion": "7.5.17", + "pointradius": 2, + "points": false, + "renderer": "flot", + "seriesOverrides": [], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "exemplar": true, + "expr": "sum(ticdc_log_puller_memory_quota{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}) by (instance, type)", + "interval": "", + "legendFormat": "{{instance}}-{{type}}", + "refId": "A" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Controller", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "bytes", + "logBase": 1, + "min": "0", + "show": true + }, + { + "format": "short", + "logBase": 1, + "show": false + } + ], + "yaxis": { + "align": false + } } ], "title": "Log Puller", diff --git a/pkg/metrics/log_puller.go b/pkg/metrics/log_puller.go index 7b791249a7..83362bd8c7 100644 --- a/pkg/metrics/log_puller.go +++ b/pkg/metrics/log_puller.go @@ -64,6 +64,13 @@ var ( Name: "resolved_ts_lag", Help: "The lag of resolved ts", }) + LogPullerMemoryQuota = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "ticdc", + Subsystem: "log_puller", + Name: "memory_quota", + Help: "The log puller local memory quota usage.", + }, []string{"type"}) SubscriptionClientResolvedTsLagGauge = prometheus.NewGauge( prometheus.GaugeOpts{ @@ -207,6 +214,7 @@ func initLogPullerMetrics(registry *prometheus.Registry) { registry.MustRegister(LogPullerPrewriteCacheRowNum) registry.MustRegister(LogPullerMatcherCount) registry.MustRegister(LogPullerResolvedTsLag) + registry.MustRegister(LogPullerMemoryQuota) registry.MustRegister(SubscriptionClientRequestedRegionCount) registry.MustRegister(SubscriptionClientWorkerPendingRegionCount) registry.MustRegister(SubscriptionClientStoreDeferredRegionCount) From e62a0401770675c704dc926ec04b0feecdba3072 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Tue, 7 Jul 2026 19:53:00 +0800 Subject: [PATCH 3/5] refactor --- logservice/logpuller/memory_quota.go | 312 ++++++++++-------- logservice/logpuller/memory_quota_test.go | 185 +++++++++-- logservice/logpuller/region_event_handler.go | 11 +- .../logpuller/region_event_handler_test.go | 8 +- logservice/logpuller/region_event_sink.go | 10 +- logservice/logpuller/region_request_cache.go | 30 +- .../logpuller/region_request_cache_test.go | 58 +++- .../logpuller/region_request_scheduler.go | 37 ++- .../logpuller/region_request_worker_test.go | 14 +- logservice/logpuller/requested_store.go | 8 +- logservice/logpuller/subscription_client.go | 17 +- .../logpuller/subscription_client_test.go | 7 +- pkg/config/debug.go | 3 + server/server.go | 1 + 14 files changed, 481 insertions(+), 220 deletions(-) diff --git a/logservice/logpuller/memory_quota.go b/logservice/logpuller/memory_quota.go index cdd58e0511..6b0ab04d3d 100644 --- a/logservice/logpuller/memory_quota.go +++ b/logservice/logpuller/memory_quota.go @@ -14,11 +14,13 @@ package logpuller import ( - "context" + "math" "sync" "sync/atomic" + "time" "github.com/pingcap/ticdc/pkg/common" + "github.com/tikv/client-go/v2/oracle" ) const ( @@ -28,6 +30,12 @@ const ( defaultResumeWarmingRatio = 0.1 defaultFreezeAllRatio = 0.9 defaultResumeAllRatio = 0.7 + + defaultScanBaseSize uint64 = 4 * 1024 * 1024 + defaultWarmingScanLagThreshold = 30 * time.Minute + defaultScanLagUnit = 10 * time.Minute + defaultScanLagWeight = 0.15 + defaultMaxScanLagFactor = 16 ) type subscriptionKind uint8 @@ -67,66 +75,61 @@ const ( admissionFreezeAllNewScans ) -type changefeedPhase uint8 - -const ( - changefeedPhaseWarming changefeedPhase = iota - changefeedPhaseNormal -) - type memoryQuotaLease struct { once sync.Once release func() } func (l *memoryQuotaLease) Release() { - if l == nil { - return - } l.once.Do(l.release) } -type changefeedQuotaState struct { - id common.ChangeFeedID - - mu sync.Mutex - phase changefeedPhase - spans map[SubscriptionID]*subscribedSpan - leases map[*memoryQuotaLease]SubscriptionID - memUsed uint64 +type subscriptionQuotaState struct { + eventLeases map[*memoryQuotaLease]uint64 + scanLeases map[*memoryQuotaLease]uint64 + scanUsed uint64 } -func newChangefeedQuotaState(id common.ChangeFeedID) *changefeedQuotaState { - return &changefeedQuotaState{ - id: id, - phase: changefeedPhaseWarming, - spans: make(map[SubscriptionID]*subscribedSpan), - leases: make(map[*memoryQuotaLease]SubscriptionID), +func newSubscriptionQuotaState() *subscriptionQuotaState { + return &subscriptionQuotaState{ + eventLeases: make(map[*memoryQuotaLease]uint64), + scanLeases: make(map[*memoryQuotaLease]uint64), } } type memoryQuotaController struct { - mu sync.Mutex - cond *sync.Cond + mu sync.Mutex capacity uint64 - used uint64 - level admissionLevel + // used tracks log puller bytes currently waiting for downstream callback. + // It is observed for admission decisions, but the receive path never waits + // on it after an event has arrived from TiKV. + used uint64 + // scanUsed tracks estimated bytes of admitted warming region scans that + // have not finished initialization yet. + scanUsed uint64 + warmingScanUsed uint64 + level admissionLevel pauseWarmingRatio float64 resumeWarmingRatio float64 freezeAllRatio float64 resumeAllRatio float64 - changefeeds map[common.ChangeFeedID]*changefeedQuotaState + scanEstimate uint64 + + subscriptions map[SubscriptionID]*subscriptionQuotaState onAvailable atomic.Value // func() } -func newMemoryQuotaController(capacity uint64) *memoryQuotaController { +func newMemoryQuotaController(capacity uint64, scanBaseSize uint64) *memoryQuotaController { if capacity == 0 { capacity = defaultLogPullerMemoryQuota } + if scanBaseSize == 0 { + scanBaseSize = defaultScanBaseSize + } c := &memoryQuotaController{ capacity: capacity, level: admissionNormal, @@ -134,9 +137,9 @@ func newMemoryQuotaController(capacity uint64) *memoryQuotaController { resumeWarmingRatio: defaultResumeWarmingRatio, freezeAllRatio: defaultFreezeAllRatio, resumeAllRatio: defaultResumeAllRatio, - changefeeds: make(map[common.ChangeFeedID]*changefeedQuotaState), + scanEstimate: scanBaseSize, + subscriptions: make(map[SubscriptionID]*subscriptionQuotaState), } - c.cond = sync.NewCond(&c.mu) return c } @@ -144,12 +147,6 @@ func (c *memoryQuotaController) SetOnAvailable(fn func()) { c.onAvailable.Store(fn) } -func (c *memoryQuotaController) WakeAll() { - c.mu.Lock() - c.cond.Broadcast() - c.mu.Unlock() -} - func (c *memoryQuotaController) onMemoryAvailable() { if fn, ok := c.onAvailable.Load().(func()); ok && fn != nil { fn() @@ -162,6 +159,17 @@ func (c *memoryQuotaController) Snapshot() (used uint64, capacity uint64, level return c.used, c.capacity, c.level } +func (c *memoryQuotaController) ScanSnapshot() ( + scanUsed uint64, + warmingScanUsed uint64, + warmingScanBudget uint64, + scanEstimate uint64, +) { + c.mu.Lock() + defer c.mu.Unlock() + return c.scanUsed, c.warmingScanUsed, c.warmingScanBudgetLocked(), c.scanEstimate +} + func (c *memoryQuotaController) addSubscription(span *subscribedSpan) { if span.meta.isSystem() { return @@ -169,33 +177,24 @@ func (c *memoryQuotaController) addSubscription(span *subscribedSpan) { c.mu.Lock() defer c.mu.Unlock() - state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) - state.mu.Lock() - state.spans[span.subID] = span - state.mu.Unlock() + c.subscriptions[span.subID] = newSubscriptionQuotaState() } func (c *memoryQuotaController) removeSubscription(span *subscribedSpan) { - if span == nil || span.meta.isSystem() { + if span.meta.isSystem() { return } var leases []*memoryQuotaLease c.mu.Lock() - state := c.changefeeds[span.meta.ChangefeedID] - if state != nil { - state.mu.Lock() - delete(state.spans, span.subID) - for lease, subID := range state.leases { - if subID == span.subID { - leases = append(leases, lease) - } - } - if len(state.spans) == 0 { - delete(c.changefeeds, span.meta.ChangefeedID) - } - state.mu.Unlock() + state := c.subscriptions[span.subID] + for lease := range state.eventLeases { + leases = append(leases, lease) + } + for lease := range state.scanLeases { + leases = append(leases, lease) } + delete(c.subscriptions, span.subID) c.mu.Unlock() for _, lease := range leases { @@ -204,76 +203,75 @@ func (c *memoryQuotaController) removeSubscription(span *subscribedSpan) { } func (c *memoryQuotaController) markSubscriptionInitialized(span *subscribedSpan) { - if span == nil || span.meta.isSystem() { - return - } - - c.mu.Lock() - state := c.changefeeds[span.meta.ChangefeedID] - c.mu.Unlock() - if state == nil { - return - } - - state.mu.Lock() - defer state.mu.Unlock() - if state.phase == changefeedPhaseNormal { - return - } - for _, subSpan := range state.spans { - if !subSpan.initialized.Load() { - return - } - } - state.phase = changefeedPhaseNormal c.onMemoryAvailable() } -func (c *memoryQuotaController) allowNewScan(span *subscribedSpan) (bool, string) { - if span == nil || span.meta.isSystem() { - return true, "" +func (c *memoryQuotaController) acquireScan(region regionInfo, currentTs uint64) (*memoryQuotaLease, bool, string) { + span := region.subscribedSpan + if span.meta.isSystem() { + return nil, true, "" } c.mu.Lock() c.refreshLevelLocked() - level := c.level - state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) - c.mu.Unlock() - - if level == admissionFreezeAllNewScans { - return false, deferReasonMemoryFreeze + if c.level == admissionFreezeAllNewScans { + c.mu.Unlock() + return nil, false, deferReasonMemoryFreeze } - if level == admissionPauseWarming { - state.mu.Lock() - phase := state.phase - state.mu.Unlock() - if phase == changefeedPhaseWarming { - return false, deferReasonMemoryWarming + bytes := c.estimateScanSizeLocked(region, currentTs) + warming := isWarmingScan(region, currentTs) + if c.isWarmingScanBlockedLocked(warming, bytes) { + c.mu.Unlock() + return nil, false, deferReasonMemoryWarming + } + + state := c.getSubscriptionStateLocked(span) + lease := &memoryQuotaLease{} + lease.release = func() { + c.mu.Lock() + if c.scanUsed >= bytes { + c.scanUsed -= bytes + } else { + c.scanUsed = 0 + } + if warming { + if c.warmingScanUsed >= bytes { + c.warmingScanUsed -= bytes + } else { + c.warmingScanUsed = 0 + } + } + if state.scanUsed >= bytes { + state.scanUsed -= bytes + } else { + state.scanUsed = 0 } + delete(state.scanLeases, lease) + c.refreshLevelLocked() + c.mu.Unlock() + + c.onMemoryAvailable() + } + c.scanUsed += bytes + if warming { + c.warmingScanUsed += bytes } - return true, "" + state.scanUsed += bytes + state.scanLeases[lease] = bytes + c.refreshLevelLocked() + c.mu.Unlock() + return lease, true, "" } -func (c *memoryQuotaController) acquireEvent( - ctx context.Context, - span *subscribedSpan, - bytes uint64, -) *memoryQuotaLease { - if span == nil || span.meta.isSystem() || bytes == 0 { +func (c *memoryQuotaController) trackEvent(span *subscribedSpan, bytes uint64) *memoryQuotaLease { + if span.meta.isSystem() || bytes == 0 { return nil } c.mu.Lock() - for c.used+bytes > c.capacity && c.used > 0 { - if ctx.Err() != nil { - c.mu.Unlock() - return nil - } - c.cond.Wait() - } c.used += bytes c.refreshLevelLocked() - state := c.getOrCreateChangefeedStateLocked(span.meta.ChangefeedID) + state := c.getSubscriptionStateLocked(span) lease := &memoryQuotaLease{} lease.release = func() { @@ -283,40 +281,85 @@ func (c *memoryQuotaController) acquireEvent( } else { c.used = 0 } + delete(state.eventLeases, lease) c.refreshLevelLocked() - c.cond.Broadcast() c.mu.Unlock() - state.mu.Lock() - delete(state.leases, lease) - if state.memUsed >= bytes { - state.memUsed -= bytes - } else { - state.memUsed = 0 - } - state.mu.Unlock() - c.onMemoryAvailable() } - - state.mu.Lock() - state.leases[lease] = span.subID - state.memUsed += bytes - state.mu.Unlock() + state.eventLeases[lease] = bytes c.mu.Unlock() return lease } -func (c *memoryQuotaController) getOrCreateChangefeedStateLocked( - changefeedID common.ChangeFeedID, -) *changefeedQuotaState { - state := c.changefeeds[changefeedID] - if state != nil { - return state +func (c *memoryQuotaController) getSubscriptionStateLocked( + span *subscribedSpan, +) *subscriptionQuotaState { + return c.subscriptions[span.subID] +} + +func (c *memoryQuotaController) estimateScanSizeLocked(region regionInfo, currentTs uint64) uint64 { + raw := float64(c.scanEstimate) * scanLagFactor(region.resolvedTs(), currentTs) + + estimate := uint64(raw) + minEstimate := c.scanEstimate + if estimate < minEstimate { + estimate = minEstimate + } + + maxEstimate := c.scanEstimate * defaultMaxScanLagFactor + if estimate > maxEstimate { + estimate = maxEstimate + } + if estimate == 0 { + estimate = c.scanEstimate + } + return estimate +} + +func scanLagFactor(startTs uint64, currentTs uint64) float64 { + lag := scanLagDuration(startTs, currentTs) + if lag <= 0 { + return 1 + } + return min(defaultMaxScanLagFactor, 1+defaultScanLagWeight*math.Log2(1+float64(lag)/float64(defaultScanLagUnit))) +} + +func isWarmingScan(region regionInfo, currentTs uint64) bool { + span := region.subscribedSpan + if span.initialized.Load() { + return false + } + return scanLagDuration(region.resolvedTs(), currentTs) >= defaultWarmingScanLagThreshold +} + +func scanLagDuration(startTs uint64, currentTs uint64) time.Duration { + if startTs == 0 || currentTs <= startTs { + return 0 + } + lag := oracle.GetTimeFromTS(currentTs).Sub(oracle.GetTimeFromTS(startTs)) + if lag <= 0 { + return 0 + } + return lag +} + +func (c *memoryQuotaController) isWarmingScanBlockedLocked(warming bool, bytes uint64) bool { + if !warming { + return false + } + if c.level == admissionPauseWarming { + return true + } + return c.warmingScanUsed+bytes > c.warmingScanBudgetLocked() +} + +func (c *memoryQuotaController) warmingScanBudgetLocked() uint64 { + if c.capacity == 0 { + return 0 } - state = newChangefeedQuotaState(changefeedID) - c.changefeeds[changefeedID] = state - return state + budget := uint64(float64(c.capacity) * c.pauseWarmingRatio) + return max(budget, c.scanEstimate) } func (c *memoryQuotaController) refreshLevelLocked() { @@ -324,7 +367,8 @@ func (c *memoryQuotaController) refreshLevelLocked() { c.level = admissionNormal return } - usage := float64(c.used) / float64(c.capacity) + pressure := max(c.used, c.scanUsed) + usage := float64(pressure) / float64(c.capacity) switch c.level { case admissionFreezeAllNewScans: if usage <= c.resumeAllRatio { diff --git a/logservice/logpuller/memory_quota_test.go b/logservice/logpuller/memory_quota_test.go index 29cbdeb53c..9ea92dcf39 100644 --- a/logservice/logpuller/memory_quota_test.go +++ b/logservice/logpuller/memory_quota_test.go @@ -14,10 +14,10 @@ package logpuller import ( - "context" "testing" "time" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/pingcap/ticdc/pkg/common" "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/oracle" @@ -32,34 +32,58 @@ func newTestQuotaSpan(subID SubscriptionID, changefeedID common.ChangeFeedID) *s return span } +func newTestQuotaRegion(span *subscribedSpan) regionInfo { + state := ®ionlock.LockedRangeState{} + if span != nil { + state.ResolvedTs.Store(span.resolvedTs.Load()) + } + return regionInfo{ + subscribedSpan: span, + lockedRangeState: state, + } +} + +func setTestQuotaSpanLag(span *subscribedSpan, lag time.Duration) uint64 { + currentTime := time.Now() + span.resolvedTs.Store(oracle.GoTimeToTS(currentTime.Add(-lag))) + return oracle.GoTimeToTS(currentTime) +} + func TestMemoryQuotaAdmissionLevels(t *testing.T) { - controller := newMemoryQuotaController(100) + controller := newMemoryQuotaController(100, 0) + controller.scanEstimate = 10 warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) normalSpan := newTestQuotaSpan(2, common.NewChangeFeedIDWithName("normal", common.DefaultKeyspaceName)) normalSpan.initialized.Store(true) + currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) + normalCurrentTs := setTestQuotaSpanLag(normalSpan, defaultWarmingScanLagThreshold+time.Minute) controller.addSubscription(warmingSpan) controller.addSubscription(normalSpan) controller.markSubscriptionInitialized(normalSpan) - ok, reason := controller.allowNewScan(warmingSpan) + lease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) require.True(t, ok) require.Empty(t, reason) + lease.Release() - softLease := controller.acquireEvent(context.Background(), normalSpan, 20) + softLease := controller.trackEvent(normalSpan, 20) t.Cleanup(softLease.Release) - ok, reason = controller.allowNewScan(warmingSpan) + lease, ok, reason = controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) require.False(t, ok) + require.Nil(t, lease) require.Equal(t, deferReasonMemoryWarming, reason) - ok, reason = controller.allowNewScan(normalSpan) + lease, ok, reason = controller.acquireScan(newTestQuotaRegion(normalSpan), normalCurrentTs) require.True(t, ok) require.Empty(t, reason) + lease.Release() - hardLease := controller.acquireEvent(context.Background(), normalSpan, 70) + hardLease := controller.trackEvent(normalSpan, 70) t.Cleanup(hardLease.Release) - ok, reason = controller.allowNewScan(normalSpan) + lease, ok, reason = controller.acquireScan(newTestQuotaRegion(normalSpan), normalCurrentTs) require.False(t, ok) + require.Nil(t, lease) require.Equal(t, deferReasonMemoryFreeze, reason) hardLease.Release() @@ -71,33 +95,32 @@ func TestMemoryQuotaAdmissionLevels(t *testing.T) { require.Equal(t, admissionNormal, level) } -func TestWarmingHighPriorityTaskBlockedByMemoryGate(t *testing.T) { - controller := newMemoryQuotaController(100) +func TestHighLagUninitializedScanBlockedByMemoryGate(t *testing.T) { + controller := newMemoryQuotaController(100, 0) + controller.scanEstimate = 10 warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) controller.addSubscription(warmingSpan) - lease := controller.acquireEvent(context.Background(), warmingSpan, 20) + currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) + lease := controller.trackEvent(warmingSpan, 20) t.Cleanup(lease.Release) - scheduler := ®ionRequestScheduler{memoryQuota: controller} - region := regionInfo{subscribedSpan: warmingSpan} - task := newRegionPriorityTask(TaskHighPrior, region, oracle.GoTimeToTS(time.Now()), 1) - - ok, reason, err := scheduler.tryAdmitTask(context.Background(), &requestedStore{}, task, region) - require.NoError(t, err) + region := newTestQuotaRegion(warmingSpan) + scanLease, ok, reason := controller.acquireScan(region, currentTs) require.False(t, ok) + require.Nil(t, scanLease) require.Equal(t, deferReasonMemoryWarming, reason) } func TestRemoveSubscriptionReleasesOnlyItsOutstandingMemory(t *testing.T) { - controller := newMemoryQuotaController(100) + controller := newMemoryQuotaController(100, 0) changefeedID := common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName) span1 := newTestQuotaSpan(1, changefeedID) span2 := newTestQuotaSpan(2, changefeedID) controller.addSubscription(span1) controller.addSubscription(span2) - lease1 := controller.acquireEvent(context.Background(), span1, 30) - lease2 := controller.acquireEvent(context.Background(), span2, 40) + lease1 := controller.trackEvent(span1, 30) + lease2 := controller.trackEvent(span2, 40) t.Cleanup(lease2.Release) used, _, _ := controller.Snapshot() @@ -115,3 +138,125 @@ func TestRemoveSubscriptionReleasesOnlyItsOutstandingMemory(t *testing.T) { used, _, _ = controller.Snapshot() require.Equal(t, uint64(0), used) } + +func TestWarmingScanBudgetLimitsOutstandingScans(t *testing.T) { + controller := newMemoryQuotaController(150, 0) + controller.scanEstimate = 10 + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) + controller.addSubscription(warmingSpan) + currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) + + lease1, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) + require.True(t, ok) + require.Empty(t, reason) + t.Cleanup(lease1.Release) + lease2, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) + require.True(t, ok) + require.Empty(t, reason) + t.Cleanup(lease2.Release) + + lease3, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) + require.False(t, ok) + require.Nil(t, lease3) + require.Equal(t, deferReasonMemoryWarming, reason) + + lease1.Release() + lease3, ok, reason = controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) + require.True(t, ok) + require.Empty(t, reason) + require.NotNil(t, lease3) + lease3.Release() +} + +func TestInitializedSubscriptionBypassesWarmingScanBudget(t *testing.T) { + controller := newMemoryQuotaController(100, 0) + controller.scanEstimate = 10 + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName)) + normalSpan := newTestQuotaSpan(2, common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName)) + normalSpan.initialized.Store(true) + warmingCurrentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) + normalCurrentTs := setTestQuotaSpanLag(normalSpan, defaultWarmingScanLagThreshold+time.Minute) + controller.addSubscription(warmingSpan) + controller.addSubscription(normalSpan) + + lease := controller.trackEvent(normalSpan, 20) + t.Cleanup(lease.Release) + + scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), warmingCurrentTs) + require.False(t, ok) + require.Nil(t, scanLease) + require.Equal(t, deferReasonMemoryWarming, reason) + + scanLease, ok, reason = controller.acquireScan(newTestQuotaRegion(normalSpan), normalCurrentTs) + require.True(t, ok) + require.Empty(t, reason) + require.NotNil(t, scanLease) + scanUsed, warmingScanUsed, _, _ := controller.ScanSnapshot() + require.Equal(t, uint64(13), scanUsed) + require.Equal(t, uint64(0), warmingScanUsed) + scanLease.Release() +} + +func TestLowLagUninitializedSubscriptionBypassesWarmingGate(t *testing.T) { + controller := newMemoryQuotaController(100, 0) + controller.scanEstimate = 10 + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) + controller.addSubscription(warmingSpan) + currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold-time.Second) + + lease := controller.trackEvent(warmingSpan, 20) + t.Cleanup(lease.Release) + + scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) + require.True(t, ok) + require.Empty(t, reason) + require.NotNil(t, scanLease) + scanUsed, warmingScanUsed, _, _ := controller.ScanSnapshot() + require.Greater(t, scanUsed, uint64(0)) + require.Equal(t, uint64(0), warmingScanUsed) + scanLease.Release() +} + +func TestInitializedSubscriptionBlockedByMemoryFreeze(t *testing.T) { + controller := newMemoryQuotaController(100, 0) + normalSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName)) + normalSpan.initialized.Store(true) + controller.addSubscription(normalSpan) + + lease := controller.trackEvent(normalSpan, 90) + t.Cleanup(lease.Release) + + scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(normalSpan), 0) + require.False(t, ok) + require.Nil(t, scanLease) + require.Equal(t, deferReasonMemoryFreeze, reason) +} + +func TestWarmingScanBudgetKeepsAdmissionWideEnough(t *testing.T) { + controller := newMemoryQuotaController(defaultLogPullerMemoryQuota, 0) + warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) + controller.addSubscription(warmingSpan) + currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) + region := newTestQuotaRegion(warmingSpan) + _, _, warmingScanBudget, scanEstimate := controller.ScanSnapshot() + require.Equal(t, defaultScanBaseSize, scanEstimate) + scanSize := controller.estimateScanSizeLocked(region, currentTs) + allowedScans := int(warmingScanBudget / scanSize) + require.GreaterOrEqual(t, allowedScans, 35) + + var leases []*memoryQuotaLease + for range allowedScans { + lease, ok, reason := controller.acquireScan(region, currentTs) + require.True(t, ok) + require.Empty(t, reason) + leases = append(leases, lease) + } + lease, ok, reason := controller.acquireScan(region, currentTs) + require.False(t, ok) + require.Nil(t, lease) + require.Equal(t, deferReasonMemoryWarming, reason) + + for _, lease := range leases { + lease.Release() + } +} diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 073b49046f..9ab9e95262 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -61,11 +61,11 @@ type regionEvent struct { } func (event *regionEvent) needMemoryQuota() bool { - return event != nil && event.entries != nil + return event.entries != nil } func (event *regionEvent) releaseMemoryQuota() { - if event == nil || event.memoryQuota == nil { + if event.memoryQuota == nil { return } event.memoryQuota.Release() @@ -73,9 +73,6 @@ func (event *regionEvent) releaseMemoryQuota() { } func (event *regionEvent) getSize() int { - if event == nil { - return 0 - } size := int(unsafe.Sizeof(*event)) if event.entries != nil { size += int(unsafe.Sizeof(*event.entries)) @@ -426,9 +423,7 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u zap.Uint64("subscriptionID", uint64(span.subID)), zap.Uint64("regionID", regionID), zap.Uint64("resolvedTs", ts)) - if state.worker != nil && state.worker.memoryQuota != nil { - state.worker.memoryQuota.markSubscriptionInitialized(span) - } + state.worker.memoryQuota.markSubscriptionInitialized(span) } lastResolvedTs := span.resolvedTs.Load() nextResolvedPhyTs := oracle.ExtractPhysical(ts) diff --git a/logservice/logpuller/region_event_handler_test.go b/logservice/logpuller/region_event_handler_test.go index daca162f77..4c27612a2a 100644 --- a/logservice/logpuller/region_event_handler_test.go +++ b/logservice/logpuller/region_event_handler_test.go @@ -80,6 +80,7 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) { worker := ®ionRequestWorker{ requestCache: &requestCache{}, tracker: newRegionTracker(0), + memoryQuota: newMemoryQuotaController(0, 0), } region := newRegionInfo( tikv.RegionVerID{}, @@ -219,6 +220,7 @@ func TestHandleResolvedTs(t *testing.T) { worker := ®ionRequestWorker{ requestCache: &requestCache{}, tracker: newRegionTracker(0), + memoryQuota: newMemoryQuotaController(0, 0), } state1 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(1, 1, 1)}, uint64(subID1), worker, nil) { @@ -359,8 +361,10 @@ func TestHandleResolvedTsThrottled(t *testing.T) { advanceInterval: 100, } span.lastAdvanceTime.Store(0) - worker := ®ionRequestWorker{} - worker.tracker = newRegionTracker(0) + worker := ®ionRequestWorker{ + tracker: newRegionTracker(0), + memoryQuota: newMemoryQuotaController(0, 0), + } state := newRegionFeedState( regionInfo{ verID: tikv.NewRegionVerID(1, 1, 1), diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 0414599093..67cab44709 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -76,12 +76,9 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { } func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { - if event.needMemoryQuota() && s.memoryQuota != nil { + if event.needMemoryQuota() { span := event.mustFirstState().region.subscribedSpan - event.memoryQuota = s.memoryQuota.acquireEvent(s.ctx, span, uint64(event.getSize())) - if event.memoryQuota == nil && s.ctx.Err() != nil { - return - } + event.memoryQuota = s.memoryQuota.trackEvent(span, uint64(event.getSize())) } // fast path if !s.paused.Load() { @@ -137,9 +134,6 @@ func (s *regionEventSink) Metrics() dynstream.Metrics[int, SubscriptionID] { } func (s *regionEventSink) Close() { - if s.memoryQuota != nil { - s.memoryQuota.WakeAll() - } s.mu.Lock() s.paused.Store(false) s.cond.Broadcast() diff --git a/logservice/logpuller/region_request_cache.go b/logservice/logpuller/region_request_cache.go index 535d01408d..d76bb16709 100644 --- a/logservice/logpuller/region_request_cache.go +++ b/logservice/logpuller/region_request_cache.go @@ -55,16 +55,21 @@ type regionReq struct { // quota is acquired before the request enters requestCache and released // when the request leaves the worker window. quota *regionRequestQuota + // scanQuota is acquired by the memory quota controller before a warming + // scan enters requestCache. It is released when the scan initializes or is + // aborted. + scanQuota *memoryQuotaLease // stage is guarded by requestCache.mu. stage regionReqStage } -func newRegionReq(region regionInfo, quota *regionRequestQuota) *regionReq { +func newRegionReq(region regionInfo, quota *regionRequestQuota, scanQuota *memoryQuotaLease) *regionReq { return ®ionReq{ regionInfo: region, createTime: time.Now(), quota: quota, + scanQuota: scanQuota, stage: regionReqStageQueued, } } @@ -108,7 +113,11 @@ func newRequestCache(maxPendingCount int, onSpaceAvailable func()) *requestCache // add admits a region request into the worker window. func (c *requestCache) add( - ctx context.Context, region regionInfo, force bool, quota *regionRequestQuota, + ctx context.Context, + region regionInfo, + force bool, + quota *regionRequestQuota, + scanQuota *memoryQuotaLease, ) (bool, error) { start := time.Now() ticker := time.NewTicker(addReqRetryInterval) @@ -116,7 +125,7 @@ func (c *requestCache) add( retries := addReqRetryLimit for { - if c.tryAdd(region, force, quota) { + if c.tryAdd(region, force, quota, scanQuota) { metrics.SubscriptionClientAddRegionRequestDuration.Observe(time.Since(start).Seconds()) return true, nil } @@ -134,7 +143,12 @@ func (c *requestCache) add( } } -func (c *requestCache) tryAdd(region regionInfo, force bool, quota *regionRequestQuota) bool { +func (c *requestCache) tryAdd( + region regionInfo, + force bool, + quota *regionRequestQuota, + scanQuota *memoryQuotaLease, +) bool { c.mu.Lock() defer c.mu.Unlock() @@ -142,7 +156,7 @@ func (c *requestCache) tryAdd(region regionInfo, force bool, quota *regionReques return false } - req := newRegionReq(region, quota) + req := newRegionReq(region, quota, scanQuota) c.requests[req] = struct{}{} c.queue.Push(req) return true @@ -235,6 +249,9 @@ func (c *requestCache) remove(req *regionReq) bool { if removed { req.quota.Release() + if req.scanQuota != nil { + req.scanQuota.Release() + } c.notifySpace() } return removed @@ -281,6 +298,9 @@ func (c *requestCache) releaseRemovedReqs(removedReqs []*regionReq) { if len(removedReqs) > 0 { for _, req := range removedReqs { req.quota.Release() + if req.scanQuota != nil { + req.scanQuota.Release() + } } c.notifySpace() } diff --git a/logservice/logpuller/region_request_cache_test.go b/logservice/logpuller/region_request_cache_test.go index c600fe8566..440a4c3e9c 100644 --- a/logservice/logpuller/region_request_cache_test.go +++ b/logservice/logpuller/region_request_cache_test.go @@ -52,7 +52,7 @@ func TestRequestCacheAdd_NormalCase(t *testing.T) { region := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) require.Equal(t, 1, cache.pendingCount()) @@ -71,20 +71,20 @@ func TestRequestCacheAdd_ForceFlag(t *testing.T) { // Fill up the cache region1 := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region1, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region1, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 1, cache.pendingCount()) // Try to add another request without force - should fail due to retry limit region2 := createTestRegionInfo(1, 2) - ok, err = cache.add(ctx, region2, false, testRegionRequestQuota()) + ok, err = cache.add(ctx, region2, false, testRegionRequestQuota(), nil) require.False(t, ok) require.NoError(t, err) // With force=true, the request bypasses the live request limit. region3 := createTestRegionInfo(1, 3) - ok, err = cache.add(ctx, region3, true, testRegionRequestQuota()) + ok, err = cache.add(ctx, region3, true, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 2, cache.pendingCount()) @@ -108,7 +108,7 @@ func TestRequestCacheAdd_ContextCancellation(t *testing.T) { // Fill up the cache region1 := createTestRegionInfo(1, 1) ctx1 := context.Background() - ok, err := cache.add(ctx1, region1, false, testRegionRequestQuota()) + ok, err := cache.add(ctx1, region1, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) @@ -117,7 +117,7 @@ func TestRequestCacheAdd_ContextCancellation(t *testing.T) { cancel() // Cancel immediately region2 := createTestRegionInfo(1, 2) - ok, err = cache.add(ctx2, region2, false, testRegionRequestQuota()) + ok, err = cache.add(ctx2, region2, false, testRegionRequestQuota(), nil) require.False(t, ok) require.Error(t, err) require.Equal(t, context.Canceled, err) @@ -129,13 +129,13 @@ func TestRequestCacheAdd_RetryLimitExceeded(t *testing.T) { // Fill up the cache region1 := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region1, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region1, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) // Try to add another request - should eventually hit retry limit region2 := createTestRegionInfo(1, 2) - ok, err = cache.add(ctx, region2, false, testRegionRequestQuota()) + ok, err = cache.add(ctx, region2, false, testRegionRequestQuota(), nil) require.False(t, ok) require.NoError(t, err) } @@ -146,13 +146,13 @@ func TestRequestCacheAdd_SpaceAvailableNotification(t *testing.T) { // Fill up the cache region1 := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region1, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region1, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 1, cache.pendingCount()) region2 := createTestRegionInfo(1, 2) - ok, err = cache.add(ctx, region2, false, testRegionRequestQuota()) + ok, err = cache.add(ctx, region2, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 2, cache.pendingCount()) @@ -172,7 +172,7 @@ func TestRequestCacheAdd_SpaceAvailableNotification(t *testing.T) { // Now we should be able to add another request region3 := createTestRegionInfo(1, 3) - ok, err = cache.add(ctx, region3, false, testRegionRequestQuota()) + ok, err = cache.add(ctx, region3, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 2, cache.pendingCount()) @@ -189,7 +189,7 @@ func TestRequestCacheAdd_ConcurrentAdds(t *testing.T) { for i := range numGoroutines { go func(id int) { region := createTestRegionInfo(SubscriptionID(id%3), uint64(id)) - ok, err := cache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) done <- err @@ -215,11 +215,11 @@ func TestRequestCacheAdd_DuplicateQueuedRequestsAreTrackedIndependently(t *testi region := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) - ok, err = cache.add(ctx, region, false, testRegionRequestQuota()) + ok, err = cache.add(ctx, region, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 2, cache.pendingCount()) @@ -237,7 +237,7 @@ func TestRequestCacheAbortReleasesSlot(t *testing.T) { region := createTestRegionInfo(1, 1) - ok, err := cache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := cache.add(ctx, region, false, testRegionRequestQuota(), nil) require.True(t, ok) require.NoError(t, err) require.Equal(t, 1, cache.pendingCount()) @@ -263,7 +263,7 @@ func TestRequestCacheAbortReleasesQuotaOnce(t *testing.T) { }, } - ok, err := cache.add(ctx, region, false, quota) + ok, err := cache.add(ctx, region, false, quota, nil) require.True(t, ok) require.NoError(t, err) @@ -274,14 +274,36 @@ func TestRequestCacheAbortReleasesQuotaOnce(t *testing.T) { require.Equal(t, int32(1), releaseCount.Load()) } +func TestRequestCacheAbortReleasesScanQuotaOnce(t *testing.T) { + cache := newRequestCache(10, nil) + ctx := context.Background() + region := createTestRegionInfo(1, 1) + var releaseCount atomic.Int32 + scanQuota := &memoryQuotaLease{ + release: func() { + releaseCount.Add(1) + }, + } + + ok, err := cache.add(ctx, region, false, testRegionRequestQuota(), scanQuota) + require.True(t, ok) + require.NoError(t, err) + + req, err := cache.pop(ctx) + require.NoError(t, err) + require.True(t, cache.finishScan(req)) + require.False(t, cache.abortScan(req)) + require.Equal(t, int32(1), releaseCount.Load()) +} + func TestRequestCacheCloseDrainsQueuedRequests(t *testing.T) { cache := newRequestCache(10, nil) ctx := context.Background() - ok, err := cache.add(ctx, createTestRegionInfo(1, 1), false, testRegionRequestQuota()) + ok, err := cache.add(ctx, createTestRegionInfo(1, 1), false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) - ok, err = cache.add(ctx, createTestRegionInfo(1, 2), false, testRegionRequestQuota()) + ok, err = cache.add(ctx, createTestRegionInfo(1, 2), false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) require.Equal(t, 2, cache.queue.Len()) diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index 13ea4683b3..5aeb1ff7ce 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -248,8 +248,14 @@ func (s *regionRequestScheduler) handleDeferredTasks(ctx context.Context, store return err } if !ok { - store.PushPendingTask(task) s.observeDeferredTask(store, reason) + if reason == deferReasonMemoryWarming || reason == deferReasonMemoryFreeze { + if s.queue.Push(task) { + s.notifyScheduler() + } + return nil + } + store.PushPendingTask(task) return nil } } @@ -279,8 +285,14 @@ func (s *regionRequestScheduler) handleNewTask( return err } if !ok { - store.PushPendingTask(task) s.observeDeferredTask(store, reason) + if reason == deferReasonMemoryWarming || reason == deferReasonMemoryFreeze { + if s.queue.Push(task) { + s.notifyScheduler() + } + return nil + } + store.PushPendingTask(task) } return nil } @@ -292,18 +304,26 @@ func (s *regionRequestScheduler) tryAdmitTask( region regionInfo, ) (bool, string, error) { force := task.Priority() <= forcedPriorityBase - if s.memoryQuota != nil { - if ok, reason := s.memoryQuota.allowNewScan(region.subscribedSpan); !ok { - return false, reason, nil - } + var scanQuota *memoryQuotaLease + currentTs := s.upstream.pdClock.CurrentTS() + quota, ok, reason := s.memoryQuota.acquireScan(region, currentTs) + if !ok { + return false, reason, nil } + scanQuota = quota acquiredQuota, ok := store.quota.TryAcquire() if !ok { + if scanQuota != nil { + scanQuota.Release() + } return false, deferReasonStoreQuota, nil } - ok, worker, err := store.AddRegion(ctx, region, force, acquiredQuota) + ok, worker, err := store.AddRegion(ctx, region, force, acquiredQuota, scanQuota) if err != nil { acquiredQuota.Release() + if scanQuota != nil { + scanQuota.Release() + } log.Warn("subscription client add region request failed", zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)), zap.Uint64("regionID", region.verID.GetID()), @@ -312,6 +332,9 @@ func (s *regionRequestScheduler) tryAdmitTask( } if !ok { acquiredQuota.Release() + if scanQuota != nil { + scanQuota.Release() + } return false, deferReasonWorkerCache, nil } diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 7164a372c6..3bb9b8c646 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -83,7 +83,7 @@ func TestAddRegionStateReleasesOverwrittenRequest(t *testing.T) { ctx := context.Background() region := createTestRegionInfo(1, 1) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) req1, err := worker.requestCache.pop(ctx) @@ -92,7 +92,7 @@ func TestAddRegionStateReleasesOverwrittenRequest(t *testing.T) { state1 := newRegionFeedState(req1.regionInfo, uint64(req1.regionInfo.subscribedSpan.subID), worker, req1) worker.tracker.Track(req1.regionInfo.subscribedSpan.subID, req1.regionInfo.verID.GetID(), state1) - ok, err = worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err = worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) require.Equal(t, 2, worker.requestCache.pendingCount()) @@ -115,7 +115,7 @@ func TestInitializedRegionStateDetachesRequest(t *testing.T) { ctx := context.Background() region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) req, err := worker.requestCache.pop(ctx) @@ -141,7 +141,7 @@ func TestDrainUnsentRegionsReleaseSlotForProcessingRegion(t *testing.T) { ctx := context.Background() region := createTestRegionInfo(1, 1) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) @@ -322,7 +322,7 @@ func TestDrainUnsentRegionsDoesNotReturnStoppedSentRegion(t *testing.T) { ctx := context.Background() region := createTestRegionInfo(1, 1) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) @@ -359,7 +359,7 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { ctx := context.Background() region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) require.Equal(t, 1, worker.requestCache.pendingCount()) @@ -411,7 +411,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { ctx := context.Background() region := prepareRegionForSendTest(createTestRegionInfo(1, 1)) - ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, region, false, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) diff --git a/logservice/logpuller/requested_store.go b/logservice/logpuller/requested_store.go index 7cf713f16a..f2feb217e0 100644 --- a/logservice/logpuller/requested_store.go +++ b/logservice/logpuller/requested_store.go @@ -166,14 +166,18 @@ func (rs *requestedStore) PendingTaskCount() int { } func (rs *requestedStore) AddRegion( - ctx context.Context, region regionInfo, force bool, quota *regionRequestQuota, + ctx context.Context, + region regionInfo, + force bool, + quota *regionRequestQuota, + scanQuota *memoryQuotaLease, ) (bool, *regionRequestWorker, error) { workers := rs.requestWorkers start := int(rs.nextWorker.Add(1)) % len(workers) for i := range len(workers) { worker := workers[(start+i)%len(workers)] - ok, err := worker.requestCache.add(ctx, region, force, quota) + ok, err := worker.requestCache.add(ctx, region, force, quota, scanQuota) if err != nil || ok { return ok, worker, err } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index ca52428cd6..b484c6472a 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -86,6 +86,8 @@ type SubscriptionClientConfig struct { PendingRegionRequestQueueSize int // MemoryQuota is the log puller local memory quota in bytes. MemoryQuota uint64 + // ScanBaseSize is the base admission cost in bytes for one warming region scan. + ScanBaseSize uint64 } type upstreamHandle struct { @@ -170,7 +172,7 @@ func NewSubscriptionClient( } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) subClient.spanRegistry = newSpanRegistry(subClient.upstream) - subClient.memoryQuota = newMemoryQuotaController(config.MemoryQuota) + subClient.memoryQuota = newMemoryQuotaController(config.MemoryQuota, config.ScanBaseSize) subClient.failureHandler = newRegionFailureHandler( subClient.upstream, @@ -205,8 +207,13 @@ func (s *subscriptionClient) runMetricsUpdater(ctx context.Context) error { metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) used, capacity, _ := s.memoryQuota.Snapshot() + scanUsed, warmingScanUsed, warmingScanBudget, scanEstimate := s.memoryQuota.ScanSnapshot() metrics.LogPullerMemoryQuota.WithLabelValues("max").Set(float64(capacity)) metrics.LogPullerMemoryQuota.WithLabelValues("used").Set(float64(used)) + metrics.LogPullerMemoryQuota.WithLabelValues("scan_used").Set(float64(scanUsed)) + metrics.LogPullerMemoryQuota.WithLabelValues("warming_scan_used").Set(float64(warmingScanUsed)) + metrics.LogPullerMemoryQuota.WithLabelValues("warming_scan_budget").Set(float64(warmingScanBudget)) + metrics.LogPullerMemoryQuota.WithLabelValues("scan_estimate").Set(float64(scanEstimate)) metrics.DynamicStreamMemoryUsage.WithLabelValues( "log-puller", "max", @@ -260,9 +267,7 @@ func (s *subscriptionClient) Subscribe( bdrMode, ) s.spanRegistry.Add(rt) - if s.memoryQuota != nil { - s.memoryQuota.addSubscription(rt) - } + s.memoryQuota.addSubscription(rt) s.eventSink.AddPath(rt) @@ -343,9 +348,7 @@ func (s *subscriptionClient) onTableDrained(rt *subscribedSpan) { zap.Uint64("subscriptionID", uint64(rt.subID)), zap.Error(err)) } - if s.memoryQuota != nil { - s.memoryQuota.removeSubscription(rt) - } + s.memoryQuota.removeSubscription(rt) s.spanRegistry.Remove(rt.subID) } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 9b4757f26e..7e61175b22 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -307,6 +307,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { resolveLockTaskCh: make(chan resolveLockTask, 1), upstream: &upstreamHandle{pdClock: pdutil.NewClock4Test()}, eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + memoryQuota: newMemoryQuotaController(0, 0), } client.ctx, client.cancel = context.WithCancel(context.Background()) defer client.cancel() @@ -351,7 +352,8 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { client := &subscriptionClient{ - eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + memoryQuota: newMemoryQuotaController(0, 0), } client.spanRegistry = newSpanRegistry(&upstreamHandle{}) client.failureHandler = newRegionFailureHandler(&upstreamHandle{}, client.onTableDrained, nil, nil) @@ -366,6 +368,7 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { rangeLock: regionlock.NewRangeLock(1, rawSpan.StartKey, rawSpan.EndKey, 100), } client.spanRegistry.Add(span) + client.memoryQuota.addSubscription(span) res1 := span.rangeLock.LockRange(context.Background(), []byte("a"), []byte("m"), 1, 1) require.Equal(t, regionlock.LockRangeStatusSuccess, res1.Status) @@ -503,7 +506,7 @@ func TestEnqueueDeregisterToAllStoresUsesControlQueue(t *testing.T) { subscribedSpan: &subscribedSpan{subID: SubscriptionID(2)}, lockedRangeState: ®ionlock.LockedRangeState{}, } - ok, err := worker.requestCache.add(ctx, dummyRegion, true, testRegionRequestQuota()) + ok, err := worker.requestCache.add(ctx, dummyRegion, true, testRegionRequestQuota(), nil) require.NoError(t, err) require.True(t, ok) diff --git a/pkg/config/debug.go b/pkg/config/debug.go index e8d4eb482b..5b3881ed15 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -69,6 +69,8 @@ type PullerConfig struct { PendingRegionRequestQueueSize int `toml:"pending-region-request-queue-size" json:"pending_region_request_queue_size"` // MemoryQuota is the log puller local memory quota in bytes. MemoryQuota uint64 `toml:"memory-quota" json:"memory_quota"` + // ScanBaseSize is the base admission cost in bytes for one warming region scan. + ScanBaseSize uint64 `toml:"scan-base-size" json:"scan_base_size"` } // NewDefaultPullerConfig return the default puller configuration @@ -79,6 +81,7 @@ func NewDefaultPullerConfig() *PullerConfig { LogRegionDetails: false, PendingRegionRequestQueueSize: 32, // This value is chosen to reduce the impact of new changefeeds on existing ones. MemoryQuota: 1024 * 1024 * 1024, + ScanBaseSize: 4 * 1024 * 1024, } } diff --git a/server/server.go b/server/server.go index ab05f99bf7..93396dae88 100644 --- a/server/server.go +++ b/server/server.go @@ -202,6 +202,7 @@ func (c *server) initialize(ctx context.Context) error { RegionRequestWorkerPerStore: 8, PendingRegionRequestQueueSize: conf.Debug.Puller.PendingRegionRequestQueueSize, MemoryQuota: conf.Debug.Puller.MemoryQuota, + ScanBaseSize: conf.Debug.Puller.ScanBaseSize, }, c.pdClient, txnutil.NewLockerResolver(), c.security, From 0b2061ba7472389a2053b27391a2a3a5ca4e4eb8 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Tue, 7 Jul 2026 21:55:29 +0800 Subject: [PATCH 4/5] fix --- logservice/eventstore/pebble.go | 2 +- logservice/logpuller/memory_quota.go | 58 +++++++++++++++---- logservice/logpuller/memory_quota_test.go | 57 +++++++++++++----- logservice/logpuller/region_event_sink.go | 6 +- .../logpuller/region_request_worker_test.go | 6 +- logservice/logpuller/subscription_client.go | 3 +- .../logpuller/subscription_client_test.go | 6 +- pkg/config/debug.go | 2 +- 8 files changed, 106 insertions(+), 34 deletions(-) diff --git a/logservice/eventstore/pebble.go b/logservice/eventstore/pebble.go index f5f48bf71b..0d950c99d0 100644 --- a/logservice/eventstore/pebble.go +++ b/logservice/eventstore/pebble.go @@ -42,7 +42,7 @@ func newPebbleOptions(dbNum int) *pebble.Options { MaxOpenFiles: maxOpenFilesPerDB, - MaxConcurrentCompactions: func() int { return 6 }, + MaxConcurrentCompactions: func() int { return 2 }, // Decrease compaction frequency L0CompactionThreshold: 20, diff --git a/logservice/logpuller/memory_quota.go b/logservice/logpuller/memory_quota.go index 6b0ab04d3d..1351445bdb 100644 --- a/logservice/logpuller/memory_quota.go +++ b/logservice/logpuller/memory_quota.go @@ -14,6 +14,7 @@ package logpuller import ( + "context" "math" "sync" "sync/atomic" @@ -30,11 +31,12 @@ const ( defaultResumeWarmingRatio = 0.1 defaultFreezeAllRatio = 0.9 defaultResumeAllRatio = 0.7 + defaultHardLimitRatio = 5.0 - defaultScanBaseSize uint64 = 4 * 1024 * 1024 + defaultScanBaseSize uint64 = 8 * 1024 * 1024 defaultWarmingScanLagThreshold = 30 * time.Minute defaultScanLagUnit = 10 * time.Minute - defaultScanLagWeight = 0.15 + defaultScanLagWeight = 0.22 defaultMaxScanLagFactor = 16 ) @@ -98,12 +100,13 @@ func newSubscriptionQuotaState() *subscriptionQuotaState { } type memoryQuotaController struct { - mu sync.Mutex + mu sync.Mutex + cond *sync.Cond capacity uint64 // used tracks log puller bytes currently waiting for downstream callback. - // It is observed for admission decisions, but the receive path never waits - // on it after an event has arrived from TiKV. + // It is observed for admission decisions, and the receive path is blocked + // only when used reaches the hard limit. used uint64 // scanUsed tracks estimated bytes of admitted warming region scans that // have not finished initialization yet. @@ -115,6 +118,7 @@ type memoryQuotaController struct { resumeWarmingRatio float64 freezeAllRatio float64 resumeAllRatio float64 + hardLimitRatio float64 scanEstimate uint64 @@ -137,9 +141,11 @@ func newMemoryQuotaController(capacity uint64, scanBaseSize uint64) *memoryQuota resumeWarmingRatio: defaultResumeWarmingRatio, freezeAllRatio: defaultFreezeAllRatio, resumeAllRatio: defaultResumeAllRatio, + hardLimitRatio: defaultHardLimitRatio, scanEstimate: scanBaseSize, subscriptions: make(map[SubscriptionID]*subscriptionQuotaState), } + c.cond = sync.NewCond(&c.mu) return c } @@ -153,6 +159,12 @@ func (c *memoryQuotaController) onMemoryAvailable() { } } +func (c *memoryQuotaController) WakeAll() { + c.mu.Lock() + c.cond.Broadcast() + c.mu.Unlock() +} + func (c *memoryQuotaController) Snapshot() (used uint64, capacity uint64, level admissionLevel) { c.mu.Lock() defer c.mu.Unlock() @@ -164,10 +176,11 @@ func (c *memoryQuotaController) ScanSnapshot() ( warmingScanUsed uint64, warmingScanBudget uint64, scanEstimate uint64, + hardLimit uint64, ) { c.mu.Lock() defer c.mu.Unlock() - return c.scanUsed, c.warmingScanUsed, c.warmingScanBudgetLocked(), c.scanEstimate + return c.scanUsed, c.warmingScanUsed, c.warmingScanBudgetLocked(), c.scanEstimate, c.hardLimitLocked() } func (c *memoryQuotaController) addSubscription(span *subscribedSpan) { @@ -263,15 +276,29 @@ func (c *memoryQuotaController) acquireScan(region regionInfo, currentTs uint64) return lease, true, "" } -func (c *memoryQuotaController) trackEvent(span *subscribedSpan, bytes uint64) *memoryQuotaLease { - if span.meta.isSystem() || bytes == 0 { +func (c *memoryQuotaController) trackEvent( + ctx context.Context, + span *subscribedSpan, + bytes uint64, +) *memoryQuotaLease { + if bytes == 0 { return nil } c.mu.Lock() + for c.used+bytes > c.hardLimitLocked() && c.used > 0 { + if ctx.Err() != nil { + c.mu.Unlock() + return nil + } + c.cond.Wait() + } c.used += bytes c.refreshLevelLocked() - state := c.getSubscriptionStateLocked(span) + var state *subscriptionQuotaState + if !span.meta.isSystem() { + state = c.getSubscriptionStateLocked(span) + } lease := &memoryQuotaLease{} lease.release = func() { @@ -281,13 +308,18 @@ func (c *memoryQuotaController) trackEvent(span *subscribedSpan, bytes uint64) * } else { c.used = 0 } - delete(state.eventLeases, lease) + if state != nil { + delete(state.eventLeases, lease) + } c.refreshLevelLocked() + c.cond.Broadcast() c.mu.Unlock() c.onMemoryAvailable() } - state.eventLeases[lease] = bytes + if state != nil { + state.eventLeases[lease] = bytes + } c.mu.Unlock() return lease } @@ -362,6 +394,10 @@ func (c *memoryQuotaController) warmingScanBudgetLocked() uint64 { return max(budget, c.scanEstimate) } +func (c *memoryQuotaController) hardLimitLocked() uint64 { + return uint64(float64(c.capacity) * c.hardLimitRatio) +} + func (c *memoryQuotaController) refreshLevelLocked() { if c.capacity == 0 { c.level = admissionNormal diff --git a/logservice/logpuller/memory_quota_test.go b/logservice/logpuller/memory_quota_test.go index 9ea92dcf39..c1bb64ca7d 100644 --- a/logservice/logpuller/memory_quota_test.go +++ b/logservice/logpuller/memory_quota_test.go @@ -14,6 +14,7 @@ package logpuller import ( + "context" "testing" "time" @@ -68,7 +69,7 @@ func TestMemoryQuotaAdmissionLevels(t *testing.T) { require.Empty(t, reason) lease.Release() - softLease := controller.trackEvent(normalSpan, 20) + softLease := controller.trackEvent(context.Background(), normalSpan, 20) t.Cleanup(softLease.Release) lease, ok, reason = controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) require.False(t, ok) @@ -79,7 +80,7 @@ func TestMemoryQuotaAdmissionLevels(t *testing.T) { require.Empty(t, reason) lease.Release() - hardLease := controller.trackEvent(normalSpan, 70) + hardLease := controller.trackEvent(context.Background(), normalSpan, 70) t.Cleanup(hardLease.Release) lease, ok, reason = controller.acquireScan(newTestQuotaRegion(normalSpan), normalCurrentTs) require.False(t, ok) @@ -101,7 +102,7 @@ func TestHighLagUninitializedScanBlockedByMemoryGate(t *testing.T) { warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) controller.addSubscription(warmingSpan) currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) - lease := controller.trackEvent(warmingSpan, 20) + lease := controller.trackEvent(context.Background(), warmingSpan, 20) t.Cleanup(lease.Release) region := newTestQuotaRegion(warmingSpan) @@ -119,8 +120,8 @@ func TestRemoveSubscriptionReleasesOnlyItsOutstandingMemory(t *testing.T) { controller.addSubscription(span1) controller.addSubscription(span2) - lease1 := controller.trackEvent(span1, 30) - lease2 := controller.trackEvent(span2, 40) + lease1 := controller.trackEvent(context.Background(), span1, 30) + lease2 := controller.trackEvent(context.Background(), span2, 40) t.Cleanup(lease2.Release) used, _, _ := controller.Snapshot() @@ -139,6 +140,36 @@ func TestRemoveSubscriptionReleasesOnlyItsOutstandingMemory(t *testing.T) { require.Equal(t, uint64(0), used) } +func TestTrackEventBlocksAtHardLimit(t *testing.T) { + controller := newMemoryQuotaController(100, 0) + controller.hardLimitRatio = 2 + span := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("cf", common.DefaultKeyspaceName)) + controller.addSubscription(span) + + lease := controller.trackEvent(context.Background(), span, 200) + t.Cleanup(lease.Release) + + acquired := make(chan struct{}) + go func() { + blockedLease := controller.trackEvent(context.Background(), span, 1) + blockedLease.Release() + close(acquired) + }() + + select { + case <-acquired: + t.Fatal("trackEvent should block at hard limit") + case <-time.After(100 * time.Millisecond): + } + + lease.Release() + select { + case <-acquired: + case <-time.After(time.Second): + t.Fatal("trackEvent should resume after memory is released") + } +} + func TestWarmingScanBudgetLimitsOutstandingScans(t *testing.T) { controller := newMemoryQuotaController(150, 0) controller.scanEstimate = 10 @@ -179,7 +210,7 @@ func TestInitializedSubscriptionBypassesWarmingScanBudget(t *testing.T) { controller.addSubscription(warmingSpan) controller.addSubscription(normalSpan) - lease := controller.trackEvent(normalSpan, 20) + lease := controller.trackEvent(context.Background(), normalSpan, 20) t.Cleanup(lease.Release) scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), warmingCurrentTs) @@ -191,8 +222,8 @@ func TestInitializedSubscriptionBypassesWarmingScanBudget(t *testing.T) { require.True(t, ok) require.Empty(t, reason) require.NotNil(t, scanLease) - scanUsed, warmingScanUsed, _, _ := controller.ScanSnapshot() - require.Equal(t, uint64(13), scanUsed) + scanUsed, warmingScanUsed, _, _, _ := controller.ScanSnapshot() + require.Equal(t, controller.estimateScanSizeLocked(newTestQuotaRegion(normalSpan), normalCurrentTs), scanUsed) require.Equal(t, uint64(0), warmingScanUsed) scanLease.Release() } @@ -204,14 +235,14 @@ func TestLowLagUninitializedSubscriptionBypassesWarmingGate(t *testing.T) { controller.addSubscription(warmingSpan) currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold-time.Second) - lease := controller.trackEvent(warmingSpan, 20) + lease := controller.trackEvent(context.Background(), warmingSpan, 20) t.Cleanup(lease.Release) scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) require.True(t, ok) require.Empty(t, reason) require.NotNil(t, scanLease) - scanUsed, warmingScanUsed, _, _ := controller.ScanSnapshot() + scanUsed, warmingScanUsed, _, _, _ := controller.ScanSnapshot() require.Greater(t, scanUsed, uint64(0)) require.Equal(t, uint64(0), warmingScanUsed) scanLease.Release() @@ -223,7 +254,7 @@ func TestInitializedSubscriptionBlockedByMemoryFreeze(t *testing.T) { normalSpan.initialized.Store(true) controller.addSubscription(normalSpan) - lease := controller.trackEvent(normalSpan, 90) + lease := controller.trackEvent(context.Background(), normalSpan, 90) t.Cleanup(lease.Release) scanLease, ok, reason := controller.acquireScan(newTestQuotaRegion(normalSpan), 0) @@ -238,11 +269,11 @@ func TestWarmingScanBudgetKeepsAdmissionWideEnough(t *testing.T) { controller.addSubscription(warmingSpan) currentTs := setTestQuotaSpanLag(warmingSpan, defaultWarmingScanLagThreshold+time.Minute) region := newTestQuotaRegion(warmingSpan) - _, _, warmingScanBudget, scanEstimate := controller.ScanSnapshot() + _, _, warmingScanBudget, scanEstimate, _ := controller.ScanSnapshot() require.Equal(t, defaultScanBaseSize, scanEstimate) scanSize := controller.estimateScanSizeLocked(region, currentTs) allowedScans := int(warmingScanBudget / scanSize) - require.GreaterOrEqual(t, allowedScans, 35) + require.GreaterOrEqual(t, allowedScans, 17) var leases []*memoryQuotaLease for range allowedScans { diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 67cab44709..a3160c3533 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -78,7 +78,10 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { if event.needMemoryQuota() { span := event.mustFirstState().region.subscribedSpan - event.memoryQuota = s.memoryQuota.trackEvent(span, uint64(event.getSize())) + event.memoryQuota = s.memoryQuota.trackEvent(s.ctx, span, uint64(event.getSize())) + if event.memoryQuota == nil && s.ctx.Err() != nil { + return + } } // fast path if !s.paused.Load() { @@ -134,6 +137,7 @@ func (s *regionEventSink) Metrics() dynstream.Metrics[int, SubscriptionID] { } func (s *regionEventSink) Close() { + s.memoryQuota.WakeAll() s.mu.Lock() s.paused.Store(false) s.cond.Broadcast() diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 3bb9b8c646..09c0b9a7e5 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -205,7 +205,7 @@ func (m *mockRegionEventDynamicStream) GetMetrics() dynstream.Metrics[int, Subsc func newDispatchResolvedTsTestWorker(regionCount int) (*regionRequestWorker, *mockRegionEventDynamicStream, *cdcpb.ResolvedTs) { ds := &mockRegionEventDynamicStream{} worker := ®ionRequestWorker{ - eventSink: ®ionEventSink{ctx: context.Background(), ds: ds}, + eventSink: ®ionEventSink{ctx: context.Background(), ds: ds, memoryQuota: newMemoryQuotaController(0, 0)}, tracker: newRegionTracker(0), } regions := make([]uint64, regionCount) @@ -351,7 +351,7 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { controlQueue: newControlQueue(), storeAddr: "store-1", upstream: &upstreamHandle{}, - eventSink: ®ionEventSink{ctx: context.Background(), ds: &mockRegionEventDynamicStream{}}, + eventSink: ®ionEventSink{ctx: context.Background(), ds: &mockRegionEventDynamicStream{}, memoryQuota: newMemoryQuotaController(0, 0)}, failureHandler: newRegionFailureHandler(&upstreamHandle{}, func(*subscribedSpan) {}, nil, nil), tracker: newRegionTracker(0), } @@ -403,7 +403,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { controlQueue: newControlQueue(), storeAddr: "store-1", upstream: &upstreamHandle{}, - eventSink: ®ionEventSink{ctx: context.Background(), ds: &mockRegionEventDynamicStream{}}, + eventSink: ®ionEventSink{ctx: context.Background(), ds: &mockRegionEventDynamicStream{}, memoryQuota: newMemoryQuotaController(0, 0)}, failureHandler: newRegionFailureHandler(&upstreamHandle{}, func(*subscribedSpan) {}, nil, nil), tracker: newRegionTracker(0), } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index b484c6472a..8b0d0e043d 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -207,13 +207,14 @@ func (s *subscriptionClient) runMetricsUpdater(ctx context.Context) error { metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) used, capacity, _ := s.memoryQuota.Snapshot() - scanUsed, warmingScanUsed, warmingScanBudget, scanEstimate := s.memoryQuota.ScanSnapshot() + scanUsed, warmingScanUsed, warmingScanBudget, scanEstimate, hardLimit := s.memoryQuota.ScanSnapshot() metrics.LogPullerMemoryQuota.WithLabelValues("max").Set(float64(capacity)) metrics.LogPullerMemoryQuota.WithLabelValues("used").Set(float64(used)) metrics.LogPullerMemoryQuota.WithLabelValues("scan_used").Set(float64(scanUsed)) metrics.LogPullerMemoryQuota.WithLabelValues("warming_scan_used").Set(float64(warmingScanUsed)) metrics.LogPullerMemoryQuota.WithLabelValues("warming_scan_budget").Set(float64(warmingScanBudget)) metrics.LogPullerMemoryQuota.WithLabelValues("scan_estimate").Set(float64(scanEstimate)) + metrics.LogPullerMemoryQuota.WithLabelValues("hard_limit").Set(float64(hardLimit)) metrics.DynamicStreamMemoryUsage.WithLabelValues( "log-puller", "max", diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 7e61175b22..4ae60fbfa2 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -306,7 +306,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { client := &subscriptionClient{ resolveLockTaskCh: make(chan resolveLockTask, 1), upstream: &upstreamHandle{pdClock: pdutil.NewClock4Test()}, - eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + eventSink: ®ionEventSink{ds: &mockDynamicStream{}, memoryQuota: newMemoryQuotaController(0, 0)}, memoryQuota: newMemoryQuotaController(0, 0), } client.ctx, client.cancel = context.WithCancel(context.Background()) @@ -352,7 +352,7 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { client := &subscriptionClient{ - eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, + eventSink: ®ionEventSink{ds: &mockDynamicStream{}, memoryQuota: newMemoryQuotaController(0, 0)}, memoryQuota: newMemoryQuotaController(0, 0), } client.spanRegistry = newSpanRegistry(&upstreamHandle{}) @@ -462,7 +462,7 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) - sink := ®ionEventSink{ctx: ctx, ds: &mockDynamicStream{}} + sink := ®ionEventSink{ctx: ctx, ds: &mockDynamicStream{}, memoryQuota: newMemoryQuotaController(0, 0)} sink.cond = sync.NewCond(&sink.mu) client := &subscriptionClient{cancel: cancel, eventSink: sink} diff --git a/pkg/config/debug.go b/pkg/config/debug.go index 5b3881ed15..4463d14d3b 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -81,7 +81,7 @@ func NewDefaultPullerConfig() *PullerConfig { LogRegionDetails: false, PendingRegionRequestQueueSize: 32, // This value is chosen to reduce the impact of new changefeeds on existing ones. MemoryQuota: 1024 * 1024 * 1024, - ScanBaseSize: 4 * 1024 * 1024, + ScanBaseSize: 8 * 1024 * 1024, } } From 5804874444e7481fe35a6168f052385f1173d003 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Wed, 8 Jul 2026 08:34:57 +0800 Subject: [PATCH 5/5] try fix --- logservice/logpuller/memory_quota.go | 2 +- logservice/logpuller/memory_quota_test.go | 3 +-- logservice/logpuller/region_event_handler.go | 8 ++------ .../logpuller/region_event_handler_test.go | 15 +++++++++++++++ logservice/logpuller/span_registry.go | 18 ++++++++++++++++++ 5 files changed, 37 insertions(+), 9 deletions(-) diff --git a/logservice/logpuller/memory_quota.go b/logservice/logpuller/memory_quota.go index 1351445bdb..4c2a4bef4e 100644 --- a/logservice/logpuller/memory_quota.go +++ b/logservice/logpuller/memory_quota.go @@ -215,7 +215,7 @@ func (c *memoryQuotaController) removeSubscription(span *subscribedSpan) { } } -func (c *memoryQuotaController) markSubscriptionInitialized(span *subscribedSpan) { +func (c *memoryQuotaController) markSubscriptionInitialized() { c.onMemoryAvailable() } diff --git a/logservice/logpuller/memory_quota_test.go b/logservice/logpuller/memory_quota_test.go index c1bb64ca7d..ee07782fc2 100644 --- a/logservice/logpuller/memory_quota_test.go +++ b/logservice/logpuller/memory_quota_test.go @@ -62,7 +62,6 @@ func TestMemoryQuotaAdmissionLevels(t *testing.T) { controller.addSubscription(warmingSpan) controller.addSubscription(normalSpan) - controller.markSubscriptionInitialized(normalSpan) lease, ok, reason := controller.acquireScan(newTestQuotaRegion(warmingSpan), currentTs) require.True(t, ok) @@ -96,7 +95,7 @@ func TestMemoryQuotaAdmissionLevels(t *testing.T) { require.Equal(t, admissionNormal, level) } -func TestHighLagUninitializedScanBlockedByMemoryGate(t *testing.T) { +func TestHighLagScanBlockedByMemoryGate(t *testing.T) { controller := newMemoryQuotaController(100, 0) controller.scanEstimate = 10 warmingSpan := newTestQuotaSpan(1, common.NewChangeFeedIDWithName("warming", common.DefaultKeyspaceName)) diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 9ab9e95262..7fce1d11b3 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -418,12 +418,8 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u } if shouldAdvance { - if ts > 0 && span.initialized.CompareAndSwap(false, true) { - log.Info("subscription client is initialized", - zap.Uint64("subscriptionID", uint64(span.subID)), - zap.Uint64("regionID", regionID), - zap.Uint64("resolvedTs", ts)) - state.worker.memoryQuota.markSubscriptionInitialized(span) + if span.tryMarkInitialized(regionID, ts) { + state.worker.memoryQuota.markSubscriptionInitialized() } lastResolvedTs := span.resolvedTs.Load() nextResolvedPhyTs := oracle.ExtractPhysical(ts) diff --git a/logservice/logpuller/region_event_handler_test.go b/logservice/logpuller/region_event_handler_test.go index 4c27612a2a..acb9ca7980 100644 --- a/logservice/logpuller/region_event_handler_test.go +++ b/logservice/logpuller/region_event_handler_test.go @@ -357,6 +357,7 @@ func TestHandleResolvedTsThrottled(t *testing.T) { span := &subscribedSpan{ subID: SubscriptionID(1), + startTs: 100, rangeLock: l, advanceInterval: 100, } @@ -377,4 +378,18 @@ func TestHandleResolvedTsThrottled(t *testing.T) { ) require.Equal(t, uint64(200), handleResolvedTs(span, state, 300)) + require.True(t, span.initialized.Load()) +} + +func TestTryMarkSpanInitializedByResolvedTs(t *testing.T) { + span := &subscribedSpan{ + subID: SubscriptionID(1), + startTs: 100, + } + + require.False(t, span.tryMarkInitialized(1, 100)) + require.False(t, span.initialized.Load()) + require.True(t, span.tryMarkInitialized(1, 101)) + require.True(t, span.initialized.Load()) + require.False(t, span.tryMarkInitialized(1, 102)) } diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go index b5cf01d018..e71dbe38f3 100644 --- a/logservice/logpuller/span_registry.go +++ b/logservice/logpuller/span_registry.go @@ -156,6 +156,24 @@ func (span *subscribedSpan) resolveStaleLocks(targetTs uint64) { zap.Any("ranges", res)) } +func (span *subscribedSpan) tryMarkInitialized(regionID uint64, resolvedTs uint64) bool { + if resolvedTs <= span.startTs { + return false + } + if span.initialized.Load() { + return false + } + if !span.initialized.CompareAndSwap(false, true) { + return false + } + log.Info("subscription client is initialized", + zap.Uint64("subscriptionID", uint64(span.subID)), + zap.Uint64("regionID", regionID), + zap.Uint64("resolvedTs", resolvedTs), + zap.Uint64("startTs", span.startTs)) + return true +} + func newSpanRegistry(upstream *upstreamHandle) *spanRegistry { return &spanRegistry{ spans: make(map[SubscriptionID]*subscribedSpan),