telemetry: extract bounded staging queue

src/core/meson.build

@@ -67,6 +67,7 @@ nixl_lib_sources = [
     'nixl_plugin_manager.cpp',
     'nixl_listener.cpp',
     'telemetry/telemetry.cpp',
+    'telemetry/telemetry_staging_queue.cpp',
     'telemetry/buffer_exporter.cpp',
     'telemetry/buffer_plugin.cpp',
     'telemetry/nop_plugin.cpp',

src/core/telemetry/telemetry.cpp

@@ -14,7 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+#include <array>
 #include <chrono>
+#include <span>
 #include <sstream>
 #include <thread>
 #include <filesystem>
@@ -111,12 +113,11 @@ nixlTelemetry::nixlTelemetry(const std::string &agent_name, const std::string &e
       maxBufferedEvents_(resolveMaxBufferedEvents()),
       exporter_(makeExporter(exporter_name)),
       metricEnabled_(resolveEnabledMetrics()),
+      stagingQueue_(maxBufferedEvents_),
       pool_(1),
       writeTask_(pool_.get_executor(), DEFAULT_TELEMETRY_RUN_INTERVAL, false) {
     // A constructed nixlTelemetry always has an exporter (makeExporter throws
-    // otherwise), so reserve the event buffer up front and start the periodic
-    // export task.
-    events_.reserve(maxBufferedEvents_);
+    // otherwise), so start the periodic export task.
     startExportTask();
 
     if (!nixlTime::fastClockUsesHwCounter()) {
@@ -272,14 +273,9 @@ nixlTelemetry::startExportTask() {
 
 bool
 nixlTelemetry::flushPendingEvents() {
-    std::vector<nixlTelemetryEvent> next_queue;
-    next_queue.reserve(maxBufferedEvents_);
-    {
-        std::lock_guard<std::mutex> lock(mutex_);
-        events_.swap(next_queue);
-    }
+    auto pending = stagingQueue_.takePending();
 
-    for (auto &event : next_queue) {
+    for (auto &event : pending) {
         // Deactivated metrics never enter the queue (filtered at the source), so
         // everything here is already exportable.
         exporter_->exportEvent(event);
@@ -297,7 +293,7 @@ nixlTelemetry::exportDroppedEvents() {
     // staging queue, it can never itself be staging-dropped; emitting only a
     // positive count keeps the no-overflow path event-free (preserving
     // exact-count contracts).
-    const uint64_t dropped = droppedEvents_.exchange(0, std::memory_order_relaxed);
+    const uint64_t dropped = stagingQueue_.takeNumDropped();
     if (dropped > 0 &&
         isMetricEnabled(nixl_telemetry_event_type_t::AGENT_TELEMETRY_EVENTS_DROPPED)) {
         exporter_->exportEvent(
@@ -329,13 +325,9 @@ nixlTelemetry::updateData(nixl_telemetry_event_type_t event_type, uint64_t value
     if (!isMetricEnabled(event_type)) {
         return;
     }
-    // agent can be multi-threaded
-    std::lock_guard<std::mutex> lock(mutex_);
-    if (events_.size() >= maxBufferedEvents_) {
-        droppedEvents_.fetch_add(1, std::memory_order_relaxed);
-        return;
-    }
-    events_.emplace_back(event_type, value);
+    // agent can be multi-threaded; the queue performs its own drop accounting,
+    // so the accepted/dropped result is intentionally not consumed here.
+    (void)stagingQueue_.tryPush({event_type, value});
 }
 
 // TODO: the next 4 update* methods may be removable -- addXferStats covers them.
@@ -398,24 +390,27 @@ nixlTelemetry::addXferStats(std::chrono::microseconds xfer_time,
         return;
     }
 
-    // Atomic per-transfer batch: all activated events land together or none do.
-    const std::lock_guard lock(mutex_);
-    if (events_.size() + batch_size > maxBufferedEvents_) {
-        droppedEvents_.fetch_add(batch_size, std::memory_order_relaxed);
-        return;
-    }
+    // Build the activated subset in a fixed-size stack array (post time, transfer
+    // time, bytes, request count) and submit it as one all-or-none batch; the
+    // queue accepts every event or drops the whole batch.
+    std::array<nixlTelemetryEvent, 4> batch;
+    size_t count = 0;
     if (post_on) {
-        events_.emplace_back(nixl_telemetry_event_type_t::AGENT_XFER_POST_TIME,
-                             static_cast<uint64_t>(post_time.count()));
+        batch[count++] = {nixl_telemetry_event_type_t::AGENT_XFER_POST_TIME,
+                          static_cast<uint64_t>(post_time.count())};
     }
     if (time_on) {
-        events_.emplace_back(nixl_telemetry_event_type_t::AGENT_XFER_TIME,
-                             static_cast<uint64_t>(xfer_time.count()));
+        batch[count++] = {nixl_telemetry_event_type_t::AGENT_XFER_TIME,
+                          static_cast<uint64_t>(xfer_time.count())};
     }
     if (bytes_on) {
-        events_.emplace_back(bytes_type, bytes);
+        batch[count++] = {bytes_type, bytes};
     }
     if (requests_on) {
-        events_.emplace_back(requests_type, 1);
+        batch[count++] = {requests_type, 1};
     }
+
+    // The queue performs its own drop accounting, so the accepted/dropped
+    // result is intentionally not consumed here.
+    (void)stagingQueue_.tryPushBatch(std::span{batch}.first(count));
 }

src/core/telemetry/telemetry.h

@@ -19,12 +19,11 @@
 
 #include "telemetry/telemetry_exporter.h"
 #include "telemetry_event.h"
+#include "telemetry_staging_queue.h"
 #include "mem_section.h"
 #include "nixl_types.h"
 
 #include <string>
-#include <vector>
-#include <mutex>
 #include <memory>
 #include <chrono>
 #include <functional>
@@ -143,14 +142,13 @@ class nixlTelemetry {
     // the source before it enters the staging queue. All-true when the variable
     // is unset (backward compatible).
     const nixl_telemetry_metric_mask_t metricEnabled_;
-    std::vector<nixlTelemetryEvent> events_;
-    std::mutex mutex_;
-    // Producer-side staging-queue drop counter: incremented from any thread when
-    // updateData / addXferStats cannot append because events_ is full, so the
-    // event never reaches any exporter. Does not track BUFFER cyclic-ring loss
-    // (a separate, downstream condition). Each flush takes and resets it
-    // (exchange) and publishes the count as an AGENT_TELEMETRY_EVENTS_DROPPED event.
-    std::atomic<uint64_t> droppedEvents_{0};
+    // Bounded producer/consumer staging queue: owns event storage, the capacity
+    // reserve, the producer mutex, capacity enforcement, single/batch insertion,
+    // the swap-drain, and the staging-drop counter. Its drops do not track BUFFER
+    // cyclic-ring loss (a separate, downstream condition); each flush takes and
+    // resets the drop count and publishes it as an AGENT_TELEMETRY_EVENTS_DROPPED
+    // event.
+    nixlTelemetryStagingQueue stagingQueue_;
     asio::thread_pool pool_;
     periodicTask writeTask_;
 };

src/core/telemetry/telemetry_staging_queue.cpp

@@ -0,0 +1,67 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+#include "telemetry_staging_queue.h"
+
+nixlTelemetryStagingQueue::nixlTelemetryStagingQueue(size_t capacity) : capacity_(capacity) {
+    events_.reserve(capacity_);
+}
+
+bool
+nixlTelemetryStagingQueue::tryPush(const nixlTelemetryEvent &event) {
+    const std::lock_guard<std::mutex> lock(mutex_);
+    if (events_.size() >= capacity_) {
+        numDroppedEvents_.fetch_add(1, std::memory_order_relaxed);
+        return false;
+    }
+    events_.push_back(event);
+    return true;
+}
+
+bool
+nixlTelemetryStagingQueue::tryPushBatch(std::span<const nixlTelemetryEvent> events) {
+    if (events.empty()) {
+        return true;
+    }
+    const std::lock_guard<std::mutex> lock(mutex_);
+    if (events.size() > capacity_ - events_.size()) {
+        numDroppedEvents_.fetch_add(events.size(), std::memory_order_relaxed);
+        return false;
+    }
+    events_.insert(events_.end(), events.begin(), events.end());
+    return true;
+}
+
+std::vector<nixlTelemetryEvent>
+nixlTelemetryStagingQueue::takePending() {
+    std::vector<nixlTelemetryEvent> pending;
+    pending.reserve(capacity_);
+    {
+        const std::lock_guard<std::mutex> lock(mutex_);
+        events_.swap(pending);
+    }
+    return pending;
+}
+
+uint64_t
+nixlTelemetryStagingQueue::takeNumDropped() noexcept {
+    return numDroppedEvents_.exchange(0, std::memory_order_relaxed);
+}
+
+size_t
+nixlTelemetryStagingQueue::capacity() const noexcept {
+    return capacity_;
+}

src/core/telemetry/telemetry_staging_queue.h

@@ -0,0 +1,107 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * 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.
+ */
+#ifndef NIXL_SRC_CORE_TELEMETRY_TELEMETRY_STAGING_QUEUE_H
+#define NIXL_SRC_CORE_TELEMETRY_TELEMETRY_STAGING_QUEUE_H
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <mutex>
+#include <span>
+#include <vector>
+
+#include "telemetry_event.h"
+
+/**
+ * @brief Bounded multi-producer/single-consumer staging queue for telemetry
+ *        events.
+ *
+ * Owns the queue mechanics that used to live inline in nixlTelemetry: the event
+ * storage and its capacity reserve, the producer/consumer mutex, capacity
+ * enforcement, all-or-none single and batch insertion, the swap-drain, and the
+ * staging-drop counter. Metric semantics (activation, event selection, exporter
+ * dispatch, synthetic dropped-event construction) stay in nixlTelemetry.
+ *
+ * Drop policy is drop-newest: once the queue is full an incoming event or batch
+ * is rejected and counted as dropped; already-queued events are never
+ * overwritten.
+ */
+class nixlTelemetryStagingQueue {
+public:
+    /**
+     * @brief Construct a queue that retains at most @p capacity events.
+     *
+     * Reserves storage for @p capacity events up front so the producer path
+     * never reallocates. Any @p capacity is accepted (a zero capacity yields a
+     * queue that drops every push); rejecting a zero telemetry buffer size is
+     * the caller's responsibility (see nixlTelemetry construction).
+     * @param capacity Maximum number of events retained before pushes are dropped.
+     */
+    explicit nixlTelemetryStagingQueue(size_t capacity);
+
+    /**
+     * @brief Append a single event if there is room.
+     *
+     * The capacity check and insertion happen under one mutex acquisition. On a
+     * full queue the event is rejected and the staging-drop count is incremented
+     * by one.
+     * @return true if the event was queued, false if rejected (dropped).
+     */
+    [[nodiscard]] bool
+    tryPush(const nixlTelemetryEvent &event);
+
+    /**
+     * @brief Append a batch of events all-or-none if the whole batch fits.
+     *
+     * The capacity check and insertion happen under one mutex acquisition. The
+     * batch is accepted completely or rejected completely; on rejection the
+     * staging-drop count is incremented by the batch size. An empty batch
+     * succeeds without locking or changing drop accounting.
+     * @param events Contiguous events to append.
+     * @return true if the batch was queued (or empty), false if rejected.
+     */
+    [[nodiscard]] bool
+    tryPushBatch(std::span<const nixlTelemetryEvent> events);
+
+    /**
+     * @brief Swap the live queue with an empty capacity-reserved vector and
+     *        return the drained events in insertion order.
+     *
+     * The swap happens under the mutex; producers can write to the fresh live
+     * vector while the consumer processes the returned one.
+     */
+    [[nodiscard]] std::vector<nixlTelemetryEvent>
+    takePending();
+
+    /**
+     * @brief Atomically take and reset the number of staging drops accumulated
+     *        since the previous call.
+     */
+    [[nodiscard]] uint64_t
+    takeNumDropped() noexcept;
+
+    [[nodiscard]] size_t
+    capacity() const noexcept;
+
+private:
+    const size_t capacity_;
+    std::vector<nixlTelemetryEvent> events_;
+    std::mutex mutex_;
+    std::atomic<uint64_t> numDroppedEvents_{0};
+};
+
+#endif

test/gtest/unit/meson.build

@@ -35,6 +35,9 @@ unit_test_deps += [util_unit_test_dep]
 subdir('tracing')
 unit_test_deps += [tracing_unit_test_dep]
 
+subdir('telemetry')
+unit_test_deps += [telemetry_unit_test_dep]
+
 cpp = meson.get_compiler('cpp')
 azure_storage_blobs = cpp.find_library('azure-storage-blobs', required: false)
 if enabled_plugins.get('AZURE_BLOB') and azure_storage_blobs.found()

test/gtest/unit/telemetry/meson.build

@@ -0,0 +1,22 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# 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.
+
+telemetry_unit_test_dep = declare_dependency(
+    sources: [
+        'telemetry_staging_queue_test.cpp',
+    ],
+    include_directories: [nixl_inc_dirs, gtest_inc_dirs],
+    dependencies: [gmock_dep, nixl_common_dep],
+)