[WIP] memory quota optimization #5589
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
|
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 |
|
[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 |
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 introduces a local memory quota controller (memoryQuotaController) for the log puller in TiCDC to manage memory usage of changefeed subscriptions and control admissions (e.g., pausing warming or freezing new scans) based on memory usage ratios. The feedback highlights critical issues and optimization opportunities: potential memory leaks in allowNewScan and acquireEvent where calling getOrCreateChangefeedStateLocked on removed subscriptions recreates and leaks state, a bug in acquireEvent where it should abort if the subscription is removed while blocked, and an optimization in regionEventHandler.Handle to store *memoryQuotaLease pointers directly instead of copying large regionEvent structs to reduce heap escapes and GC pressure.
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.
| 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, "" | ||
| } |
There was a problem hiding this comment.
The allowNewScan method calls getOrCreateChangefeedStateLocked which creates a new changefeedQuotaState if it does not exist. If a subscription has already been removed, calling allowNewScan will recreate the state and put it back into c.changefeeds. Since removeSubscription is only called once when the subscription is removed, this recreated state will never be cleaned up, leading to a permanent memory leak. To fix this, check if the subscription is still active before accessing or creating the state.
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, ok := c.changefeeds[span.meta.ChangefeedID]
if !ok {
c.mu.Unlock()
return true, ""
}
state.mu.Lock()
if _, ok := state.spans[span.subID]; !ok {
state.mu.Unlock()
c.mu.Unlock()
return true, ""
}
phase := state.phase
state.mu.Unlock()
c.mu.Unlock()
if level == admissionFreezeAllNewScans {
return false, deferReasonMemoryFreeze
}
if level == admissionPauseWarming {
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 | ||
| } |
There was a problem hiding this comment.
The acquireEvent method calls getOrCreateChangefeedStateLocked which creates a new changefeedQuotaState if it does not exist. If a subscription has already been removed, calling acquireEvent will recreate the state and put it back into c.changefeeds, leaking it permanently. Additionally, if the subscription is removed while acquireEvent is blocked on c.cond.Wait(), it should abort and return nil instead of acquiring the quota. To fix this, check if the subscription is still active before the loop and after waking up from c.cond.Wait().
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()
state, ok := c.changefeeds[span.meta.ChangefeedID]
if !ok {
c.mu.Unlock()
return nil
}
state.mu.Lock()
if _, ok := state.spans[span.subID]; !ok {
state.mu.Unlock()
c.mu.Unlock()
return nil
}
state.mu.Unlock()
for c.used+bytes > c.capacity && c.used > 0 {
if ctx.Err() != nil {
c.mu.Unlock()
return nil
}
c.cond.Wait()
state, ok = c.changefeeds[span.meta.ChangefeedID]
if !ok {
c.mu.Unlock()
return nil
}
state.mu.Lock()
if _, ok := state.spans[span.subID]; !ok {
state.mu.Unlock()
c.mu.Unlock()
return nil
}
state.mu.Unlock()
}
c.used += bytes
c.refreshLevelLocked()
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
}| 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.
Instead of copying the large regionEvent structs and taking their addresses (which causes them to escape to the heap and increases GC pressure), you can store the *memoryQuotaLease pointers directly in a slice. This is much more efficient and idiomatic.
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, avoiding unnecessary struct dereferencing.
| 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