diff --git a/docs/design/2026-08-12-ticdc-log-puller-memory-quota.md b/docs/design/2026-08-12-ticdc-log-puller-memory-quota.md new file mode 100644 index 0000000000..e4ef5e3bce --- /dev/null +++ b/docs/design/2026-08-12-ticdc-log-puller-memory-quota.md @@ -0,0 +1,429 @@ +# TiCDC Log Puller Memory Quota Design + +This document describes the memory quota mechanism implemented by the +new-architecture Log Puller. The mechanism is owned by +[`memoryQuotaController`](../../logservice/logpuller/memory_quota.go) and is +shared by the Region event receive path and Region initial-scan admission. + +Related code includes: + +- [`region_event_sink.go`](../../logservice/logpuller/region_event_sink.go): + accounts entry events before they enter the dynamic stream. +- [`region_event_handler.go`](../../logservice/logpuller/region_event_handler.go): + releases event memory after downstream consumption or event drop. +- [`region_admission_controller.go`](../../logservice/logpuller/region_admission_controller.go): + acquires and releases estimated initial-scan memory. +- [`scan_priority.go`](../../logservice/logpuller/scan_priority.go): determines + whether a Region scan has high or low priority. +- [`pkg/config/debug.go`](../../pkg/config/debug.go): defines the Log Puller + memory quota configuration. +- [`pkg/metrics/log_puller.go`](../../pkg/metrics/log_puller.go): defines the + quota metrics. + +## 1. Background + +The Log Puller has two sources of memory pressure: + +1. TiKV entry events that have been received but are still retained by the Log + Puller or its downstream consumer. +2. Region initial scans that have been admitted but have not yet completed. + +The first source is measurable after an event arrives. The second must be +controlled before data arrives, so it is represented by a memory estimate. +Controlling only one source is insufficient: + +- Limiting only buffered events reacts too late when many initial scans start + concurrently. +- Limiting only initial scans does not protect the process when downstream + consumption stalls and received events remain retained for a long time. + +The controller therefore combines event accounting with scan admission while +keeping their hot paths and wake-up conditions separate. + +## 2. Goals and non-goals + +The design has the following goals: + +1. Bound the growth of retained Region entry events when downstream is slow. +2. Reduce the number of new low-priority initial scans before event memory + reaches the receive-path hard limit. +3. Preserve progress for high-priority recovery and caught-up workloads. +4. Keep event accounting inexpensive because it runs once per received entry + batch. +5. Wake blocked goroutines without lost notifications during release, + cancellation, or subscription shutdown. +6. Make ownership explicit so every successful acquisition has exactly one + release path. + +The mechanism is not intended to: + +- Measure the complete Go heap or process RSS. +- Enforce a strict upper bound equal to the configured soft quota. +- Provide a separate quota or fairness policy for each subscription. +- Reclaim events or cancel active scans when pressure increases. +- Replace the Region request window maintained by each request worker. + +## 3. Architecture + +One `subscriptionClient` creates one `memoryQuotaController` and shares it with +the event sink and all Region request workers. + +```mermaid +flowchart LR + T[TiKV event stream] --> S[regionEventSink] + S -->|AcquireEvent| Q[memoryQuotaController] + S --> D[dynamic stream] + D --> H[regionEventHandler] + H --> C[downstream consumer] + H -->|ReleaseEvent| Q + + R[Region request scheduler] --> A[regionAdmissionController] + A -->|AcquireScan| Q + A --> W[Region request worker] + W -->|initial scan finishes or aborts| A + A -->|ReleaseScan| Q +``` + +The controller tracks two values: + +| Value | Meaning | Accounting model | +| --- | --- | --- | +| `used` | Bytes retained by received entry events. | Estimated from the actual event batch after it arrives. | +| `scanUsed` | Predicted bytes for admitted, unfinished initial scans. | Reserved before a scan starts and released when it finishes or aborts. | + +The combined pressure is: + +```text +pressure = max(used, scanUsed) +``` + +The values are deliberately not added. `scanUsed` predicts event memory that +an initial scan may produce, while `used` measures that memory after events +arrive. Adding them would increasingly count the same pressure twice while a +scan is producing events. + +## 4. Configuration and thresholds + +The settings are under the server debug puller configuration: + +```toml +[debug.puller] +memory-quota = 1073741824 +scan-base-size = 8388608 +``` + +| Setting | Default | Meaning | +| --- | ---: | --- | +| `memory-quota` | 1 GiB | Local soft capacity, denoted by `Q`. | +| `scan-base-size` | 8 MiB | Base estimate for one admitted initial scan, denoted by `B`. | + +Zero values are replaced by these defaults during configuration validation. +The following thresholds are derived internally: + +| Threshold | Value | Purpose | +| --- | ---: | --- | +| Pause low-priority scans | `ceil(0.15 * Q)` | Enter scan throttling early. | +| Resume low-priority scans | `floor(0.05 * Q)` | Resume with hysteresis. | +| Event receive hard limit | `2 * Q` | Block additional entry events. | +| Maximum scan estimate | `16 * B` | Bound one scan's predicted charge. | + +With the defaults, low-priority scan admission pauses around 153.6 MiB, +resumes around 51.2 MiB, and event receiving blocks around 2 GiB. + +`memory-quota` is called a soft capacity because high-priority scans may pass +the scan gate and already-owned event memory is not discarded. The receive +hard limit is a separate safety threshold rather than the value exported as +the configured quota. + +## 5. Event memory accounting + +### 5.1 What is accounted + +Only Region events containing TiKV entries acquire event memory. Resolved-ts +events and Region error notifications do not. + +For an entry event, `regionEvent.getSize()` estimates: + +- The `regionEvent` value. +- Entry wrapper structures. +- Every row structure. +- Key, value, and old-value byte slices. +- The slice of Region state pointers. + +This is an accounting estimate, not a heap profiler. It intentionally does not +include every allocator, runtime, dynamic-stream, or downstream data-structure +overhead. + +### 5.2 Acquisition and the hard limit + +`regionEventSink.Push()` calls `AcquireEvent()` before pushing an entry event +into the dynamic stream. Below the hard limit, acquisition uses an atomic +compare-and-swap loop and takes no mutex. + +For current usage `U`, batch size `E`, and hard limit `H = 2Q`: + +```text +U == 0 -> admit E +U > 0 and U + E <= H -> admit E +U > 0 and U + E > H -> wait +``` + +The empty-controller exception allows one oversized event batch to make +progress even when that batch alone is larger than the hard limit. Without it, +such a batch could never be admitted. Consequently, the hard limit is a +backpressure point, not an absolute maximum. + +If admission cannot proceed, the receiver waits until one of these conditions +is true: + +- Event memory is released and the acquisition retry succeeds. +- The subscription is stopped. +- The subscription client context is canceled. + +While waiting, `AcquireEvent()` returns `false` when it observes either of the +latter two cases, and the event is not pushed into the dynamic stream. The +uncontended fast path checks context cancellation but does not inspect the span +state; normal Region deregistration prevents stopped spans from continuing to +produce events. + +### 5.3 Ownership and release + +A successful event acquisition remains owned until one of these terminal +paths releases it: + +| Path | Release point | +| --- | --- | +| Dynamic stream drops the event | `regionEventHandler.OnDrop()` | +| Downstream consumes asynchronously | The downstream wake callback, after the KV event cache is cleared and resolved-ts is advanced | +| Downstream does not retain the batch | Immediately after synchronous handling | +| Entry event produces no retained KV events | At the end of handler processing | + +When several dynamic-stream events are handled as a batch, their +`memoryBytes` values are summed and released together. Stopping a subscription +does not erase already-owned memory; the normal drop, callback, or handler path +still performs the release. + +### 5.4 Event waiter notification + +Event receivers use a close-and-replace channel protocol: + +1. A receiver registers itself as a waiter. +2. It loads the current `ready` channel under the notifier mutex. +3. It retries acquisition and checks whether the span has stopped. +4. Only then does it block on `ready` or context cancellation. + +`ReleaseEvent()` closes the current channel and installs a new one for future +waiters. The retry after registration prevents a release immediately before or +during registration from becoming a lost wake-up. An atomic waiter count is +only a fast-path hint used to avoid unnecessary broadcasts. + +## 6. Initial-scan admission + +### 6.1 Scan estimate + +An initial scan reserves estimated memory before the Region request is sent. +For base size `B` and resolved-ts lag `L`, the estimate is: + +```text +lagFactor = min(16, 1 + 0.22 * log2(1 + L / 10 minutes)) +estimate = clamp(B * lagFactor, B, 16 * B) +``` + +The logarithmic factor gives older scans a larger charge without allowing one +Region to monopolize the entire quota. A scan with no positive lag is charged +`B`; the maximum charge is `16B`. + +The estimate is predictive. It is not adjusted to match the exact bytes later +received from that Region. + +### 6.2 Priority and admission states + +The scan priority policy marks a Region HIGH when any of these conditions is +true: + +- The request inherited HIGH priority from an earlier attempt. +- The Region resolved-ts is within the configured old-start-ts lag threshold. +- The subscribed span has caught up once; this state is sticky across later + retries. + +Other scans are LOW priority. Priority is also sent to TiKV/CSE and controls +the local Region request queue and request-worker window. + +The memory quota controller has two admission states: + +| State | LOW priority | HIGH priority | +| --- | --- | --- | +| `normal` | Admitted | Admitted | +| `pauseLowPriority` | Waits on `scanReady` | Admitted | + +The transition rules use `pressure = max(used, scanUsed)`: + +```mermaid +stateDiagram-v2 + [*] --> normal + normal --> pauseLowPriority: pressure >= 15% of Q + pauseLowPriority --> normal: pressure <= 5% of Q +``` + +Hysteresis prevents scans from repeatedly stopping and starting around one +threshold. HIGH priority scans are an escape path: they remain eligible while +LOW priority backlog is paused, subject to the request worker's maximum +window. + +Admission is decided using pressure before adding the new scan estimate. This +allows one LOW priority scan to cross the pause threshold and make progress; +subsequent LOW priority scans wait. HIGH priority scans can continue increasing +`scanUsed` beyond the soft threshold. + +### 6.3 Interaction with the Region request window + +Memory admission is applied after the per-worker Region request window check. +Both conditions must allow a scan: + +1. The worker must have an available ordinary or maximum-window slot. +2. The global memory quota must admit the scan. + +LOW priority requests use the ordinary window. HIGH priority requests can use +the larger window configured by `region-request-max-window-multiplier` and also +bypass `pauseLowPriority`. These two controls serve different purposes: the +window bounds per-worker concurrency, while the quota coordinates memory +pressure across all workers. + +### 6.4 Scan lease lifecycle + +Successful admission returns a byte charge stored in a `regionReq` lease. The +lease is released when: + +- The Region emits its initialization completion event. +- The request is canceled because the subscription stopped. +- The store stream fails or exits. +- Another request cleanup path aborts the scan. + +`finish()` and `abort()` share an atomic compare-and-swap, so concurrent cleanup +paths release the scan estimate and request-window slot exactly once. + +If a queued task belongs to an already-stopped span, `AcquireScan()` admits it +with a zero-byte lease. This lets the task reach the normal stopped-request +cleanup path without consuming quota or remaining blocked forever. + +### 6.5 Scan waiter notification + +Rejected scans wait on the current `scanReady` channel. The controller closes +and replaces this channel when a transition can make scans eligible: + +- Event usage falls far enough to change `pauseLowPriority` to `normal`. +- Releasing a scan estimate changes the state to `normal`. +- Subscription stop or client shutdown explicitly calls `WakeAll()`. + +The waiting admission loop always rechecks the worker window, span state, and +memory state after waking. The channel is a broadcast signal, not a reservation +for a particular worker. + +## 7. Shutdown and cancellation + +Stopping a subscription first marks its span as stopped and then calls +`WakeAll()`: + +- Blocked event receivers wake and recheck acquisition, cancellation, and the + stopped span state. +- Blocked scan admissions wake and receive a zero-byte lease for cleanup. + +Closing the event sink also calls `WakeAll()` so receivers can observe context +cancellation. Wake-up does not release memory on behalf of an owner. Existing +event and scan charges remain until their corresponding drop, callback, +finish, or abort path runs. + +This separation is important: notification changes scheduling, while release +changes accounting. + +## 8. Concurrency model and invariants + +The implementation separates synchronization by access frequency: + +| State | Synchronization | +| --- | --- | +| Event `used` | `atomic.Uint64` and compare-and-swap | +| Event waiter count | Atomic counter | +| Event ready channel | `eventMemoryNotifier.mu` | +| Scan usage, admission level, and ready channel | `scanMu` | +| Scan waiter count | Atomic counter | +| Per-worker pending queue and inflight window | `regionAdmissionController.state` mutex | + +The main invariants are: + +1. Every successful nonzero acquisition has one terminal release. +2. `used` changes only through `AcquireEvent()` and `ReleaseEvent()`. +3. `scanUsed` changes only while holding `scanMu`. +4. A `regionReq` releases its scan charge and worker slot at most once. +5. Waiters recheck their predicate after registration and after every wake-up. +6. Notifications never transfer ownership and do not imply successful + admission. +7. Stopping a span wakes blocked work but does not invalidate ownership already + handed to downstream code. + +## 9. Observability + +The subscription client updates quota metrics every ten seconds. + +| Metric | Meaning | +| --- | --- | +| `ticdc_log_puller_memory_quota{type="max"}` | Configured soft capacity `Q`. | +| `ticdc_log_puller_memory_quota{type="used"}` | Accounted retained event bytes. | +| `ticdc_log_puller_memory_quota{type="scan_estimated"}` | Estimated bytes reserved by active initial scans. | +| `ticdc_log_puller_memory_quota_event_waiter_count` | Event receivers currently waiting at the hard limit. | +| `ticdc_log_puller_memory_quota_scan_waiter_count` | Region scans currently waiting at the scan gate. | +| `ticdc_log_puller_memory_quota_event_wait_duration` | Event receive wait duration histogram. | +| `ticdc_log_puller_memory_quota_scan_wait_duration` | Scan admission wait duration histogram. | + +The Grafana dashboards expose three panels: + +- **Memory Quota** for logpuller quota values from + `ticdc_log_puller_memory_quota`. For compatibility, the panel also reads the + legacy log-puller series from `ticdc_dynamic_stream_memory_usage`. +- **Memory Quota Waiters** for current event and scan waiters. +- **Memory Quota Wait Duration** for average and P99 wait latency. + +Operationally: + +- Rising `scan_estimated` followed by scan waiters means LOW priority initial scans + are being intentionally paced. +- Rising `used` with event waiters means downstream retention has reached the + receive hard limit. +- Persistent HIGH `scan_estimated` without scan waiters can be expected when active + requests are HIGH priority, because they bypass the soft scan gate. + +## 10. Limitations and trade-offs + +### 10.1 Approximate rather than exact accounting + +Event size is estimated from selected Go structures and payload bytes, and +scan memory is predicted from lag. The metrics should be interpreted as Log +Puller quota state, not exact heap usage. + +### 10.2 Progress over a strict cap + +One oversized event can enter an empty controller, and HIGH priority scans can +continue above the soft quota. These exceptions avoid deadlock and protect +recovery progress, at the cost of allowing temporary overshoot. + +### 10.3 Global rather than per-subscription fairness + +All subscriptions using the client share the controller. The design protects +the Log Puller as a whole but does not reserve memory for a particular +subscription or prevent one active workload from consuming most of the +accounted memory. + +### 10.4 Predictive and actual pressure overlap + +Using `max(used, scanUsed)` avoids double counting, but it is intentionally +conservative in only one dimension at a time. If scan estimates and retained +events represent unrelated workloads, their combined real memory can be +higher than the reported pressure. The separate event hard limit remains the +last receive-path backpressure point. + +### 10.5 No forced reclamation + +The controller blocks new work and waits for current owners to release memory. +It does not discard downstream-owned events or revoke active scan leases. This +keeps ownership and correctness simple, but recovery depends on the downstream +callback and request cleanup paths continuing to run. diff --git a/logservice/logpuller/memory_quota.go b/logservice/logpuller/memory_quota.go new file mode 100644 index 0000000000..f2228f2ad8 --- /dev/null +++ b/logservice/logpuller/memory_quota.go @@ -0,0 +1,396 @@ +// 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "math" + "sync" + "sync/atomic" + "time" + + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/tikv/client-go/v2/oracle" +) + +const ( + // defaultPauseLowPriorityRatio pauses new low-priority scans when memory + // pressure reaches 15% of the soft capacity. + defaultPauseLowPriorityRatio = 0.15 + + // defaultResumeLowPriorityRatio resumes low-priority scans after memory + // pressure falls to 5% of the soft capacity. + defaultResumeLowPriorityRatio = 0.05 + + // defaultHardLimitRatio blocks receiving more events when accounted event + // memory reaches twice the soft capacity. + defaultHardLimitRatio = 2.0 + + // defaultScanLagUnit is the lag unit used by the logarithmic scan estimate. + defaultScanLagUnit = 10 * time.Minute + + // defaultScanLagWeight controls how quickly the scan estimate grows with lag. + defaultScanLagWeight = 0.22 + + // defaultMaxScanLagFactor caps one scan estimate at this multiple of the base. + defaultMaxScanLagFactor = 16 +) + +type admissionLevel uint8 + +const ( + admissionNormal admissionLevel = iota + admissionPauseLowPriority +) + +// eventMemoryNotifier wakes event receivers that are waiting for memory. Each +// notification closes the current ready channel to wake all current waiters, +// then creates a new channel for future waiters. +// +// To avoid missing a notification, a receiver waits in this order: +// +// 1. Register the waiter. +// 2. Read the current ready channel under mu. +// 3. Recheck memory and the span state before blocking on that channel. +// +// If a notification happens just before registration, the final recheck sees +// the released memory or stopped span, so the receiver does not block. +type eventMemoryNotifier struct { + mu sync.Mutex + ready chan struct{} + waiters atomic.Int64 +} + +func newEventMemoryNotifier() *eventMemoryNotifier { + return &eventMemoryNotifier{ready: make(chan struct{})} +} + +func (n *eventMemoryNotifier) wait( + ctx context.Context, + span *subscribedSpan, + tryAcquire func() bool, +) bool { + n.waiters.Add(1) + defer n.waiters.Add(-1) + for { + n.mu.Lock() + ready := n.ready + n.mu.Unlock() + + // This check must stay after waiter registration and loading ready. It + // closes both windows in which notify could otherwise be lost. + if tryAcquire() { + return true + } + if span.stopped.Load() { + return false + } + select { + case <-ready: + case <-ctx.Done(): + return false + } + } +} + +func (n *eventMemoryNotifier) notify() { + // waiters is only a fast-path hint. A waiter that registers after this load + // rechecks memory and the span state before blocking, so observing a stale + // zero cannot lose a wakeup. Observing a stale nonzero only causes a harmless + // extra broadcast. + if n.waiters.Load() == 0 { + return + } + n.mu.Lock() + close(n.ready) + n.ready = make(chan struct{}) + n.mu.Unlock() +} + +// memoryQuotaController coordinates memory pressure from two sources: +// retained event memory and admitted initial scans. +// +// Event memory tracks bytes kept alive until downstream finishes consuming +// them. It may exceed the soft capacity temporarily, but the receive path +// blocks once it reaches the hard limit. +// +// Initial scans are charged by estimate instead of measured bytes. Each +// admitted scan starts from scanBaseSize, grows logarithmically with scan lag, +// and is capped at maxScanLagFactor times the base size. Scan admission +// compares max(event used, scan used) with the soft capacity: low-priority +// scans pause at pauseLowPriorityLimit and resume at +// resumeLowPriorityLimit, while high-priority scans continue to make progress. +type memoryQuotaController struct { + capacity uint64 + // used tracks event bytes retained until downstream finishes consuming them. + // Event accounting is on the receive hot path, so acquiring and releasing + // memory only use atomic operations while usage is below the hard limit. + used atomic.Uint64 + + // eventNotifier owns the wait protocol used after the hard limit is reached. + eventNotifier *eventMemoryNotifier + scanWaiters atomic.Int64 + + // scanMu guards scan admission state and scanReady. Scan admission happens + // once per region rather than once per event batch, so it is intentionally + // kept simple instead of adding atomics to every field. + scanMu sync.Mutex + // scanUsed tracks the estimated memory of all admitted initial scans. + scanUsed uint64 + level admissionLevel + // scanReady is replaced and closed when a memory transition can make a + // rejected scan eligible. Workers wait on this channel directly, avoiding a + // synchronous broadcast to every store and request worker. + scanReady chan struct{} + + pauseLowPriorityLimit uint64 + resumeLowPriorityLimit uint64 + hardLimit uint64 + + scanEstimate uint64 +} + +func newMemoryQuotaController(capacity, scanBaseSize uint64) *memoryQuotaController { + hardLimit := uint64(math.MaxUint64) + if capacity <= math.MaxUint64/uint64(defaultHardLimitRatio) { + hardLimit = capacity * uint64(defaultHardLimitRatio) + } + c := &memoryQuotaController{ + capacity: capacity, + level: admissionNormal, + pauseLowPriorityLimit: uint64(math.Ceil(float64(capacity) * defaultPauseLowPriorityRatio)), + resumeLowPriorityLimit: uint64(float64(capacity) * defaultResumeLowPriorityRatio), + hardLimit: hardLimit, + scanEstimate: scanBaseSize, + eventNotifier: newEventMemoryNotifier(), + scanReady: make(chan struct{}), + } + return c +} + +// WakeAll wakes quota waiters so they can observe cancellation or a stopped span. +func (c *memoryQuotaController) WakeAll() { + c.eventNotifier.notify() + c.NotifyScanAdmission() +} + +// AcquireScan admits one region scan and returns its memory estimate. +func (c *memoryQuotaController) AcquireScan( + region regionInfo, + currentTs uint64, +) (bytes uint64, retry <-chan struct{}, admitted bool) { + span := region.subscribedSpan + if span.stopped.Load() { + // Let stale tasks reach the worker's stopped-subscription cleanup path + // without consuming scan quota. + return 0, nil, true + } + + c.scanMu.Lock() + defer c.scanMu.Unlock() + c.refreshLevelLocked() + lowPriority := isLowPriorityScan(region, currentTs) + // Admission is based on the pressure before accounting this scan. This lets + // one scan make progress even when its estimate alone exceeds the threshold. + if lowPriority && c.level == admissionPauseLowPriority { + return 0, c.scanReady, false + } + bytes = c.estimateScanSizeLocked(region, currentTs) + c.scanUsed += bytes + c.refreshLevelLocked() + return bytes, nil, true +} + +// ReleaseScan releases the estimate owned by an admitted region scan. +func (c *memoryQuotaController) ReleaseScan(bytes uint64) { + if bytes == 0 { + return + } + c.scanMu.Lock() + previousLevel := c.level + c.scanUsed = subtractFloor(c.scanUsed, bytes) + c.refreshLevelLocked() + if c.level < previousLevel { + c.notifyScanAdmissionLocked() + } + c.scanMu.Unlock() +} + +// AcquireEvent accounts one event batch. Below the hard limit its hot path is +// a context check and an atomic compare-and-swap; it does not allocate or take +// a mutex. +func (c *memoryQuotaController) AcquireEvent( + ctx context.Context, + span *subscribedSpan, + bytes uint64, +) bool { + if ctx.Err() != nil { + return false + } + if c.tryAcquireEvent(bytes) { + return true + } + + start := time.Now() + acquired := c.eventNotifier.wait(ctx, span, func() bool { + return c.tryAcquireEvent(bytes) + }) + metrics.LogPullerMemoryQuotaEventWaitDuration.Observe(time.Since(start).Seconds()) + return acquired +} + +func (c *memoryQuotaController) tryAcquireEvent(bytes uint64) bool { + for { + used := c.used.Load() + if used > 0 && wouldExceed(used, bytes, c.hardLimit) { + return false + } + if bytes > math.MaxUint64-used { + return false + } + if c.used.CompareAndSwap(used, used+bytes) { + return true + } + } +} + +// ReleaseEvent releases event memory after downstream has consumed the event. +func (c *memoryQuotaController) ReleaseEvent(bytes uint64) { + if bytes == 0 { + return + } + var previousUsed, used uint64 + for { + previousUsed = c.used.Load() + used = subtractFloor(previousUsed, bytes) + if c.used.CompareAndSwap(previousUsed, used) { + break + } + } + if crossesDown(previousUsed, used, c.resumeLowPriorityLimit) { + c.refreshAdmissionAndNotify() + } + c.eventNotifier.notify() +} + +// NotifyScanAdmission wakes workers so they can recheck span state and admission. +func (c *memoryQuotaController) NotifyScanAdmission() { + c.scanMu.Lock() + c.notifyScanAdmissionLocked() + c.scanMu.Unlock() +} + +// UpdateMetrics reports the current event-memory and scan-admission state. +func (c *memoryQuotaController) UpdateMetrics() { + c.scanMu.Lock() + used := c.used.Load() + scanUsed := c.scanUsed + c.scanMu.Unlock() + + metrics.LogPullerMemoryQuota.WithLabelValues("max").Set(float64(c.capacity)) + metrics.LogPullerMemoryQuota.WithLabelValues("used").Set(float64(used)) + metrics.LogPullerMemoryQuota.WithLabelValues("scan_estimated").Set(float64(scanUsed)) + metrics.LogPullerMemoryQuotaEventWaiterCount.Set( + float64(c.eventNotifier.waiters.Load())) + metrics.LogPullerMemoryQuotaScanWaiterCount.Set( + float64(c.scanWaiters.Load())) +} + +func (c *memoryQuotaController) notifyScanAdmissionLocked() { + close(c.scanReady) + c.scanReady = make(chan struct{}) +} + +func (c *memoryQuotaController) refreshAdmissionAndNotify() { + c.scanMu.Lock() + previousLevel := c.level + c.refreshLevelLocked() + if c.level < previousLevel { + c.notifyScanAdmissionLocked() + } + c.scanMu.Unlock() +} + +func (c *memoryQuotaController) estimateScanSizeLocked(region regionInfo, currentTs uint64) uint64 { + raw := float64(c.scanEstimate) * scanLagFactor(region.resolvedTs(), currentTs) + estimate := uint64(math.MaxUint64) + if raw < float64(math.MaxUint64) { + estimate = max(uint64(raw), c.scanEstimate) + } + maxEstimate := uint64(math.MaxUint64) + if c.scanEstimate <= math.MaxUint64/defaultMaxScanLagFactor { + maxEstimate = c.scanEstimate * defaultMaxScanLagFactor + } + if estimate > maxEstimate { + estimate = maxEstimate + } + if estimate == 0 { + estimate = c.scanEstimate + } + return estimate +} + +func scanLagFactor(startTs, currentTs uint64) float64 { + lag := regionScanLag(currentTs, startTs) + if lag <= 0 { + return 1 + } + return min(defaultMaxScanLagFactor, + 1+defaultScanLagWeight*math.Log2(1+float64(lag)/float64(defaultScanLagUnit))) +} + +func regionScanLag(currentTs, checkpointTs uint64) time.Duration { + currentTime := oracle.GetTimeFromTS(currentTs) + checkpointTime := oracle.GetTimeFromTS(checkpointTs) + if !currentTime.After(checkpointTime) { + return 0 + } + return currentTime.Sub(checkpointTime) +} + +func isLowPriorityScan(region regionInfo, _ uint64) bool { + return !isHighScanPriority(region.scanPriority) +} + +func (c *memoryQuotaController) refreshLevelLocked() { + // scanUsed predicts the event memory an initial scan may produce, so adding + // it to actual event bytes would count the same pressure twice. + pressure := max(c.used.Load(), c.scanUsed) + switch c.level { + case admissionPauseLowPriority: + if pressure <= c.resumeLowPriorityLimit { + c.level = admissionNormal + } + default: + if pressure >= c.pauseLowPriorityLimit { + c.level = admissionPauseLowPriority + } + } +} + +func wouldExceed(used, bytes, limit uint64) bool { + return bytes > limit || used > limit-bytes +} + +func crossesDown(previous, current, threshold uint64) bool { + return previous > threshold && current <= threshold +} + +func subtractFloor(value, delta uint64) uint64 { + if value < delta { + return 0 + } + return value - delta +} diff --git a/logservice/logpuller/memory_quota_test.go b/logservice/logpuller/memory_quota_test.go new file mode 100644 index 0000000000..8ec1eab982 --- /dev/null +++ b/logservice/logpuller/memory_quota_test.go @@ -0,0 +1,478 @@ +// 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package logpuller + +import ( + "context" + "math" + "testing" + "time" + + "github.com/pingcap/kvproto/pkg/cdcpb" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/pdutil" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/oracle" +) + +type memoryQuotaTestState struct { + used uint64 + scanUsed uint64 + level admissionLevel +} + +func getMemoryQuotaTestState(quota *memoryQuotaController) memoryQuotaTestState { + quota.scanMu.Lock() + defer quota.scanMu.Unlock() + return memoryQuotaTestState{ + used: quota.used.Load(), + scanUsed: quota.scanUsed, + level: quota.level, + } +} + +func newTestQuotaSpan(subID SubscriptionID) *subscribedSpan { + span := &subscribedSpan{subID: subID} + span.resolvedTs.Store(oracle.GoTimeToTS(time.Now())) + return span +} + +func newTestQuotaRegion(span *subscribedSpan) regionInfo { + state := ®ionlock.LockedRangeState{} + state.ResolvedTs.Store(span.resolvedTs.Load()) + return regionInfo{ + subscribedSpan: span, + lockedRangeState: state, + } +} + +func newTestQuotaRegionWithPriority( + span *subscribedSpan, + priority cdcpb.ScanPriority, +) regionInfo { + region := newTestQuotaRegion(span) + region.scanPriority = priority + return region +} + +func setTestQuotaSpanLag(span *subscribedSpan, lag time.Duration) uint64 { + now := time.Now() + span.resolvedTs.Store(oracle.GoTimeToTS(now.Add(-lag))) + return oracle.GoTimeToTS(now) +} + +func TestMemoryQuotaUpdateMetrics(t *testing.T) { + quota := newMemoryQuotaController(66, 8) + span := newTestQuotaSpan(1) + require.True(t, quota.AcquireEvent(context.Background(), span, 55)) + t.Cleanup(func() { quota.ReleaseEvent(55) }) + + quota.scanMu.Lock() + quota.scanUsed = 7 + quota.scanMu.Unlock() + quota.eventNotifier.waiters.Store(2) + t.Cleanup(func() { quota.eventNotifier.waiters.Store(0) }) + quota.scanWaiters.Store(3) + t.Cleanup(func() { quota.scanWaiters.Store(0) }) + + quota.UpdateMetrics() + + require.Equal(t, float64(66), testutil.ToFloat64( + metrics.LogPullerMemoryQuota.WithLabelValues("max"))) + require.Equal(t, float64(55), testutil.ToFloat64( + metrics.LogPullerMemoryQuota.WithLabelValues("used"))) + require.Equal(t, float64(7), testutil.ToFloat64( + metrics.LogPullerMemoryQuota.WithLabelValues("scan_estimated"))) + require.Equal(t, float64(2), + testutil.ToFloat64(metrics.LogPullerMemoryQuotaEventWaiterCount)) + require.Equal(t, float64(3), + testutil.ToFloat64(metrics.LogPullerMemoryQuotaScanWaiterCount)) +} + +func TestMemoryQuotaAdmissionLevels(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + lowPrioritySpan := newTestQuotaSpan(1) + highPrioritySpan := newTestQuotaSpan(2) + lowPriorityTs := setTestQuotaSpanLag(lowPrioritySpan, time.Hour) + highPriorityTs := setTestQuotaSpanLag(highPrioritySpan, time.Hour) + + require.True(t, quota.AcquireEvent(context.Background(), highPrioritySpan, 5)) + require.True(t, quota.AcquireEvent(context.Background(), highPrioritySpan, 10)) + _, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(lowPrioritySpan, cdcpb.ScanPriority_SCAN_PRIORITY_LOW), + lowPriorityTs, + ) + require.False(t, admitted) + + scanBytes, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(highPrioritySpan, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + highPriorityTs, + ) + require.True(t, admitted) + quota.ReleaseScan(scanBytes) + + require.True(t, quota.AcquireEvent(context.Background(), highPrioritySpan, 45)) + require.True(t, quota.AcquireEvent(context.Background(), highPrioritySpan, 20)) + scanBytes, _, admitted = quota.AcquireScan( + newTestQuotaRegionWithPriority(highPrioritySpan, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + highPriorityTs, + ) + require.True(t, admitted) + quota.ReleaseScan(scanBytes) + + quota.ReleaseEvent(20) + state := getMemoryQuotaTestState(quota) + require.Equal(t, admissionPauseLowPriority, state.level) + quota.ReleaseEvent(45) + state = getMemoryQuotaTestState(quota) + require.Equal(t, admissionPauseLowPriority, state.level) + quota.ReleaseEvent(10) + state = getMemoryQuotaTestState(quota) + require.Equal(t, admissionNormal, state.level) + quota.ReleaseEvent(5) +} + +func TestMemoryQuotaReleaseEventClampsToZero(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, time.Hour) + + require.True(t, quota.AcquireEvent(context.Background(), span, 20)) + _, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_LOW), + currentTs, + ) + require.False(t, admitted) + + quota.ReleaseEvent(30) + state := getMemoryQuotaTestState(quota) + require.Zero(t, state.used) + require.Equal(t, admissionNormal, state.level) + + require.True(t, quota.AcquireEvent(context.Background(), span, 1)) + quota.ReleaseEvent(1) +} + +func TestMemoryQuotaDerivedLimitsSaturate(t *testing.T) { + quota := newMemoryQuotaController(math.MaxUint64, math.MaxUint64/2+1) + require.Equal(t, uint64(math.MaxUint64), quota.hardLimit) + + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, 24*time.Hour) + scanBytes, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + currentTs, + ) + require.True(t, admitted) + require.Equal(t, uint64(math.MaxUint64), scanBytes) + quota.ReleaseScan(scanBytes) +} + +func TestMemoryQuotaSpanStopKeepsOwnedMemoryUntilRelease(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span1 := newTestQuotaSpan(1) + span2 := newTestQuotaSpan(2) + + require.True(t, quota.AcquireEvent(context.Background(), span1, 30)) + require.True(t, quota.AcquireEvent(context.Background(), span2, 40)) + scanBytes, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(span1, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + span1.resolvedTs.Load(), + ) + require.True(t, admitted) + require.NotZero(t, scanBytes) + + span1.stopped.Store(true) + quota.WakeAll() + state := getMemoryQuotaTestState(quota) + require.Equal(t, uint64(70), state.used) + require.Equal(t, scanBytes, state.scanUsed) + + quota.ReleaseEvent(30) + quota.ReleaseScan(scanBytes) + state = getMemoryQuotaTestState(quota) + require.Equal(t, uint64(40), state.used) + + // Late tasks reach the stopped-subscription cleanup path without consuming + // scan quota. + scanBytes, _, admitted = quota.AcquireScan( + newTestQuotaRegionWithPriority(span1, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + span1.resolvedTs.Load(), + ) + require.True(t, admitted) + require.Zero(t, scanBytes) + + quota.ReleaseEvent(40) + state = getMemoryQuotaTestState(quota) + require.Zero(t, state.used) +} + +func TestMemoryQuotaBlockedEventStopsWhenSpanStops(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + quota.hardLimit = 100 + span := newTestQuotaSpan(1) + + require.True(t, quota.AcquireEvent(context.Background(), span, 100)) + acquired := make(chan bool, 1) + go func() { + acquired <- quota.AcquireEvent(context.Background(), span, 1) + }() + + select { + case <-acquired: + t.Fatal("event memory should wait at the hard limit") + case <-time.After(100 * time.Millisecond): + } + + span.stopped.Store(true) + quota.WakeAll() + select { + case ok := <-acquired: + require.False(t, ok) + case <-time.After(time.Second): + t.Fatal("stopping the subscription did not wake the blocked event") + } + quota.ReleaseEvent(100) +} + +func TestMemoryQuotaBlockedEventResumesAfterRelease(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span := newTestQuotaSpan(1) + + require.True(t, quota.AcquireEvent(context.Background(), span, 200)) + acquired := make(chan bool, 1) + go func() { + acquired <- quota.AcquireEvent(context.Background(), span, 1) + }() + + select { + case <-acquired: + t.Fatal("event memory should wait at the hard limit") + case <-time.After(100 * time.Millisecond): + } + + quota.ReleaseEvent(200) + select { + case ok := <-acquired: + require.True(t, ok) + quota.ReleaseEvent(1) + case <-time.After(time.Second): + t.Fatal("event memory did not resume after memory was released") + } +} + +func TestMemoryQuotaBlockedEventStopsOnContextCancellation(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + quota.hardLimit = 100 + span := newTestQuotaSpan(1) + ctx, cancel := context.WithCancel(context.Background()) + + require.True(t, quota.AcquireEvent(context.Background(), span, 100)) + acquired := make(chan bool, 1) + go func() { + acquired <- quota.AcquireEvent(ctx, span, 1) + }() + require.Eventually(t, func() bool { + return quota.eventNotifier.waiters.Load() == 1 + }, time.Second, time.Millisecond) + + // Cancellation must stop the waiter without a memory release or an explicit + // quota notification. + cancel() + select { + case ok := <-acquired: + require.False(t, ok) + case <-time.After(time.Second): + t.Fatal("context cancellation did not stop the blocked event") + } + quota.ReleaseEvent(100) +} + +func TestMemoryQuotaConcurrentWaitersDoNotLoseWakeups(t *testing.T) { + const waiterCount = 32 + quota := newMemoryQuotaController(100, 10) + quota.hardLimit = 1 + span := newTestQuotaSpan(1) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Hold the only available byte until every goroutine is waiting. Releasing + // it wakes all waiters; each successful waiter then releases it for the next. + require.True(t, quota.AcquireEvent(ctx, span, 1)) + results := make(chan bool, waiterCount) + for range waiterCount { + go func() { + acquired := quota.AcquireEvent(ctx, span, 1) + if acquired { + quota.ReleaseEvent(1) + } + results <- acquired + }() + } + require.Eventually(t, func() bool { + return quota.eventNotifier.waiters.Load() == waiterCount + }, time.Second, time.Millisecond) + + quota.ReleaseEvent(1) + for range waiterCount { + select { + case acquired := <-results: + require.True(t, acquired) + case <-ctx.Done(): + t.Fatal("event waiter did not make progress") + } + } + state := getMemoryQuotaTestState(quota) + require.Zero(t, state.used) +} + +func TestMemoryQuotaLowPriorityScanUsesCurrentPressure(t *testing.T) { + quota := newMemoryQuotaController(100, 20) + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, time.Hour) + region := newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_LOW) + + bytes1, _, admitted := quota.AcquireScan(region, currentTs) + require.True(t, admitted) + require.NotZero(t, bytes1) + state := getMemoryQuotaTestState(quota) + require.Greater(t, state.scanUsed, quota.pauseLowPriorityLimit) + + _, _, admitted = quota.AcquireScan(region, currentTs) + require.False(t, admitted) + + quota.ReleaseScan(bytes1) + bytes2, _, admitted := quota.AcquireScan(region, currentTs) + require.True(t, admitted) + quota.ReleaseScan(bytes2) +} + +func TestMemoryQuotaLowLagScanBypassesWarmingGate(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, time.Minute) + + require.True(t, quota.AcquireEvent(context.Background(), span, 20)) + scanBytes, _, admitted := quota.AcquireScan( + newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH), + currentTs, + ) + require.True(t, admitted) + require.NotZero(t, scanBytes) + state := getMemoryQuotaTestState(quota) + require.NotZero(t, state.scanUsed) + + quota.ReleaseScan(scanBytes) + quota.ReleaseEvent(20) +} + +func TestAdmissionWaitsForMemoryAndReleasesScanMemory(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, time.Hour) + clock := pdutil.NewClock4Test().(*pdutil.Clock4Test) + clock.SetTS(currentTs) + controller := newRegionAdmissionController(1, 1, quota, clock) + + require.True(t, quota.AcquireEvent(context.Background(), span, 20)) + region := newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_LOW) + require.True(t, controller.submit(newRegionPriorityTask(region, 1))) + + type popResult struct { + req *regionReq + err error + } + result := make(chan popResult, 1) + go func() { + req, err := controller.pop(context.Background(), nil) + result <- popResult{req: req, err: err} + }() + select { + case <-result: + t.Fatal("low-priority scan should wait while memory is under pressure") + case <-time.After(100 * time.Millisecond): + } + + quota.ReleaseEvent(20) + var resultValue popResult + select { + case resultValue = <-result: + case <-time.After(time.Second): + t.Fatal("scan admission was not notified after memory became available") + } + require.NoError(t, resultValue.err) + req := resultValue.req + state := getMemoryQuotaTestState(quota) + require.NotZero(t, state.scanUsed) + require.True(t, req.abort()) + state = getMemoryQuotaTestState(quota) + require.Zero(t, state.scanUsed) +} + +func TestAdmissionWakesWhenBlockedSpanStops(t *testing.T) { + quota := newMemoryQuotaController(100, 10) + span := newTestQuotaSpan(1) + currentTs := setTestQuotaSpanLag(span, time.Hour) + clock := pdutil.NewClock4Test().(*pdutil.Clock4Test) + clock.SetTS(currentTs) + controller := newRegionAdmissionController(1, 1, quota, clock) + + require.True(t, quota.AcquireEvent(context.Background(), span, 20)) + require.True(t, controller.submit(newRegionPriorityTask( + newTestQuotaRegionWithPriority(span, cdcpb.ScanPriority_SCAN_PRIORITY_LOW), 1))) + + type popResult struct { + req *regionReq + err error + } + result := make(chan popResult, 1) + go func() { + req, err := controller.pop(context.Background(), nil) + result <- popResult{req: req, err: err} + }() + select { + case <-result: + t.Fatal("low-priority scan should wait while memory is under pressure") + case <-time.After(100 * time.Millisecond): + } + + span.stopped.Store(true) + quota.WakeAll() + select { + case result := <-result: + require.NoError(t, result.err) + require.Zero(t, result.req.scanBytes) + require.True(t, result.req.abort()) + case <-time.After(time.Second): + t.Fatal("stopping the span did not wake scan admission") + } + quota.ReleaseEvent(20) +} + +func BenchmarkMemoryQuotaEventAccounting(b *testing.B) { + quota := newMemoryQuotaController(1024*1024*1024, 8*1024*1024) + span := newTestQuotaSpan(1) + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if !quota.AcquireEvent(ctx, span, 1) { + b.Fatal("failed to acquire event memory") + } + quota.ReleaseEvent(1) + } +} diff --git a/logservice/logpuller/region_admission_controller.go b/logservice/logpuller/region_admission_controller.go index 1564aaed74..bd4537b89f 100644 --- a/logservice/logpuller/region_admission_controller.go +++ b/logservice/logpuller/region_admission_controller.go @@ -22,6 +22,7 @@ import ( "github.com/pingcap/log" "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/pdutil" "github.com/pingcap/ticdc/utils/heap" "go.uber.org/zap" ) @@ -36,6 +37,7 @@ type regionReq struct { regionInfo regionInfo createTime time.Time controller *regionAdmissionController + scanBytes uint64 released atomic.Bool } @@ -72,7 +74,7 @@ func (r *regionReq) release() bool { if !r.released.CompareAndSwap(false, true) { return false } - r.controller.release() + r.controller.release(r.scanBytes) return true } @@ -96,6 +98,11 @@ type regionAdmissionController struct { // closed prevents new submissions and makes waiting workers exit. closed bool } + + // memoryQuota gates initial scans using the log puller's global memory + // pressure. pdClock provides the current TS used for scan estimation. + memoryQuota *memoryQuotaController + pdClock pdutil.Clock // notify wakes workers when a request is submitted or an admission slot is // released. The one-element buffer prevents a wakeup from being lost between // checking the admission condition and waiting on this channel. Notifications @@ -109,7 +116,12 @@ type regionAdmissionStats struct { inflight int } -func newRegionAdmissionController(currentWindow, maxWindowMultiplier int) *regionAdmissionController { +func newRegionAdmissionController( + currentWindow int, + maxWindowMultiplier int, + memoryQuota *memoryQuotaController, + pdClock pdutil.Clock, +) *regionAdmissionController { if currentWindow <= 0 { currentWindow = 1 } @@ -123,6 +135,8 @@ func newRegionAdmissionController(currentWindow, maxWindowMultiplier int) *regio controller := ®ionAdmissionController{ currentWindow: currentWindow, maxWindow: maxWindow, + memoryQuota: memoryQuota, + pdClock: pdClock, notify: make(chan struct{}, 1), } controller.state.pending = heap.NewHeap[*regionPriorityTask]() @@ -141,8 +155,8 @@ func (c *regionAdmissionController) submit(task *regionPriorityTask) bool { return true } -// pop waits for an eligible request. If interrupt is signaled first, it returns -// nil without an error so the worker can handle its control queue. +// pop waits for an eligible request. If interrupt is signaled first, it +// returns nil without an error so the worker can handle the interrupt source. func (c *regionAdmissionController) pop( ctx context.Context, interrupt <-chan struct{}, @@ -153,7 +167,7 @@ func (c *regionAdmissionController) pop( c.state.Unlock() return nil, context.Canceled } - request := c.popEligibleLocked() + request, scanBytes, memoryReady := c.popEligibleLocked() if request != nil { c.state.inflight++ c.state.Unlock() @@ -161,30 +175,62 @@ func (c *regionAdmissionController) pop( regionInfo: request.regionInfo, createTime: time.Now(), controller: c, + scanBytes: scanBytes, }, nil } c.state.Unlock() + waitingForMemory := memoryReady != nil + if waitingForMemory { + c.memoryQuota.scanWaiters.Add(1) + } + waitStart := time.Time{} + if waitingForMemory { + waitStart = time.Now() + } select { case <-c.notify: + case <-memoryReady: case <-interrupt: + if waitingForMemory { + c.memoryQuota.scanWaiters.Add(-1) + metrics.LogPullerMemoryQuotaScanWaitDuration.Observe(time.Since(waitStart).Seconds()) + } return nil, nil case <-ctx.Done(): + if waitingForMemory { + c.memoryQuota.scanWaiters.Add(-1) + metrics.LogPullerMemoryQuotaScanWaitDuration.Observe(time.Since(waitStart).Seconds()) + } return nil, ctx.Err() } + if waitingForMemory { + c.memoryQuota.scanWaiters.Add(-1) + metrics.LogPullerMemoryQuotaScanWaitDuration.Observe(time.Since(waitStart).Seconds()) + } } } -func (c *regionAdmissionController) popEligibleLocked() *regionPriorityTask { +func (c *regionAdmissionController) popEligibleLocked() ( + *regionPriorityTask, + uint64, + <-chan struct{}, +) { request, ok := c.state.pending.PeekTop() if !ok { - return nil + return nil, 0, nil } if c.state.inflight >= c.windowFor(request) { - return nil + return nil, 0, nil + } + + scanBytes, memoryReady, admitted := c.memoryQuota.AcquireScan( + request.regionInfo, c.pdClock.CurrentTS()) + if !admitted { + return nil, 0, memoryReady } request, _ = c.state.pending.PopTop() - return request + return request, scanBytes, nil } func (c *regionAdmissionController) windowFor(request *regionPriorityTask) int { @@ -194,7 +240,8 @@ func (c *regionAdmissionController) windowFor(request *regionPriorityTask) int { return c.currentWindow } -func (c *regionAdmissionController) release() { +func (c *regionAdmissionController) release(scanBytes uint64) { + c.memoryQuota.ReleaseScan(scanBytes) c.state.Lock() if c.state.inflight > 0 { c.state.inflight-- diff --git a/logservice/logpuller/region_admission_controller_test.go b/logservice/logpuller/region_admission_controller_test.go index 13fc2ed2bc..acab949286 100644 --- a/logservice/logpuller/region_admission_controller_test.go +++ b/logservice/logpuller/region_admission_controller_test.go @@ -23,6 +23,7 @@ import ( "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/ticdc/heartbeatpb" "github.com/pingcap/ticdc/logservice/logpuller/regionlock" + "github.com/pingcap/ticdc/pkg/pdutil" "github.com/stretchr/testify/require" "github.com/tikv/client-go/v2/oracle" "github.com/tikv/client-go/v2/tikv" @@ -60,8 +61,21 @@ func submitRegionForAdmission( require.True(t, controller.submit(task)) } +func newTestRegionAdmissionController( + currentWindow int, + maxWindowMultiplier int, +) *regionAdmissionController { + clock := pdutil.NewClock4Test() + return newRegionAdmissionController( + currentWindow, + maxWindowMultiplier, + newMemoryQuotaController(1024*1024*1024, 8*1024*1024), + clock, + ) +} + func TestRegionAdmissionControllerNormalWindow(t *testing.T) { - controller := newRegionAdmissionController(1, 2) + controller := newTestRegionAdmissionController(1, 2) currentTs := oracle.GoTimeToTS(time.Now()) checkpointTs := oracle.GoTimeToTS(time.Now().Add(-time.Hour)) region1 := prepareRegionForAdmission(createTestRegionInfo(1, 1), checkpointTs) @@ -86,7 +100,7 @@ func TestRegionAdmissionControllerNormalWindow(t *testing.T) { } func TestRegionAdmissionControllerHighPriorityUsesMaxWindow(t *testing.T) { - controller := newRegionAdmissionController(1, 2) + controller := newTestRegionAdmissionController(1, 2) currentTs := oracle.GoTimeToTS(time.Now()) slowCheckpointTs := oracle.GoTimeToTS(time.Now().Add(-time.Hour)) @@ -99,7 +113,8 @@ func TestRegionAdmissionControllerHighPriorityUsesMaxWindow(t *testing.T) { submitRegionForAdmission(t, controller, prepareRegionForAdmission(createTestRegionInfo(1, 2), slowCheckpointTs), currentTs) - highPriorityRegion := prepareRegionForAdmission(createTestRegionInfo(1, 3), slowCheckpointTs) + highPriorityRegion := prepareRegionForAdmission( + createTestRegionInfo(1, 3), slowCheckpointTs) highPriorityRegion.scanPriority = cdcpb.ScanPriority_SCAN_PRIORITY_HIGH submitRegionForAdmission(t, controller, highPriorityRegion, currentTs) @@ -122,7 +137,7 @@ func TestRegionAdmissionControllerHighPriorityUsesMaxWindow(t *testing.T) { } func TestRegionAdmissionControllerPrioritizesHighPriorityRegion(t *testing.T) { - controller := newRegionAdmissionController(1, 2) + controller := newTestRegionAdmissionController(1, 2) currentTs := oracle.GoTimeToTS(time.Now()) slowCheckpointTs := oracle.GoTimeToTS(time.Now().Add(-time.Hour)) @@ -137,7 +152,8 @@ func TestRegionAdmissionControllerPrioritizesHighPriorityRegion(t *testing.T) { currentTs) highPriorityRegion := prepareRegionForAdmission(createTestRegionInfo(1, 3), slowCheckpointTs) highPriorityRegion.scanPriority = cdcpb.ScanPriority_SCAN_PRIORITY_HIGH - submitRegionForAdmission(t, controller, highPriorityRegion, currentTs) + submitRegionForAdmission(t, controller, + highPriorityRegion, currentTs) req2, err := controller.pop(t.Context(), nil) require.NoError(t, err) @@ -152,7 +168,7 @@ func TestRegionAdmissionControllerPrioritizesHighPriorityRegion(t *testing.T) { } func TestRegionAdmissionLeaseReleasedOnce(t *testing.T) { - controller := newRegionAdmissionController(1, 1) + controller := newTestRegionAdmissionController(1, 1) currentTs := oracle.GoTimeToTS(time.Now()) region := prepareRegionForAdmission(createTestRegionInfo(1, 1), currentTs) submitRegionForAdmission(t, controller, region, currentTs) @@ -188,7 +204,7 @@ func TestRegionAdmissionLeaseReleasedOnce(t *testing.T) { } func TestRegionAdmissionControllerClose(t *testing.T) { - controller := newRegionAdmissionController(1, 1) + controller := newTestRegionAdmissionController(1, 1) controller.close() region := prepareRegionForAdmission(createTestRegionInfo(1, 1), 1) require.False(t, controller.submit(newRegionPriorityTask(region, 1))) @@ -198,7 +214,7 @@ func TestRegionAdmissionControllerClose(t *testing.T) { } func TestRegionAdmissionControllerDrainPending(t *testing.T) { - controller := newRegionAdmissionController(1, 1) + controller := newTestRegionAdmissionController(1, 1) region1 := prepareRegionForAdmission(createTestRegionInfo(1, 1), 1) region2 := prepareRegionForAdmission(createTestRegionInfo(1, 2), 1) submitRegionForAdmission(t, controller, region1, 1) diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index e13b91cbd0..24ad6f4140 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -55,6 +55,13 @@ type regionEvent struct { entries *cdcpb.Event_Entries_ resolvedTs uint64 + // memoryBytes is released when this event is dropped or when the derived KV + // events no longer need to be retained by the log puller. + memoryBytes uint64 +} + +func (event *regionEvent) needsMemoryAccounting() bool { + return event.entries != nil } func (event *regionEvent) getSize() int { @@ -121,7 +128,9 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) } newResolvedTs := uint64(0) + memoryBytes := uint64(0) for _, event := range events { + memoryBytes += event.memoryBytes if len(event.states) == 1 && event.states[0].isStale() { hasError = true h.handleRegionError(event.states[0]) @@ -147,9 +156,13 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) span.advanceResolvedTs(newResolvedTs) } } + releaseMemoryQuota := func() { + h.eventSink.memoryQuota.ReleaseEvent(memoryBytes) + } 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()) @@ -166,10 +179,12 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) if !await { span.clearKVEventsCache() tryAdvanceResolvedTs() + releaseMemoryQuota() } return await } else { tryAdvanceResolvedTs() + releaseMemoryQuota() } return false } @@ -226,8 +241,18 @@ func (h *regionEventHandler) GetType(event regionEvent) dynstream.EventType { } func (h *regionEventHandler) OnDrop(event regionEvent) interface{} { + h.eventSink.memoryQuota.ReleaseEvent(event.memoryBytes) // TODO: Distinguish between drop events caused by "path not found" errors and memory control. - state := event.mustFirstState() + if len(event.states) == 0 || event.states[0] == nil { + log.Error("drop invalid region event", + zap.Bool("hasEntries", event.entries != nil), + zap.Uint64("resolvedTs", event.resolvedTs), + zap.Int("states", len(event.states)), + zap.Uint64("memoryBytes", event.memoryBytes)) + return nil + } + + state := event.states[0] fields := []zap.Field{ zap.Bool("hasEntries", event.entries != nil), zap.Uint64("resolvedTs", event.resolvedTs), diff --git a/logservice/logpuller/region_event_handler_test.go b/logservice/logpuller/region_event_handler_test.go index fa6e76eaad..b4cf4237dc 100644 --- a/logservice/logpuller/region_event_handler_test.go +++ b/logservice/logpuller/region_event_handler_test.go @@ -49,7 +49,10 @@ import ( func TestHandleEventEntryEventOutOfOrder(t *testing.T) { // initialize option := dynstream.NewOption() - ds := dynstream.NewParallelDynamicStream("test", ®ionEventHandler{}, option) + handler := ®ionEventHandler{eventSink: ®ionEventSink{ + memoryQuota: newMemoryQuotaController(1024*1024*1024, 8*1024*1024), + }} + ds := dynstream.NewParallelDynamicStream("test", handler, option) ds.Start() span := heartbeatpb.TableSpan{ @@ -72,7 +75,6 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) { subID: subID, span: span, startTs: 1000, // not used - rangeLock: regionlock.NewRangeLock(uint64(subID), span.StartKey, span.EndKey, 1000), consumeKVEvents: consumeKVEvents, advanceResolvedTs: advanceResolvedTs, advanceInterval: 0, @@ -83,16 +85,13 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) { tracker: newRegionTracker(), } region := newRegionInfo( - tikv.NewRegionVerID(1, 1, 1), + tikv.RegionVerID{}, span, &tikv.RPCContext{}, subSpan, false, ) - lockResult := subSpan.rangeLock.LockRange( - context.Background(), span.StartKey, span.EndKey, 1, 1) - require.Equal(t, regionlock.LockRangeStatusSuccess, lockResult.Status) - region.lockedRangeState = lockResult.LockedRangeState + region.lockedRangeState = ®ionlock.LockedRangeState{} state := newRegionFeedState(region, 1, worker, nil) // Receive prewrite2 with empty value. @@ -210,9 +209,10 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) { func TestHandleResolvedTs(t *testing.T) { // initialize option := dynstream.NewOption() - pdClock := pdutil.NewClock4Test() - pdClock.(*pdutil.Clock4Test).SetTS(10) - ds := dynstream.NewParallelDynamicStream("test", ®ionEventHandler{}, option) + handler := ®ionEventHandler{eventSink: ®ionEventSink{ + memoryQuota: newMemoryQuotaController(1024*1024*1024, 8*1024*1024), + }} + ds := dynstream.NewParallelDynamicStream("test", handler, option) ds.Start() consumeKVEvents := func(events []common.RawKVEntry, _ func()) bool { return false } // not used @@ -226,28 +226,24 @@ func TestHandleResolvedTs(t *testing.T) { tracker: newRegionTracker(), } state1 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(1, 1, 1)}, uint64(subID1), worker, nil) - var subSpan1 *subscribedSpan { span := heartbeatpb.TableSpan{ TableID: 100, StartKey: common.ToComparableKey([]byte{}), // TODO: remove spanz dependency EndKey: common.ToComparableKey(common.UpperBoundKey), } - subSpan1 = &subscribedSpan{ + subSpan := &subscribedSpan{ subID: subID1, span: heartbeatpb.TableSpan{}, rangeLock: regionlock.NewRangeLock(uint64(subID1), span.StartKey, span.EndKey, 1), consumeKVEvents: consumeKVEvents, advanceResolvedTs: advanceResolvedTs, advanceInterval: 0, - priorityPolicy: newScanPriorityPolicy(pdClock, 30*time.Minute), + priorityPolicy: newScanPriorityPolicy(pdutil.NewClock4Test(), 30*time.Minute), } - ds.AddPath(subID1, subSpan1, dynstream.AreaSettings{}) - state1.region.subscribedSpan = subSpan1 - lockResult := subSpan1.rangeLock.LockRange( - context.Background(), span.StartKey, span.EndKey, 1, 1) - require.Equal(t, regionlock.LockRangeStatusSuccess, lockResult.Status) - state1.region.lockedRangeState = lockResult.LockedRangeState + ds.AddPath(subID1, subSpan, dynstream.AreaSettings{}) + state1.region.subscribedSpan = subSpan + state1.region.lockedRangeState = ®ionlock.LockedRangeState{} state1.setInitialized() state1.updateResolvedTs(9) } @@ -267,14 +263,11 @@ func TestHandleResolvedTs(t *testing.T) { consumeKVEvents: consumeKVEvents, advanceResolvedTs: advanceResolvedTs, advanceInterval: 0, - priorityPolicy: newScanPriorityPolicy(pdClock, 30*time.Minute), + priorityPolicy: newScanPriorityPolicy(pdutil.NewClock4Test(), 30*time.Minute), } ds.AddPath(subID2, subSpan, dynstream.AreaSettings{}) state2.region.subscribedSpan = subSpan - lockResult := subSpan.rangeLock.LockRange( - context.Background(), span.StartKey, span.EndKey, 2, 2) - require.Equal(t, regionlock.LockRangeStatusSuccess, lockResult.Status) - state2.region.lockedRangeState = lockResult.LockedRangeState + state2.region.lockedRangeState = ®ionlock.LockedRangeState{} state2.setInitialized() state2.updateResolvedTs(11) } @@ -294,14 +287,11 @@ func TestHandleResolvedTs(t *testing.T) { consumeKVEvents: consumeKVEvents, advanceResolvedTs: advanceResolvedTs, advanceInterval: 0, - priorityPolicy: newScanPriorityPolicy(pdClock, 30*time.Minute), + priorityPolicy: newScanPriorityPolicy(pdutil.NewClock4Test(), 30*time.Minute), } ds.AddPath(subID3, subSpan, dynstream.AreaSettings{}) state3.region.subscribedSpan = subSpan - lockResult := subSpan.rangeLock.LockRange( - context.Background(), span.StartKey, span.EndKey, 3, 3) - require.Equal(t, regionlock.LockRangeStatusSuccess, lockResult.Status) - state3.region.lockedRangeState = lockResult.LockedRangeState + state3.region.lockedRangeState = ®ionlock.LockedRangeState{} state3.updateResolvedTs(8) } @@ -346,7 +336,6 @@ func TestHandleResolvedTs(t *testing.T) { require.Equal(t, uint64(10), state1.getLastResolvedTs()) require.Equal(t, uint64(11), state2.getLastResolvedTs()) require.Equal(t, uint64(8), state3.getLastResolvedTs()) - require.True(t, subSpan1.priorityPolicy.everCaughtUp.Load()) } func TestHandleResolvedTsThrottled(t *testing.T) { @@ -394,63 +383,109 @@ func TestHandleResolvedTsThrottled(t *testing.T) { require.Equal(t, uint64(200), handleResolvedTs(span, state, 300)) } -func TestSpanInitializedAfterAllRangesInitialized(t *testing.T) { - ctx := context.Background() - rangeLock := regionlock.NewRangeLock(1, []byte("a"), []byte("z"), 100) - firstLock := rangeLock.LockRange(ctx, []byte("a"), []byte("m"), 1, 1) - require.Equal(t, regionlock.LockRangeStatusSuccess, firstLock.Status) - secondLock := rangeLock.LockRange(ctx, []byte("m"), []byte("z"), 2, 1) - require.Equal(t, regionlock.LockRangeStatusSuccess, secondLock.Status) - - span := &subscribedSpan{ - subID: SubscriptionID(1), - startTs: 100, - span: heartbeatpb.TableSpan{StartKey: []byte("a"), EndKey: []byte("z")}, - rangeLock: rangeLock, - priorityPolicy: newScanPriorityPolicy(pdutil.NewClock4Test(), 30*time.Minute), +func TestHandleEntriesReleasesMemoryAfterDownstreamCallback(t *testing.T) { + quota := newMemoryQuotaController(1024, 8) + span := newTestQuotaSpan(1) + callbackCh := make(chan func(), 1) + span.consumeKVEvents = func(_ []common.RawKVEntry, callback func()) bool { + callbackCh <- callback + return true } - span.resolvedTs.Store(span.startTs) - worker := ®ionRequestWorker{tracker: newRegionTracker()} - newState := func( - regionID uint64, regionSpan heartbeatpb.TableSpan, - lockedRangeState *regionlock.LockedRangeState, - ) *regionFeedState { - state := newRegionFeedState( - regionInfo{ - verID: tikv.NewRegionVerID(regionID, 1, 1), - span: regionSpan, - rpcCtx: &tikv.RPCContext{}, - subscribedSpan: span, - lockedRangeState: lockedRangeState, - }, - uint64(span.subID), - worker, - nil, - ) - return state + span.advanceResolvedTs = func(uint64) {} + + lockedState := ®ionlock.LockedRangeState{} + lockedState.ResolvedTs.Store(100) + state := ®ionFeedState{ + region: regionInfo{ + verID: tikv.NewRegionVerID(1, 1, 1), + rpcCtx: &tikv.RPCContext{}, + subscribedSpan: span, + lockedRangeState: lockedState, + }, } - firstState := newState(1, - heartbeatpb.TableSpan{StartKey: []byte("a"), EndKey: []byte("m")}, - firstLock.LockedRangeState) - secondState := newState(2, - heartbeatpb.TableSpan{StartKey: []byte("m"), EndKey: []byte("z")}, - secondLock.LockedRangeState) - - handler := ®ionEventHandler{} - initializedEvent := func(state *regionFeedState) regionEvent { - return regionEvent{ - states: []*regionFeedState{state}, - entries: &cdcpb.Event_Entries_{Entries: &cdcpb.Event_Entries{ - Entries: []*cdcpb.Event_Row{{Type: cdcpb.Event_INITIALIZED}}, + require.True(t, quota.AcquireEvent(context.Background(), span, 10)) + handler := ®ionEventHandler{eventSink: ®ionEventSink{ + ds: newMockRegionEventSinkStream(), + memoryQuota: quota, + }} + + await := handler.Handle(span, regionEvent{ + states: []*regionFeedState{state}, + memoryBytes: 10, + entries: &cdcpb.Event_Entries_{Entries: &cdcpb.Event_Entries{ + Entries: []*cdcpb.Event_Row{{ + Type: cdcpb.Event_COMMITTED, + OpType: cdcpb.Event_Row_PUT, + CommitTs: 101, }}, - } + }}, + }) + require.True(t, await) + quotaState := getMemoryQuotaTestState(quota) + require.Equal(t, uint64(10), quotaState.used) + + callback := <-callbackCh + callback() + quotaState = getMemoryQuotaTestState(quota) + require.Zero(t, quotaState.used) +} + +func TestOnDropInvalidEventReleasesMemory(t *testing.T) { + testCases := []struct { + name string + states []*regionFeedState + }{ + {name: "empty states"}, + {name: "nil state", states: []*regionFeedState{nil}}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + quota := newMemoryQuotaController(1024, 8) + span := newTestQuotaSpan(1) + require.True(t, quota.AcquireEvent(context.Background(), span, 10)) + handler := ®ionEventHandler{eventSink: ®ionEventSink{memoryQuota: quota}} + + require.NotPanics(t, func() { + handler.OnDrop(regionEvent{states: testCase.states, memoryBytes: 10}) + }) + require.Zero(t, getMemoryQuotaTestState(quota).used) + }) } +} + +func TestSpanInitializedAfterFullRangeCoverage(t *testing.T) { + const startTs = 100 + span := &subscribedSpan{ + subID: 1, + startTs: startTs, + span: heartbeatpb.TableSpan{ + StartKey: []byte("a"), + EndKey: []byte("z"), + }, + } + firstState := newRegionFeedState(regionInfo{ + verID: tikv.NewRegionVerID(1, 1, 1), + span: heartbeatpb.TableSpan{ + StartKey: []byte("a"), + EndKey: []byte("m"), + }, + subscribedSpan: span, + lockedRangeState: ®ionlock.LockedRangeState{}, + }, uint64(span.subID), ®ionRequestWorker{}, nil) + secondState := newRegionFeedState(regionInfo{ + verID: tikv.NewRegionVerID(2, 1, 1), + span: heartbeatpb.TableSpan{ + StartKey: []byte("m"), + EndKey: []byte("z"), + }, + subscribedSpan: span, + lockedRangeState: ®ionlock.LockedRangeState{}, + }, uint64(span.subID), ®ionRequestWorker{}, nil) - require.False(t, handler.Handle(span, initializedEvent(firstState))) + span.markRegionInitialized(firstState) require.False(t, span.initialized.Load()) - require.Equal(t, uint64(0), handleResolvedTs(span, firstState, span.startTs)) - require.False(t, handler.Handle(span, initializedEvent(secondState))) + span.markRegionInitialized(secondState) require.True(t, span.initialized.Load()) - require.Equal(t, span.startTs, handleResolvedTs(span, secondState, span.startTs)) } diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index a89a0555bc..64c837f320 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -15,32 +15,26 @@ package logpuller import ( "context" - "sync" - "sync/atomic" "github.com/pingcap/log" - "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/utils/dynstream" "go.uber.org/zap" ) -// regionEventSink delivers region events to dynstream and owns push-side flow control. +// regionEventSink delivers region events to dynstream and accounts their memory. type regionEventSink struct { - // mu/cond coordinate the paused push path with pause/resume and shutdown signals. - mu sync.Mutex - cond *sync.Cond - // paused tracks whether region event pushing is temporarily held back by feedback. - paused atomic.Bool - // stopped marks the sink as shutting down so blocked pushers can exit instead of waiting for resume. - stopped atomic.Bool + ctx context.Context + ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] - // ds owns the dynstream used to deliver region events and receive flow-control feedback. - ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] + memoryQuota *memoryQuotaController } -func newRegionEventSink(failureHandler *regionFailureHandler) *regionEventSink { - sink := ®ionEventSink{} - sink.cond = sync.NewCond(&sink.mu) +func newRegionEventSink( + ctx context.Context, + failureHandler *regionFailureHandler, + memoryQuota *memoryQuotaController, +) *regionEventSink { + sink := ®ionEventSink{ctx: ctx, memoryQuota: memoryQuota} option := dynstream.NewOption() // Note: it is max batch size of the kv sent from tikv(not committed rows) @@ -48,7 +42,6 @@ func newRegionEventSink(failureHandler *regionFailureHandler) *regionEventSink { // 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 ds := dynstream.NewParallelDynamicStream( "log-puller", ®ionEventHandler{eventSink: sink, failureHandler: failureHandler}, @@ -60,8 +53,7 @@ func newRegionEventSink(failureHandler *regionFailureHandler) *regionEventSink { } 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)) @@ -77,108 +69,25 @@ func (s *regionEventSink) Wake(subID SubscriptionID) { } func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { - if s.stopped.Load() { - return - } - // fast path - if !s.paused.Load() { - s.ds.Push(subID, event) - return - } - - // slow path: wait until paused is false - s.mu.Lock() - for s.paused.Load() && !s.stopped.Load() { - s.cond.Wait() - } - stopped := s.stopped.Load() - s.mu.Unlock() - - if stopped { - return - } - s.ds.Push(subID, event) -} - -func (s *regionEventSink) Run(ctx context.Context) error { - for { - select { - case <-ctx.Done(): - s.stop() - return nil - case feedback := <-s.ds.Feedback(): - switch feedback.FeedbackType { - case dynstream.PauseArea: - s.pause() - log.Info("subscription client pause push region event") - case dynstream.ResumeArea: - s.resume() - log.Info("subscription client resume push region event") - case dynstream.ReleasePath, dynstream.ResumePath: - // Ignore it, because it is no need to pause and resume a path in puller. - } + if event.needsMemoryAccounting() { + span := event.mustFirstState().region.subscribedSpan + event.memoryBytes = uint64(event.getSize()) + // AcquireEvent only returns false after shutdown or when the + // subscription has already been stopped. + if !s.memoryQuota.AcquireEvent(s.ctx, span, event.memoryBytes) { + return } } + s.ds.Push(subID, event) } func (s *regionEventSink) UpdateMetrics() { dsMetrics := s.ds.GetMetrics() metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) - - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) == 0 { - return - } - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) != 1 { - log.Warn("subscription client should have exactly one area") - return - } - - 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())) } func (s *regionEventSink) Close() { - s.stop() + s.memoryQuota.WakeAll() s.ds.Close() } - -func (s *regionEventSink) pause() { - s.mu.Lock() - defer s.mu.Unlock() - if s.stopped.Load() || s.paused.Load() { - return - } - s.paused.Store(true) -} - -func (s *regionEventSink) resume() { - s.mu.Lock() - defer s.mu.Unlock() - if !s.paused.Load() { - return - } - s.paused.Store(false) - s.cond.Broadcast() -} - -func (s *regionEventSink) stop() { - if !s.stopped.CompareAndSwap(false, true) { - return - } - s.mu.Lock() - s.paused.Store(false) - s.cond.Broadcast() - s.mu.Unlock() -} diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go index 0698d3fb4a..ceab22fabb 100644 --- a/logservice/logpuller/region_event_sink_test.go +++ b/logservice/logpuller/region_event_sink_test.go @@ -15,28 +15,25 @@ package logpuller import ( "context" - "sync" - "sync/atomic" "testing" "time" - "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/kvproto/pkg/cdcpb" + "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/utils/dynstream" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/tikv" ) type mockRegionEventSinkStream struct { - feedbackCh chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan] - pushCount atomic.Int32 - pushCh chan struct{} - metrics dynstream.Metrics[int, SubscriptionID] + eventCh chan regionEvent + metrics dynstream.Metrics[int, SubscriptionID] } func newMockRegionEventSinkStream() *mockRegionEventSinkStream { return &mockRegionEventSinkStream{ - feedbackCh: make(chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan], 2), - pushCh: make(chan struct{}, 1), + eventCh: make(chan regionEvent, 1), } } @@ -44,15 +41,14 @@ func (s *mockRegionEventSinkStream) Start() {} func (s *mockRegionEventSinkStream) Close() {} -func (s *mockRegionEventSinkStream) Push(_ SubscriptionID, _ regionEvent) { - s.pushCount.Add(1) - s.pushCh <- struct{}{} +func (s *mockRegionEventSinkStream) Push(_ SubscriptionID, event regionEvent) { + s.eventCh <- event } func (s *mockRegionEventSinkStream) Wake(_ SubscriptionID) {} func (s *mockRegionEventSinkStream) Feedback() <-chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan] { - return s.feedbackCh + return nil } func (s *mockRegionEventSinkStream) AddPath(_ SubscriptionID, _ *subscribedSpan, _ ...dynstream.AreaSettings) error { @@ -71,191 +67,100 @@ func (s *mockRegionEventSinkStream) GetMetrics() dynstream.Metrics[int, Subscrip return s.metrics } -func newTestRegionEventSink( - ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler], -) *regionEventSink { - sink := ®ionEventSink{ds: ds} - sink.cond = sync.NewCond(&sink.mu) - return sink -} - -func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - +func TestRegionEventSinkUpdateMetrics(t *testing.T) { ds := newMockRegionEventSinkStream() - sink := newTestRegionEventSink(ds) - - runErrCh := make(chan error, 1) - go func() { - runErrCh <- sink.Run(ctx) - }() - - ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ - FeedbackType: dynstream.PauseArea, + ds.metrics = dynstream.Metrics[int, SubscriptionID]{ + EventChanSize: 33, + PendingQueueLen: 44, } - require.Eventually(t, sink.paused.Load, time.Second, 10*time.Millisecond) - - pushDone := make(chan struct{}) - go func() { - sink.Push(SubscriptionID(1), regionEvent{resolvedTs: 100}) - close(pushDone) - }() + sink := ®ionEventSink{ds: ds} - select { - case <-pushDone: - t.Fatal("Push should block while the sink is paused") - case <-time.After(100 * time.Millisecond): - } - require.Equal(t, int32(0), ds.pushCount.Load()) + sink.UpdateMetrics() - ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ - FeedbackType: dynstream.ResumeArea, - } - require.Eventually(t, func() bool { return !sink.paused.Load() }, time.Second, 10*time.Millisecond) + require.Equal(t, float64(33), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) + require.Equal(t, float64(44), testutil.ToFloat64(metricSubscriptionClientDSPendingQueueLen)) +} - select { - case <-ds.pushCh: - case <-time.After(time.Second): - t.Fatal("Push should resume after ResumeArea feedback") +func TestRegionEventSinkTracksEntriesUntilDrop(t *testing.T) { + quota := newMemoryQuotaController(1024, 8) + span := newTestQuotaSpan(1) + state := ®ionFeedState{ + region: regionInfo{subscribedSpan: span}, + worker: ®ionRequestWorker{}, } - select { - case <-pushDone: - case <-time.After(time.Second): - t.Fatal("Push should return after ResumeArea feedback") - } - - cancel() - select { - case err := <-runErrCh: - require.NoError(t, err) - case <-time.After(time.Second): - t.Fatal("Run should exit after context cancellation") + ds := newMockRegionEventSinkStream() + sink := ®ionEventSink{ + ctx: context.Background(), + ds: ds, + memoryQuota: quota, } -} - -func TestRegionEventSinkUpdateMetrics(t *testing.T) { - t.Run("empty area metrics returns after queue gauges", func(t *testing.T) { - ds := newMockRegionEventSinkStream() - ds.metrics = dynstream.Metrics[int, SubscriptionID]{ - EventChanSize: 11, - PendingQueueLen: 22, - } - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "max", - "default", - "default", - ).Set(123) - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "used", - "default", - "default", - ).Set(456) - - sink := ®ionEventSink{ - ds: ds, - } - sink.UpdateMetrics() - - require.Equal(t, float64(11), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) - require.Equal(t, float64(22), testutil.ToFloat64(metricSubscriptionClientDSPendingQueueLen)) - require.Equal(t, float64(123), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "max", - "default", - "default", - ))) - require.Equal(t, float64(456), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "used", - "default", - "default", - ))) - }) - - t.Run("single area metrics updates memory gauges", func(t *testing.T) { - ds := newMockRegionEventSinkStream() - ds.metrics = dynstream.Metrics[int, SubscriptionID]{ - EventChanSize: 33, - PendingQueueLen: 44, - MemoryControl: dynstream.MemoryMetric[int, SubscriptionID]{ - AreaMemoryMetrics: []dynstream.AreaMemoryMetric[int, SubscriptionID]{ - { - UsedMemoryValue: 55, - MaxMemoryValue: 66, - PathMaxMemoryValue: 66, - }, - }, - }, - } - - sink := ®ionEventSink{ - ds: ds, - } - sink.UpdateMetrics() - - require.Equal(t, float64(33), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) - require.Equal(t, float64(44), testutil.ToFloat64(metricSubscriptionClientDSPendingQueueLen)) - require.Equal(t, float64(66), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "max", - "default", - "default", - ))) - require.Equal(t, float64(55), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "used", - "default", - "default", - ))) + sink.Push(span.subID, regionEvent{ + states: []*regionFeedState{state}, + entries: &cdcpb.Event_Entries_{Entries: &cdcpb.Event_Entries{ + Entries: []*cdcpb.Event_Row{{Key: []byte("key"), Value: []byte("value")}}, + }}, }) + pushed := <-ds.eventCh + require.NotZero(t, pushed.memoryBytes) + quotaState := getMemoryQuotaTestState(quota) + require.NotZero(t, quotaState.used) + + (®ionEventHandler{eventSink: sink}).OnDrop(pushed) + quotaState = getMemoryQuotaTestState(quota) + require.Zero(t, quotaState.used) } -func TestRegionEventSinkRunCancelUnblocksPush(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - - ds := newMockRegionEventSinkStream() - sink := newTestRegionEventSink(ds) - - runErrCh := make(chan error, 1) - go func() { - runErrCh <- sink.Run(ctx) - }() - - ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ - FeedbackType: dynstream.PauseArea, - } - require.Eventually(t, sink.paused.Load, time.Second, 10*time.Millisecond) - - pushDone := make(chan struct{}) - go func() { - sink.Push(SubscriptionID(1), regionEvent{resolvedTs: 100}) - close(pushDone) - }() - - select { - case <-pushDone: - t.Fatal("Push should block while the sink is paused") - case <-time.After(100 * time.Millisecond): +func TestRegionEventSinkRemovePathReleasesQueuedEventMemory(t *testing.T) { + quota := newMemoryQuotaController(1024*1024, 8) + sink := newRegionEventSink(context.Background(), nil, quota) + defer sink.Close() + + span := newTestQuotaSpan(1) + span.resolvedTs.Store(100) + callbackCh := make(chan func(), 1) + span.consumeKVEvents = func(_ []common.RawKVEntry, callback func()) bool { + callbackCh <- callback + return true } - require.Equal(t, int32(0), ds.pushCount.Load()) - - cancel() - - select { - case <-pushDone: - case <-time.After(time.Second): - t.Fatal("Push should be unblocked by Run context cancellation") + span.advanceResolvedTs = func(uint64) {} + sink.AddPath(span) + + worker := ®ionRequestWorker{} + region := newTestQuotaRegion(span) + region.rpcCtx = &tikv.RPCContext{} + state := newRegionFeedState( + region, + uint64(span.subID), + worker, + nil, + ) + newEvent := func(commitTs uint64) regionEvent { + return regionEvent{ + states: []*regionFeedState{state}, + entries: &cdcpb.Event_Entries_{Entries: &cdcpb.Event_Entries{ + Entries: []*cdcpb.Event_Row{{ + Type: cdcpb.Event_COMMITTED, + OpType: cdcpb.Event_Row_PUT, + CommitTs: commitTs, + }}, + }}, + } } - require.Equal(t, int32(0), ds.pushCount.Load()) - select { - case err := <-runErrCh: - require.NoError(t, err) - case <-time.After(time.Second): - t.Fatal("Run should exit after context cancellation") - } + sink.Push(span.subID, newEvent(101)) + callback := <-callbackCh + firstEventUsed := getMemoryQuotaTestState(quota).used + require.NotZero(t, firstEventUsed) + + // The first event blocks the path until callback is invoked, so this event + // remains queued when the path is removed. + sink.Push(span.subID, newEvent(102)) + require.Greater(t, getMemoryQuotaTestState(quota).used, firstEventUsed) + require.NoError(t, sink.RemovePath(span.subID)) + + callback() + require.Eventually(t, func() bool { + return getMemoryQuotaTestState(quota).used == 0 + }, time.Second, 10*time.Millisecond) } diff --git a/logservice/logpuller/region_request_scheduler.go b/logservice/logpuller/region_request_scheduler.go index 4d4e5caadd..2118b246ba 100644 --- a/logservice/logpuller/region_request_scheduler.go +++ b/logservice/logpuller/region_request_scheduler.go @@ -42,6 +42,7 @@ type regionRequestScheduler struct { upstream *upstreamHandle eventSink *regionEventSink failureHandler *regionFailureHandler + memoryQuota *memoryQuotaController // taskQueue orders all regions before they are assigned to a TiKV store. taskQueue *priorityqueue.PriorityQueue[*regionPriorityTask] @@ -63,6 +64,7 @@ func newRegionRequestScheduler( upstream *upstreamHandle, eventSink *regionEventSink, failureHandler *regionFailureHandler, + memoryQuota *memoryQuotaController, ) *regionRequestScheduler { pullerConfig := config.GetGlobalServerConfig().Debug.Puller workerCount := regionRequestWorkerPerStore @@ -71,6 +73,7 @@ func newRegionRequestScheduler( upstream: upstream, eventSink: eventSink, failureHandler: failureHandler, + memoryQuota: memoryQuota, taskQueue: priorityqueue.New[*regionPriorityTask](), workerCount: workerCount, workerWindow: workerWindow, @@ -182,6 +185,7 @@ func (s *regionRequestScheduler) getOrCreateStore( s.workerCount, s.workerWindow, s.maxWindowMultiplier, + s.memoryQuota, ) // The scheduler run loop is the only writer. Publish the store after its // immutable worker list is complete, then start its workers. diff --git a/logservice/logpuller/region_request_scheduler_test.go b/logservice/logpuller/region_request_scheduler_test.go index 86b6dd37be..7ae501c8f1 100644 --- a/logservice/logpuller/region_request_scheduler_test.go +++ b/logservice/logpuller/region_request_scheduler_test.go @@ -35,12 +35,12 @@ func TestRegionRequestSchedulerBroadcastDeregisterUsesWorkerControlQueue(t *test worker1 := ®ionRequestWorker{ storeAddr: "store-1", - admission: newRegionAdmissionController(1, 1), + admission: newTestRegionAdmissionController(1, 1), controlQueue: newControlQueue(), } worker2 := ®ionRequestWorker{ storeAddr: "store-2", - admission: newRegionAdmissionController(1, 1), + admission: newTestRegionAdmissionController(1, 1), controlQueue: newControlQueue(), } store1 := ®ionRequestStore{workers: []*regionRequestWorker{worker1}} @@ -72,8 +72,8 @@ func TestRegionRequestSchedulerBroadcastDeregisterUsesWorkerControlQueue(t *test func TestRegionRequestSchedulerRequestedRegionCountAggregatesStores(t *testing.T) { scheduler := ®ionRequestScheduler{} - worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} + worker1 := ®ionRequestWorker{admission: newTestRegionAdmissionController(2, 1)} + worker2 := ®ionRequestWorker{admission: newTestRegionAdmissionController(2, 1)} scheduler.stores.Store("store-1", ®ionRequestStore{workers: []*regionRequestWorker{worker1}}) scheduler.stores.Store("store-2", ®ionRequestStore{workers: []*regionRequestWorker{worker2}}) @@ -125,7 +125,7 @@ func TestRegionRequestSchedulerReschedulesRegionWhenStoreSubmitFails(t *testing. context.Background(), rawSpan.StartKey, rawSpan.EndKey, location.Region.GetID(), location.Region.GetVer()) require.Equal(t, regionlock.LockRangeStatusSuccess, lockRes.Status) - admission := newRegionAdmissionController(1, 1) + admission := newTestRegionAdmissionController(1, 1) admission.close() store := ®ionRequestStore{workers: []*regionRequestWorker{{admission: admission}}} diff --git a/logservice/logpuller/region_request_store.go b/logservice/logpuller/region_request_store.go index db7de05006..fe3f1202a5 100644 --- a/logservice/logpuller/region_request_store.go +++ b/logservice/logpuller/region_request_store.go @@ -37,13 +37,21 @@ func newRegionRequestStore( workerCount int, workerWindow int, maxWindowMultiplier int, + memoryQuota *memoryQuotaController, ) *regionRequestStore { store := ®ionRequestStore{ workers: make([]*regionRequestWorker, 0, workerCount), } for range workerCount { store.workers = append(store.workers, newRegionRequestWorker( - upstream, eventSink, failureHandler, storeAddr, workerWindow, maxWindowMultiplier)) + upstream, + eventSink, + failureHandler, + storeAddr, + workerWindow, + maxWindowMultiplier, + memoryQuota, + )) } return store } diff --git a/logservice/logpuller/region_request_store_test.go b/logservice/logpuller/region_request_store_test.go index abc84add3e..e9d470e633 100644 --- a/logservice/logpuller/region_request_store_test.go +++ b/logservice/logpuller/region_request_store_test.go @@ -22,8 +22,8 @@ import ( ) func TestRegionRequestStoreDistributesRegionsAcrossWorkers(t *testing.T) { - worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} + worker1 := ®ionRequestWorker{admission: newTestRegionAdmissionController(1, 1)} + worker2 := ®ionRequestWorker{admission: newTestRegionAdmissionController(1, 1)} store := ®ionRequestStore{ workers: []*regionRequestWorker{worker1, worker2}, } @@ -39,8 +39,8 @@ func TestRegionRequestStoreDistributesRegionsAcrossWorkers(t *testing.T) { } func TestRegionRequestStoreRequestedRegionCountIncludesPendingAndInflight(t *testing.T) { - worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} + worker1 := ®ionRequestWorker{admission: newTestRegionAdmissionController(2, 1)} + worker2 := ®ionRequestWorker{admission: newTestRegionAdmissionController(2, 1)} store := ®ionRequestStore{ workers: []*regionRequestWorker{worker1, worker2}, } @@ -59,8 +59,8 @@ func TestRegionRequestStoreRequestedRegionCountIncludesPendingAndInflight(t *tes } func TestRegionRequestStoreCloseClosesWorkerAdmissions(t *testing.T) { - worker1 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} - worker2 := ®ionRequestWorker{admission: newRegionAdmissionController(1, 1)} + worker1 := ®ionRequestWorker{admission: newTestRegionAdmissionController(1, 1)} + worker2 := ®ionRequestWorker{admission: newTestRegionAdmissionController(1, 1)} store := ®ionRequestStore{ workers: []*regionRequestWorker{worker1, worker2}, } diff --git a/logservice/logpuller/region_request_worker.go b/logservice/logpuller/region_request_worker.go index 3d3e0dca4b..3943c067fb 100644 --- a/logservice/logpuller/region_request_worker.go +++ b/logservice/logpuller/region_request_worker.go @@ -112,6 +112,7 @@ func newRegionRequestWorker( storeAddr string, currentWindow int, maxWindowMultiplier int, + memoryQuota *memoryQuotaController, ) *regionRequestWorker { workerID := workerIDGen.Add(1) return ®ionRequestWorker{ @@ -120,9 +121,14 @@ func newRegionRequestWorker( eventSink: eventSink, failureHandler: failureHandler, storeAddr: storeAddr, - admission: newRegionAdmissionController(currentWindow, maxWindowMultiplier), - controlQueue: newControlQueue(), - tracker: newRegionTracker(), + admission: newRegionAdmissionController( + currentWindow, + maxWindowMultiplier, + memoryQuota, + upstream.pdClock, + ), + controlQueue: newControlQueue(), + tracker: newRegionTracker(), } } diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index 32228d8e7c..6362bf7ebe 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -143,7 +143,7 @@ func TestRunStreamCancelsBlockingReceiveWhenSenderExits(t *testing.T) { defer pdClient.Close() cluster.AddStore(1, storeAddr) - admission := newRegionAdmissionController(1, 1) + admission := newTestRegionAdmissionController(1, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), @@ -272,7 +272,7 @@ func errCacheLen(handler *regionFailureHandler) int { } func TestRegionRequestWorkerIgnoresDuplicateActiveRegion(t *testing.T) { - admission := newRegionAdmissionController(10, 1) + admission := newTestRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, storeAddr: "store-1", @@ -431,7 +431,7 @@ func benchmarkDispatchResolvedTsEvent(b *testing.B, regionCount int, useLegacy b } func TestWaitForRegionRequestDrainsIdleControlQueue(t *testing.T) { - admission := newRegionAdmissionController(1, 1) + admission := newTestRegionAdmissionController(1, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), @@ -513,7 +513,7 @@ func BenchmarkDispatchResolvedTsEventSmallBatchCurrent(b *testing.B) { } func TestStoppedStateRemovesSentRequest(t *testing.T) { - admission := newRegionAdmissionController(10, 1) + admission := newTestRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, tracker: newRegionTracker(), @@ -539,7 +539,7 @@ func TestRunStreamFailurePushesTrackedRegionToEventSink(t *testing.T) { upstream: &upstreamHandle{pd: pdClient, credential: &security.Credential{}}, eventSink: ®ionEventSink{ds: ds}, failureHandler: handler, - admission: newRegionAdmissionController(10, 1), + admission: newTestRegionAdmissionController(10, 1), controlQueue: newControlQueue(), tracker: newRegionTracker(), storeAddr: "127.0.0.1:1", @@ -589,7 +589,7 @@ func TestRunStreamFailureReportsPendingRegionsToFailureHandler(t *testing.T) { upstream: &upstreamHandle{pd: pdClient, credential: &security.Credential{}}, eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, failureHandler: handler, - admission: newRegionAdmissionController(10, 1), + admission: newTestRegionAdmissionController(10, 1), controlQueue: newControlQueue(), tracker: newRegionTracker(), storeAddr: "127.0.0.1:1", @@ -618,7 +618,7 @@ func TestRunStreamFailureReportsPendingRegionsToFailureHandler(t *testing.T) { } func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { - admission := newRegionAdmissionController(10, 1) + admission := newTestRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), @@ -649,7 +649,7 @@ func TestProcessRegionSendTaskSendFailureCleansSentRequest(t *testing.T) { } func TestProcessRegionSendTaskDoesNotSendRemovedRequest(t *testing.T) { - admission := newRegionAdmissionController(1, 1) + admission := newTestRegionAdmissionController(1, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), @@ -697,7 +697,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - admission := newRegionAdmissionController(10, 1) + admission := newTestRegionAdmissionController(10, 1) worker := ®ionRequestWorker{ admission: admission, controlQueue: newControlQueue(), @@ -732,7 +732,7 @@ func TestProcessRegionSendTaskSendEOFIsRetriable(t *testing.T) { func TestProcessRegionSendTaskHandlesDeregisterFromControlQueue(t *testing.T) { ds := &mockRegionEventDynamicStream{} worker := ®ionRequestWorker{ - admission: newRegionAdmissionController(1, 1), + admission: newTestRegionAdmissionController(1, 1), controlQueue: newControlQueue(), storeAddr: "store-1", upstream: &upstreamHandle{clusterID: 42}, diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 8fa9d17238..af039a9ebd 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -137,6 +137,8 @@ type subscriptionClient struct { spanRegistry *spanRegistry // regionScheduler assigns locked region requests to per-store workers. regionScheduler *regionRequestScheduler + // memoryQuota owns event-memory accounting and initial-scan admission. + memoryQuota *memoryQuotaController // rangeTaskCh is used to receive range tasks. // The tasks will be handled in `handleRangeTask` goroutine. @@ -167,18 +169,26 @@ func NewSubscriptionClient( resolveLockRateLimiter: newResolveLockRateLimiter(), } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) + pullerConfig := config.GetGlobalServerConfig().Debug.Puller + subClient.memoryQuota = newMemoryQuotaController( + pullerConfig.MemoryQuota, pullerConfig.ScanBaseSize) subClient.failureHandler = newRegionFailureHandler( subClient.upstream.regionCache, subClient.onTableDrained, subClient.scheduleRegionRequest, subClient.scheduleRangeRequest, ) - subClient.eventSink = newRegionEventSink(subClient.failureHandler) + subClient.eventSink = newRegionEventSink( + subClient.ctx, + subClient.failureHandler, + subClient.memoryQuota, + ) subClient.spanRegistry = newSpanRegistry(subClient.upstream.pd, subClient.upstream.pdClock) subClient.regionScheduler = newRegionRequestScheduler( subClient.upstream, subClient.eventSink, subClient.failureHandler, + subClient.memoryQuota, ) return subClient } @@ -202,6 +212,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ticker.C: s.regionScheduler.UpdateMetrics() s.eventSink.UpdateMetrics() + s.memoryQuota.UpdateMetrics() s.spanRegistry.UpdateMetrics() } } @@ -283,7 +294,6 @@ func (s *subscriptionClient) Run(ctx context.Context) error { // actual startup order. g.Go(func() error { return s.handleRangeTasks(ctx) }) g.Go(func() error { return s.regionScheduler.Run(ctx, g) }) - g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.failureHandler.Run(ctx) }) g.Go(func() error { return s.spanRegistry.Run(ctx) }) g.Go(func() error { return s.handleResolveLockTasks(ctx) }) @@ -309,6 +319,8 @@ func (s *subscriptionClient) setTableStopped(rt *subscribedSpan) { // Set stopped to true so we can stop handling region events from the table, // then notify every existing worker to deregister the subscription. if rt.stopped.CompareAndSwap(false, true) { + // Wake event receivers and scan admission so they can observe stopped. + s.memoryQuota.WakeAll() s.regionScheduler.BroadcastDeregister(rt.subID, rt.filterLoop) if rt.rangeLock.Stop() { s.onTableDrained(rt) diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 818a745e54..33d9a827cd 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -306,6 +306,7 @@ func TestResolveLockTaskDroppedWhenChannelFull(t *testing.T) { func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { client := &subscriptionClient{ resolveLockTaskCh: make(chan resolveLockTask, 1), + memoryQuota: newMemoryQuotaController(1024, 8), } client.ctx, client.cancel = context.WithCancel(context.Background()) defer client.cancel() @@ -424,27 +425,49 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] } func TestRegionEventSinkPushUnblocksOnClientClose(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + quota := newMemoryQuotaController(10, 8) + span := &subscribedSpan{subID: 1} + require.True(t, quota.AcquireEvent(ctx, span, 20)) + t.Cleanup(func() { quota.ReleaseEvent(20) }) + sink := ®ionEventSink{ - ds: &mockDynamicStream{}, + ctx: ctx, + ds: &mockDynamicStream{}, + memoryQuota: quota, } - sink.cond = sync.NewCond(&sink.mu) client := &subscriptionClient{eventSink: sink} client.regionScheduler = ®ionRequestScheduler{ taskQueue: priorityqueue.New[*regionPriorityTask](), } - client.ctx, client.cancel = context.WithCancel(context.Background()) + client.ctx = ctx + client.cancel = cancel - sink.paused.Store(true) + event := regionEvent{ + states: []*regionFeedState{{ + region: regionInfo{subscribedSpan: span}, + }}, + entries: &cdcpb.Event_Entries_{ + Entries: &cdcpb.Event_Entries{ + Entries: []*cdcpb.Event_Row{{ + Key: []byte("key"), + Value: []byte("value"), + }}, + }, + }, + } done := make(chan struct{}) go func() { - sink.Push(SubscriptionID(1), regionEvent{}) + sink.Push(SubscriptionID(1), event) close(done) }() select { case <-done: - t.Fatal("regionEventSink.Push should block when paused") + t.Fatal("regionEventSink.Push should block when event memory is exhausted") case <-time.After(100 * time.Millisecond): } diff --git a/metrics/grafana/ticdc_new_arch.json b/metrics/grafana/ticdc_new_arch.json index 0a3f4bbd37..cb01262efe 100644 --- a/metrics/grafana/ticdc_new_arch.json +++ b/metrics/grafana/ticdc_new_arch.json @@ -8166,6 +8166,13 @@ "interval": "", "legendFormat": "{{instance}}-{{type}}", "refId": "A" + }, + { + "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}}-quota-{{type}}", + "refId": "B" } ], "thresholds": [], @@ -8815,6 +8822,214 @@ "align": false, "alignLevel": null } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Event receivers blocked at the memory hard limit and region scans blocked at the scan admission gate.", + "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": "ticdc_log_puller_memory_quota_event_waiter_count{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}", + "interval": "", + "legendFormat": "{{instance}}-event-waiters", + "refId": "A" + }, + { + "exemplar": true, + "expr": "ticdc_log_puller_memory_quota_scan_waiter_count{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}", + "interval": "", + "legendFormat": "{{instance}}-scan-waiters", + "refId": "B" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Waiters", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "short", + "logBase": 1, + "min": "0", + "show": true + }, + { + "format": "short", + "logBase": 1, + "show": false + } + ], + "yaxis": { + "align": false + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Time spent waiting at the event hard limit or the scan admission gate.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 0, + "fillGradient": 0, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 57 + }, + "hiddenSeries": false, + "id": 26002, + "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": "histogram_quantile(0.99, sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_bucket{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (le, instance))", + "interval": "", + "legendFormat": "{{instance}}-event-p99", + "refId": "A" + }, + { + "exemplar": true, + "expr": "sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_sum{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance) / sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_count{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance)", + "interval": "", + "legendFormat": "{{instance}}-event-avg", + "refId": "B" + }, + { + "exemplar": true, + "expr": "histogram_quantile(0.99, sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_bucket{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (le, instance))", + "interval": "", + "legendFormat": "{{instance}}-scan-p99", + "refId": "C" + }, + { + "exemplar": true, + "expr": "sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_sum{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance) / sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_count{k8s_cluster=~\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance)", + "interval": "", + "legendFormat": "{{instance}}-scan-avg", + "refId": "D" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Wait Duration", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "s", + "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 f920ac2ccd..d11e96e9c1 100644 --- a/metrics/nextgengrafana/ticdc_new_arch_next_gen.json +++ b/metrics/nextgengrafana/ticdc_new_arch_next_gen.json @@ -8166,6 +8166,13 @@ "interval": "", "legendFormat": "{{instance}}-{{type}}", "refId": "A" + }, + { + "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}}-quota-{{type}}", + "refId": "B" } ], "thresholds": [], @@ -8815,6 +8822,214 @@ "align": false, "alignLevel": null } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Event receivers blocked at the memory hard limit and region scans blocked at the scan admission gate.", + "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": "ticdc_log_puller_memory_quota_event_waiter_count{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}", + "interval": "", + "legendFormat": "{{instance}}-event-waiters", + "refId": "A" + }, + { + "exemplar": true, + "expr": "ticdc_log_puller_memory_quota_scan_waiter_count{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}", + "interval": "", + "legendFormat": "{{instance}}-scan-waiters", + "refId": "B" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Waiters", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "short", + "logBase": 1, + "min": "0", + "show": true + }, + { + "format": "short", + "logBase": 1, + "show": false + } + ], + "yaxis": { + "align": false + } + }, + { + "aliasColors": {}, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Time spent waiting at the event hard limit or the scan admission gate.", + "fieldConfig": { + "defaults": {}, + "overrides": [] + }, + "fill": 0, + "fillGradient": 0, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 57 + }, + "hiddenSeries": false, + "id": 26002, + "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": "histogram_quantile(0.99, sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_bucket{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (le, instance))", + "interval": "", + "legendFormat": "{{instance}}-event-p99", + "refId": "A" + }, + { + "exemplar": true, + "expr": "sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_sum{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance) / sum(rate(ticdc_log_puller_memory_quota_event_wait_duration_count{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance)", + "interval": "", + "legendFormat": "{{instance}}-event-avg", + "refId": "B" + }, + { + "exemplar": true, + "expr": "histogram_quantile(0.99, sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_bucket{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (le, instance))", + "interval": "", + "legendFormat": "{{instance}}-scan-p99", + "refId": "C" + }, + { + "exemplar": true, + "expr": "sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_sum{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance) / sum(rate(ticdc_log_puller_memory_quota_scan_wait_duration_count{k8s_cluster=~\"$k8s_cluster\", sharedpool_id=\"$tidb_cluster\", instance=~\"$ticdc_instance\"}[1m])) by (instance)", + "interval": "", + "legendFormat": "{{instance}}-scan-avg", + "refId": "D" + } + ], + "thresholds": [], + "timeFrom": null, + "timeRegions": [], + "timeShift": null, + "title": "Memory Quota Wait Duration", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [] + }, + "yaxes": [ + { + "format": "s", + "logBase": 1, + "min": "0", + "show": true + }, + { + "format": "short", + "logBase": 1, + "show": false + } + ], + "yaxis": { + "align": false + } } ], "title": "Log Puller", diff --git a/pkg/config/debug.go b/pkg/config/debug.go index 54a5ba868c..f9ffe4aec1 100644 --- a/pkg/config/debug.go +++ b/pkg/config/debug.go @@ -25,6 +25,13 @@ const ( // DefaultOldStartTsScanLowPriorityThreshold is the default lag threshold for // classifying scan tasks as low priority. DefaultOldStartTsScanLowPriorityThreshold = 10 * time.Minute + + // DefaultLogPullerMemoryQuota is the default Log Puller soft memory limit. + DefaultLogPullerMemoryQuota uint64 = 1024 * 1024 * 1024 + + // DefaultLogPullerScanBaseSize is the default base memory estimate for one + // initial scan. + DefaultLogPullerScanBaseSize uint64 = 8 * 1024 * 1024 ) // DebugConfig represents config for ticdc unexposed feature configurations @@ -84,6 +91,14 @@ type PullerConfig struct { // Scans within this threshold are scheduled as high priority. Older scans // remain low priority until their span catches up once. OldStartTsScanLowPriorityThreshold TomlDuration `toml:"old-start-ts-scan-low-priority-threshold" json:"old_start_ts_scan_low_priority_threshold"` + // MemoryQuota is the log puller's local soft memory limit in bytes. + MemoryQuota uint64 `toml:"memory-quota" json:"memory_quota"` + // ScanBaseSize is the base memory estimate reserved for one admitted initial + // scan. The actual estimate grows logarithmically with scan lag, up to a + // bounded multiple of this value. The estimate contributes to MemoryQuota + // pressure and throttles new low-priority scans; it is not an actual memory + // allocation or a per-scan hard limit. + ScanBaseSize uint64 `toml:"scan-base-size" json:"scan_base_size"` } // NewDefaultPullerConfig return the default puller configuration @@ -96,6 +111,8 @@ func NewDefaultPullerConfig() *PullerConfig { RegionRequestMaxWindowMultiplier: 4, // Allows high-priority scans to use up to 4 * PendingRegionRequestQueueSize. OldStartTsScanLowPriorityThreshold: TomlDuration( DefaultOldStartTsScanLowPriorityThreshold), + MemoryQuota: DefaultLogPullerMemoryQuota, + ScanBaseSize: DefaultLogPullerScanBaseSize, } } @@ -117,6 +134,16 @@ func (c *PullerConfig) ValidateAndAdjust() { if c.OldStartTsScanLowPriorityThreshold <= 0 { c.OldStartTsScanLowPriorityThreshold = TomlDuration(DefaultOldStartTsScanLowPriorityThreshold) } + if c.MemoryQuota == 0 { + log.Warn("log puller memory quota must be positive, use default value", + zap.Uint64("default", defaultCfg.MemoryQuota)) + c.MemoryQuota = defaultCfg.MemoryQuota + } + if c.ScanBaseSize == 0 { + log.Warn("log puller scan base size must be positive, use default value", + zap.Uint64("default", defaultCfg.ScanBaseSize)) + c.ScanBaseSize = defaultCfg.ScanBaseSize + } } type EventStoreConfig struct { diff --git a/pkg/config/debug_test.go b/pkg/config/debug_test.go index ced2d50bff..3f2b3428fa 100644 --- a/pkg/config/debug_test.go +++ b/pkg/config/debug_test.go @@ -29,6 +29,8 @@ func TestPullerConfigValidateAndAdjustRegionRequestWindow(t *testing.T) { TomlDuration(DefaultOldStartTsScanLowPriorityThreshold), defaultCfg.OldStartTsScanLowPriorityThreshold, ) + require.Equal(t, uint64(1024*1024*1024), defaultCfg.MemoryQuota) + require.Equal(t, uint64(8*1024*1024), defaultCfg.ScanBaseSize) cfg := &PullerConfig{ PendingRegionRequestQueueSize: -1, @@ -39,4 +41,6 @@ func TestPullerConfigValidateAndAdjustRegionRequestWindow(t *testing.T) { require.Equal(t, defaultCfg.PendingRegionRequestQueueSize, cfg.PendingRegionRequestQueueSize) require.Equal(t, defaultCfg.RegionRequestMaxWindowMultiplier, cfg.RegionRequestMaxWindowMultiplier) require.Equal(t, defaultCfg.OldStartTsScanLowPriorityThreshold, cfg.OldStartTsScanLowPriorityThreshold) + require.Equal(t, defaultCfg.MemoryQuota, cfg.MemoryQuota) + require.Equal(t, defaultCfg.ScanBaseSize, cfg.ScanBaseSize) } diff --git a/pkg/metrics/log_puller.go b/pkg/metrics/log_puller.go index d50c5a8526..34af5325ab 100644 --- a/pkg/metrics/log_puller.go +++ b/pkg/metrics/log_puller.go @@ -64,6 +64,43 @@ 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"}) + LogPullerMemoryQuotaEventWaiterCount = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: "ticdc", + Subsystem: "log_puller", + Name: "memory_quota_event_waiter_count", + Help: "The number of event receivers waiting at the log puller memory hard limit.", + }) + LogPullerMemoryQuotaEventWaitDuration = prometheus.NewHistogram( + prometheus.HistogramOpts{ + Namespace: "ticdc", + Subsystem: "log_puller", + Name: "memory_quota_event_wait_duration", + Help: "The duration in seconds that an event receiver waits at the log puller memory hard limit.", + Buckets: prometheus.ExponentialBuckets(0.001, 2, 24), + }) + LogPullerMemoryQuotaScanWaiterCount = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: "ticdc", + Subsystem: "log_puller", + Name: "memory_quota_scan_waiter_count", + Help: "The number of region scans waiting at the log puller memory quota gate.", + }) + LogPullerMemoryQuotaScanWaitDuration = prometheus.NewHistogram( + prometheus.HistogramOpts{ + Namespace: "ticdc", + Subsystem: "log_puller", + Name: "memory_quota_scan_wait_duration", + Help: "The duration in seconds that a region scan waits at the log puller memory quota gate.", + Buckets: prometheus.ExponentialBuckets(0.001, 2, 24), + }) SubscriptionClientResolvedTsLagGauge = prometheus.NewGauge( prometheus.GaugeOpts{ @@ -156,6 +193,11 @@ func initLogPullerMetrics(registry *prometheus.Registry) { registry.MustRegister(LogPullerPrewriteCacheRowNum) registry.MustRegister(LogPullerMatcherCount) registry.MustRegister(LogPullerResolvedTsLag) + registry.MustRegister(LogPullerMemoryQuota) + registry.MustRegister(LogPullerMemoryQuotaEventWaiterCount) + registry.MustRegister(LogPullerMemoryQuotaEventWaitDuration) + registry.MustRegister(LogPullerMemoryQuotaScanWaiterCount) + registry.MustRegister(LogPullerMemoryQuotaScanWaitDuration) registry.MustRegister(SubscriptionClientRequestedRegionCount) registry.MustRegister(RegionRequestFinishScanDuration) registry.MustRegister(SubscriptionClientSubscribedRegionCount)