[WIP] memory quota optimization#5584
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
do-not-merge/work-in-progress
|
[FORMAT CHECKER NOTIFICATION] Notice: To remove the 📖 For more info, you can check the "Contribute Code" section in the development guide. |
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the log puller's resource control and request scheduling in TiCDC, introducing a local memory quota controller and a decentralized, priority-based store scheduler with per-store request workers and local request caches. The review feedback highlights several critical improvements: guarding against a divide-by-zero panic in requestedStore, preventing an infinite rescheduling loop for stopped subscriptions in regionFailureHandler, checking for context cancellation early in acquireEvent, and using the loop's context in the scheduler. Additionally, the reviewer suggests removing dead code in regionFailureHandler and optimizing memory allocations in regionEventHandler by storing quota leases directly.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| regionRequestWorkerPerStore := scheduler.config.RegionRequestWorkerPerStore | ||
| perWorkerQueueSize := pendingRegionRequestQueueSize / int(regionRequestWorkerPerStore) |
There was a problem hiding this comment.
If regionRequestWorkerPerStore is configured as 0, perWorkerQueueSize := pendingRegionRequestQueueSize / int(regionRequestWorkerPerStore) will panic with a divide-by-zero error. We should guard against this by ensuring regionRequestWorkerPerStore is at least 1.
regionRequestWorkerPerStore := scheduler.config.RegionRequestWorkerPerStore
if regionRequestWorkerPerStore == 0 {
regionRequestWorkerPerStore = 1
}
perWorkerQueueSize := pendingRegionRequestQueueSize / int(regionRequestWorkerPerStore)| } | ||
| r.cache.add(errInfo) | ||
| } |
There was a problem hiding this comment.
If the subscription is already stopped (errInfo.subscribedSpan.stopped.Load() is true), and UnlockRange returns false, we should not add the error to r.cache. Doing so causes handleError to reschedule the region request for a stopped subscription, leading to an infinite loop of rescheduling and reporting. Returning early when the subscription is stopped prevents this.
if errInfo.subscribedSpan.stopped.Load() {
return
}
r.cache.add(errInfo)| c.mu.Lock() | ||
| for c.used+bytes > c.capacity && c.used > 0 { |
There was a problem hiding this comment.
In acquireEvent, if the context ctx is already cancelled when the function is called, it might still allocate memory quota and return a lease if c.used+bytes <= c.capacity (since the loop is skipped). We should check ctx.Err() at the beginning of the lock block to prevent acquiring quota on a cancelled context.
| c.mu.Lock() | |
| for c.used+bytes > c.capacity && c.used > 0 { | |
| c.mu.Lock() | |
| if ctx.Err() != nil { | |
| c.mu.Unlock() | |
| return nil | |
| } | |
| for c.used+bytes > c.capacity && c.used > 0 { |
| // Run is the only consumer of storeAvailable, so Len > 0 means this | ||
| // GetWithContext will not block. | ||
| if s.storeAvailable.Len() > 0 { | ||
| store, ok, err := s.storeAvailable.GetWithContext(context.Background()) |
There was a problem hiding this comment.
Using context.Background() instead of the loop's ctx in GetWithContext prevents the call from being interrupted if ctx is cancelled. Even if Len() > 0 makes it unlikely to block under normal conditions, using ctx is much safer and ensures robust shutdown behavior.
| store, ok, err := s.storeAvailable.GetWithContext(context.Background()) | |
| store, ok, err := s.storeAvailable.GetWithContext(ctx) |
| type errCache struct { | ||
| sync.Mutex | ||
| cache []regionErrorInfo | ||
| errCh chan regionErrorInfo |
There was a problem hiding this comment.
The errCh channel, along with dispatchBatch and dispatch methods in errCache, are completely unused in the new implementation because failureHandler.Run directly pops and processes batches from r.cache.popBatch inline. Removing this dead code will improve maintainability.
| quotaEvents := make([]*regionEvent, 0, len(events)) | ||
| for _, event := range events { | ||
| event := event | ||
| if event.memoryQuota != nil { | ||
| quotaEvents = append(quotaEvents, &event) | ||
| } |
There was a problem hiding this comment.
Defining quotaEvents as []*regionEvent and taking the address of a loop-local copy of event (event := event) causes these local copies to escape to the heap because releaseMemoryQuota is called asynchronously inside the consumeKVEvents callback. Since we only need to release the memory quota, we can store *memoryQuotaLease directly in a quotaLeases slice, avoiding any heap allocation of regionEvent copies.
quotaLeases := make([]*memoryQuotaLease, 0, len(events))
for _, event := range events {
if event.memoryQuota != nil {
quotaLeases = append(quotaLeases, event.memoryQuota)
}| releaseMemoryQuota := func() { | ||
| for _, event := range quotaEvents { | ||
| event.releaseMemoryQuota() | ||
| } | ||
| } |
There was a problem hiding this comment.
Update releaseMemoryQuota to release the leases directly from the quotaLeases slice.
| releaseMemoryQuota := func() { | |
| for _, event := range quotaEvents { | |
| event.releaseMemoryQuota() | |
| } | |
| } | |
| releaseMemoryQuota := func() { | |
| for _, lease := range quotaLeases { | |
| lease.Release() | |
| } | |
| } |
1 participant
What problem does this PR solve?
Issue Number: close #xxx
What is changed and how it works?
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note