telemetry: extract bounded staging queue#1945
Conversation
Move the mutex-backed bounded staging queue out of nixlTelemetry into a focused internal nixlTelemetryStagingQueue class, with no behavior change. The queue owns event storage and its capacity reserve, the producer mutex, capacity enforcement, all-or-none single and batch insertion, the swap-drain, and the staging-drop counter. nixlTelemetry keeps metric activation, event selection, periodic scheduling, exporter dispatch, and synthetic dropped-event construction. Metric gating stays at the producer (deactivated types never enter the queue and are not counted as drops), batch insertion remains all-or-none, drop accounting stays exact, the drain still swaps under one mutex and exports outside it, and shutdown behavior is unchanged. This extracted interface (tryPush/tryPushBatch/takePending/exchangeDropped/capacity) is the stable seam for the later NIX-1544 low-lock implementation. Add focused and concurrent-producer unit tests under test/gtest/unit/telemetry, run through the existing CTest unit target. NIX-1542 Signed-off-by: Efraim Eygin <eeygin@nvidia.com>
|
👋 Hi e-eygin! Thank you for contributing to ai-dynamo/nixl. Your PR reviewers will review your contribution then trigger the CI to test your changes. 🚀 |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Enterprise Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughA bounded telemetry staging queue replaces manual event buffering and drop accounting. Telemetry updates and transfer statistics enqueue through the queue, exports drain pending events, and unit tests cover queue semantics and concurrency. ChangesTelemetry staging queue
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant TelemetryProducer
participant nixlTelemetryStagingQueue
participant PeriodicExportTask
TelemetryProducer->>nixlTelemetryStagingQueue: tryPush or tryPushBatch
PeriodicExportTask->>nixlTelemetryStagingQueue: takePending
nixlTelemetryStagingQueue-->>PeriodicExportTask: pending events
PeriodicExportTask->>nixlTelemetryStagingQueue: takeNumDropped
nixlTelemetryStagingQueue-->>PeriodicExportTask: dropped-event count
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/core/telemetry/telemetry_staging_queue.cpp`:
- Line 25: Mark the local std::lock_guard instance at the shown mutex
acquisition, and the corresponding instances at the other two lock sites, as
const. Do not change locking behavior or the surrounding critical sections.
In `@src/core/telemetry/telemetry_staging_queue.h`:
- Around line 43-45: Add a Doxygen `@brief` comment immediately before the public
constructor nixlTelemetryStagingQueue, documenting its purpose and the capacity
parameter, including whether a capacity of 0 is valid according to the
implementation’s behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Enterprise
Run ID: 2ad69118-fca5-4a22-95a4-de4dbc274da2
📒 Files selected for processing (8)
src/core/meson.buildsrc/core/telemetry/telemetry.cppsrc/core/telemetry/telemetry.hsrc/core/telemetry/telemetry_staging_queue.cppsrc/core/telemetry/telemetry_staging_queue.htest/gtest/unit/meson.buildtest/gtest/unit/telemetry/meson.buildtest/gtest/unit/telemetry/telemetry_staging_queue_test.cpp
Mark the three local std::lock_guard instances in nixlTelemetryStagingQueue const, matching the repo convention already used in addXferStats. Add a Doxygen @brief to the constructor documenting the capacity parameter and that a zero capacity is accepted (the queue then drops every push); rejecting a zero telemetry buffer size stays the caller's responsibility. No behavior change. Signed-off-by: Efraim Eygin <eeygin@nvidia.com>
|
/build |
|
🤖 CI Triage Agent — TL;DR: The vLLM sanity test failed because both vLLM engines aborted at startup with Full analysisSummary: Stage 535 "Run vLLM sanity" failed; both prefill/decode vLLM servers crashed on EngineCore init and the Root cause: vLLM 0.23.0 EngineCore computed negative available KV-cache memory under Implicated commit: unknown (the 0.3 value predates this change — likely a vLLM-version bump making the budget too tight; not attributable to a specific commit from the logs) File: Suggested fix: Raise the vLLM memory budget so KV-cache memory is positive after weights + overhead + the 1 GB NIXL buffer — e.g. set a vLLM-specific Related: none found in the provided tooling; the vLLM error links to https://docs.vllm.ai/en/latest/configuration/conserving_memory/ for the |
|
🤖 CI Triage Agent — TL;DR: The Full analysisSummary: Stage 248 ("Run CPP tests", UCX v1.22.x variant) aborted after ~61 min when Jenkins sent an interrupt during Root cause: Wall-clock kill caused by a severe per-test slowdown (not a hang — output is continuous to the kill). Telemetry-path tests regressed sharply: Implicated commit: [REDACTED:Hex High Entropy String] (PR #1945, "telemetry: extract bounded staging queue") File: src/core/telemetry/telemetry.cpp — hot-path Suggested fix: Profile the telemetry enqueue path introduced by #1945 in the threadpool/dedicated-progress-thread configuration and remove the per-event contention — e.g. avoid taking Related: PR #1945 (#1945); UCX bump PR #1868 (v1.22.x) which is the branch that tips the suite over the wall-clock budget; PR #1880 (ucx dedicated-thread fix) for context on the threadpool path.
|
- tryPushBatch: rewrite the capacity check as `count > capacity_ - events_.size()` to avoid a theoretical size_t overflow in the sum (events_.size() <= capacity_ always holds, so the subtraction never underflows). - telemetry.h: drop the now-unused <vector> and <mutex> includes; neither std::vector nor std::mutex is referenced there after the extraction. No behavior change. Signed-off-by: Efraim Eygin <eeygin@nvidia.com>
- tryPushBatch now takes std::span<const nixlTelemetryEvent> instead of a
pointer+count pair; the repo builds as C++20, so no C++17 workaround is
needed. addXferStats submits its activated prefix via
std::span{batch}.first(count).
- Rename the drop counter to convey it is a count, not the events: member
droppedEvents_ -> numDroppedEvents_ and exchangeDropped() -> takeNumDropped()
(parallels takePending()'s take-and-reset semantics).
- Mark the whole queue interface [[nodiscard]]; the producer call sites that
intentionally ignore the accepted/dropped result now (void)-cast it.
No behavior change.
Signed-off-by: Efraim Eygin <eeygin@nvidia.com>
|
/build |
|
/ok to test f3cd5f9 |
|
🤖 CI Triage Agent — TL;DR: The wheel built fine; the failure is in the "Build sanity image" stage where Full analysisSummary: Jenkins stage 483 "Build sanity image" (aarch64 vLLM image) failed with exit code 100 during Root cause: The Implicated commit: None — not a code regression. (The failure is environmental; commit 3285ae3 / PR #1945 did not touch File: Suggested fix:
Related: none found. Note: |
|
🤖 CI Triage Agent — TL;DR: The build succeeded; the "Allocate DL EP Environment" stage failed because a Slurm Full analysisSummary: Stage 224 "Allocate DL EP Environment" failed on a Slurm resource allocation timeout after waiting ~1 hour for a GB200 node. Root cause: The Implicated commit: unknown — not a code regression; PR #1945 (commit 3285ae3) code compiled and installed cleanly. File: unknown (infrastructure) — the failing command originates from the shared Suggested fix: Retry the build — this is a transient cluster-availability failure, not a code issue. If GB200 ( Related: none found. |
|
🤖 CI Triage Agent — TL;DR: The wheel built fine; the "Build sanity image" stage failed because Full analysisSummary: Stage 470 "Build sanity image" (and its parent stage 332 build_helper_vllm) failed while building Root cause: Implicated commit: None (transient network failure). The failing File: contrib/Dockerfile.vllm:22-25 Suggested fix: Re-run the build; the failure is an external network outage to Related: none |
|
🤖 CI Triage Agent — TL;DR: The wheel build succeeded; the "Build sanity image" stage failed because Full analysisSummary: Jenkins stage 483 "Build sanity image" (aarch64 vLLM sanity image) failed at Root cause: The build node could not reach Implicated commit: None — not a code regression. File: Suggested fix: Re-run the build — this is a transient infrastructure/network issue on the aarch64 agent. If it recurs: (1) restore outbound access to Related: none |
|
🤖 CI Triage Agent — TL;DR: The build compiled and pushed fine; the job failed in "Allocate DL EP Environment" because Full analysisSummary: Root cause: In stage 183/200, the pipeline ran Implicated commit: none — 3285ae3 built and packaged successfully; failure is infrastructure/cluster-side. File: N/A (Jenkins pipeline Slurm allocation step; Suggested fix: Retry the build — this is transient cluster contention on the Related: none found. |
|
🤖 CI Triage Agent — TL;DR: The build/image steps all succeeded; the job failed in the "Allocate DL EP Environment" stage because Full analysisSummary: Jenkins stages "Allocate DL EP Environment" (node 183 and 200) failed after ~3600 s waiting for a GB200 SLURM node that never became available. Root cause: SLURM Implicated commit: none — infrastructure/resource contention, unrelated to commit 3285ae3. File: N/A (Jenkins pipeline SLURM allocation step invoking Suggested fix: Retry the CI job — this is a cluster-availability failure. If GB200 contention on Related: none |
3 participants
What?
Extracts the bounded, mutex-backed telemetry staging queue out of
nixlTelemetryinto a focused internal class,nixlTelemetryStagingQueue(src/core/telemetry/telemetry_staging_queue.{h,cpp}),with no change in behavior.
The new class owns all queue mechanics:
tryPush) and batch (tryPushBatch) insertion,takePending),exchangeDropped).nixlTelemetrykeeps everything metric-specific: per-metric activation,event selection, periodic scheduling, exporter dispatch, and synthetic
dropped-event construction.
updateData(),addXferStats(),flushPendingEvents(), andexportDroppedEvents()are rewired onto the newinterface.
Adds focused and concurrent-producer unit tests under
test/gtest/unit/telemetry/, wired into the existing CTestunittarget.No public API, metric, exporter, or BUFFER schema change.
Why?
nixlTelemetrymixed two independent responsibilities: mapping telemetryoperations to events/exporting them, and implementing a bounded
multi-producer/single-consumer queue. Isolating the queue behind a stable
interface makes the queue mechanics independently testable and provides the
seam for the follow-up low-lock implementation (NIX-1544) to replace the queue
without touching metric logic.
Tracking: NIX-1542 (Perf-3, parent NIX-1359).
How?
Behavior is preserved exactly:
the queue and are not counted as drops.
under one mutex acquisition; a rejected batch increments drops by its length,
and a zero-length batch is a no-op with no lock and no drop-accounting change.
mutex and exports outside the lock; drained order is preserved.
AGENT_TELEMETRY_EVENTS_DROPPEDevent still bypasses the queue.The interface uses pointer + count for batch insertion so it stays C++17-clean
(no
std::spandependency);addXferStats()builds its activated subset in afixed-size stack array and submits the populated prefix with no allocation.
Validated: staging-queue unit tests plus existing core-telemetry, Prometheus,
and DOCA telemetry suites pass on the C++17 and C++20 builds; the extracted
queue (including the concurrent-producer test) passes under ThreadSanitizer and
AddressSanitizer.
Summary by CodeRabbit