From eea0fba4833fdba500c87ec90e7a3047d0c98b3f Mon Sep 17 00:00:00 2001 From: Efraim Eygin Date: Sat, 11 Jul 2026 17:10:59 +0000 Subject: [PATCH 1/4] telemetry: extract bounded staging queue 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 --- src/core/meson.build | 1 + src/core/telemetry/telemetry.cpp | 51 ++-- src/core/telemetry/telemetry.h | 16 +- .../telemetry/telemetry_staging_queue.cpp | 67 ++++++ src/core/telemetry/telemetry_staging_queue.h | 98 ++++++++ test/gtest/unit/meson.build | 3 + test/gtest/unit/telemetry/meson.build | 22 ++ .../telemetry_staging_queue_test.cpp | 221 ++++++++++++++++++ 8 files changed, 441 insertions(+), 38 deletions(-) create mode 100644 src/core/telemetry/telemetry_staging_queue.cpp create mode 100644 src/core/telemetry/telemetry_staging_queue.h create mode 100644 test/gtest/unit/telemetry/meson.build create mode 100644 test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp diff --git a/src/core/meson.build b/src/core/meson.build index ee542bf94d..cfec304973 100644 --- a/src/core/meson.build +++ b/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', diff --git a/src/core/telemetry/telemetry.cpp b/src/core/telemetry/telemetry.cpp index 7dab1b8177..b7bec6ae8a 100644 --- a/src/core/telemetry/telemetry.cpp +++ b/src/core/telemetry/telemetry.cpp @@ -14,6 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +#include #include #include #include @@ -111,12 +112,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 +272,9 @@ nixlTelemetry::startExportTask() { bool nixlTelemetry::flushPendingEvents() { - std::vector next_queue; - next_queue.reserve(maxBufferedEvents_); - { - std::lock_guard 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 +292,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_.exchangeDropped(); if (dropped > 0 && isMetricEnabled(nixl_telemetry_event_type_t::AGENT_TELEMETRY_EVENTS_DROPPED)) { exporter_->exportEvent( @@ -329,13 +324,8 @@ 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 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. + stagingQueue_.tryPush({event_type, value}); } // TODO: the next 4 update* methods may be removable -- addXferStats covers them. @@ -398,24 +388,25 @@ 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 batch; + size_t count = 0; if (post_on) { - events_.emplace_back(nixl_telemetry_event_type_t::AGENT_XFER_POST_TIME, - static_cast(post_time.count())); + batch[count++] = {nixl_telemetry_event_type_t::AGENT_XFER_POST_TIME, + static_cast(post_time.count())}; } if (time_on) { - events_.emplace_back(nixl_telemetry_event_type_t::AGENT_XFER_TIME, - static_cast(xfer_time.count())); + batch[count++] = {nixl_telemetry_event_type_t::AGENT_XFER_TIME, + static_cast(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}; } + + stagingQueue_.tryPushBatch(batch.data(), count); } diff --git a/src/core/telemetry/telemetry.h b/src/core/telemetry/telemetry.h index acbbd1b536..056d122264 100644 --- a/src/core/telemetry/telemetry.h +++ b/src/core/telemetry/telemetry.h @@ -19,6 +19,7 @@ #include "telemetry/telemetry_exporter.h" #include "telemetry_event.h" +#include "telemetry_staging_queue.h" #include "mem_section.h" #include "nixl_types.h" @@ -143,14 +144,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 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 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_; }; diff --git a/src/core/telemetry/telemetry_staging_queue.cpp b/src/core/telemetry/telemetry_staging_queue.cpp new file mode 100644 index 0000000000..e19fbab418 --- /dev/null +++ b/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) { + std::lock_guard lock(mutex_); + if (events_.size() >= capacity_) { + droppedEvents_.fetch_add(1, std::memory_order_relaxed); + return false; + } + events_.push_back(event); + return true; +} + +bool +nixlTelemetryStagingQueue::tryPushBatch(const nixlTelemetryEvent *events, size_t count) { + if (count == 0) { + return true; + } + std::lock_guard lock(mutex_); + if (events_.size() + count > capacity_) { + droppedEvents_.fetch_add(count, std::memory_order_relaxed); + return false; + } + events_.insert(events_.end(), events, events + count); + return true; +} + +std::vector +nixlTelemetryStagingQueue::takePending() { + std::vector pending; + pending.reserve(capacity_); + { + std::lock_guard lock(mutex_); + events_.swap(pending); + } + return pending; +} + +uint64_t +nixlTelemetryStagingQueue::exchangeDropped() noexcept { + return droppedEvents_.exchange(0, std::memory_order_relaxed); +} + +size_t +nixlTelemetryStagingQueue::capacity() const noexcept { + return capacity_; +} diff --git a/src/core/telemetry/telemetry_staging_queue.h b/src/core/telemetry/telemetry_staging_queue.h new file mode 100644 index 0000000000..f2646fa839 --- /dev/null +++ b/src/core/telemetry/telemetry_staging_queue.h @@ -0,0 +1,98 @@ +/* + * 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 +#include +#include +#include +#include + +#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: + 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). + */ + 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 @p count. A zero-length batch succeeds + * without locking or changing drop accounting. + * @param events Pointer to @p count contiguous events. + * @param count Number of events in the batch. + * @return true if the batch was queued (or empty), false if rejected. + */ + bool + tryPushBatch(const nixlTelemetryEvent *events, size_t count); + + /** + * @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. + */ + std::vector + takePending(); + + /** + * @brief Atomically take and reset the staging drops accumulated since the + * previous call. + */ + uint64_t + exchangeDropped() noexcept; + + [[nodiscard]] size_t + capacity() const noexcept; + +private: + const size_t capacity_; + std::vector events_; + std::mutex mutex_; + std::atomic droppedEvents_{0}; +}; + +#endif diff --git a/test/gtest/unit/meson.build b/test/gtest/unit/meson.build index b68a5b96ef..a749e9739a 100644 --- a/test/gtest/unit/meson.build +++ b/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() diff --git a/test/gtest/unit/telemetry/meson.build b/test/gtest/unit/telemetry/meson.build new file mode 100644 index 0000000000..6b2cdb162e --- /dev/null +++ b/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], +) diff --git a/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp b/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp new file mode 100644 index 0000000000..25d685129c --- /dev/null +++ b/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp @@ -0,0 +1,221 @@ +/* + * 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 + +#include +#include +#include +#include +#include +#include + +#include "telemetry/telemetry_staging_queue.h" + +namespace { + +constexpr auto kType = nixl_telemetry_event_type_t::AGENT_TX_BYTES; + +nixlTelemetryEvent +makeEvent(uint64_t value) { + return {kType, value}; +} + +TEST(telemetryStagingQueueTest, ConstructorRetainsCapacityEmptyNoDrops) { + nixlTelemetryStagingQueue queue(8); + EXPECT_EQ(queue.capacity(), 8u); + EXPECT_TRUE(queue.takePending().empty()); + EXPECT_EQ(queue.exchangeDropped(), 0u); +} + +TEST(telemetryStagingQueueTest, SinglePushPreservesTypeAndValue) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush({nixl_telemetry_event_type_t::AGENT_XFER_TIME, 42})); + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 1u); + EXPECT_EQ(pending[0].eventType_, nixl_telemetry_event_type_t::AGENT_XFER_TIME); + EXPECT_EQ(pending[0].value_, 42u); +} + +TEST(telemetryStagingQueueTest, ExactlyCapacitySinglePushesSucceed) { + nixlTelemetryStagingQueue queue(3); + for (uint64_t i = 0; i < 3; ++i) { + EXPECT_TRUE(queue.tryPush(makeEvent(i))); + } + EXPECT_EQ(queue.exchangeDropped(), 0u); +} + +TEST(telemetryStagingQueueTest, PushBeyondCapacityFailsAndDropsByOne) { + nixlTelemetryStagingQueue queue(2); + ASSERT_TRUE(queue.tryPush(makeEvent(0))); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + EXPECT_FALSE(queue.tryPush(makeEvent(2))); + EXPECT_EQ(queue.exchangeDropped(), 1u); +} + +TEST(telemetryStagingQueueTest, FittingBatchAppendedCompletelyInOrder) { + nixlTelemetryStagingQueue queue(8); + const std::vector batch = {makeEvent(10), makeEvent(20), makeEvent(30)}; + ASSERT_TRUE(queue.tryPushBatch(batch.data(), batch.size())); + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 3u); + EXPECT_EQ(pending[0].value_, 10u); + EXPECT_EQ(pending[1].value_, 20u); + EXPECT_EQ(pending[2].value_, 30u); + EXPECT_EQ(queue.exchangeDropped(), 0u); +} + +TEST(telemetryStagingQueueTest, NonFittingBatchRejectedWithoutPartialInsertion) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + ASSERT_TRUE(queue.tryPush(makeEvent(2))); + const std::vector batch = {makeEvent(3), makeEvent(4), makeEvent(5)}; + EXPECT_FALSE(queue.tryPushBatch(batch.data(), batch.size())); + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 2u); + EXPECT_EQ(pending[0].value_, 1u); + EXPECT_EQ(pending[1].value_, 2u); +} + +TEST(telemetryStagingQueueTest, RejectedBatchDropCountEqualsBatchLength) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + ASSERT_TRUE(queue.tryPush(makeEvent(2))); + const std::vector batch = {makeEvent(3), makeEvent(4), makeEvent(5)}; + EXPECT_FALSE(queue.tryPushBatch(batch.data(), batch.size())); + EXPECT_EQ(queue.exchangeDropped(), batch.size()); +} + +TEST(telemetryStagingQueueTest, EmptyBatchSucceedsWithoutChangingState) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush(makeEvent(7))); + EXPECT_TRUE(queue.tryPushBatch(nullptr, 0)); + EXPECT_EQ(queue.exchangeDropped(), 0u); + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 1u); + EXPECT_EQ(pending[0].value_, 7u); +} + +TEST(telemetryStagingQueueTest, TakePendingReturnsAllInOrderAndEmpties) { + nixlTelemetryStagingQueue queue(4); + for (uint64_t i = 0; i < 3; ++i) { + ASSERT_TRUE(queue.tryPush(makeEvent(i))); + } + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 3u); + for (uint64_t i = 0; i < 3; ++i) { + EXPECT_EQ(pending[i].value_, i); + } + EXPECT_TRUE(queue.takePending().empty()); +} + +TEST(telemetryStagingQueueTest, SecondDrainIsEmpty) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + EXPECT_EQ(queue.takePending().size(), 1u); + EXPECT_TRUE(queue.takePending().empty()); +} + +TEST(telemetryStagingQueueTest, AcceptsEventsAfterDrain) { + nixlTelemetryStagingQueue queue(2); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + ASSERT_TRUE(queue.tryPush(makeEvent(2))); + (void)queue.takePending(); + ASSERT_TRUE(queue.tryPush(makeEvent(3))); + ASSERT_TRUE(queue.tryPush(makeEvent(4))); + auto pending = queue.takePending(); + ASSERT_EQ(pending.size(), 2u); + EXPECT_EQ(pending[0].value_, 3u); + EXPECT_EQ(pending[1].value_, 4u); +} + +TEST(telemetryStagingQueueTest, ExchangeDroppedReturnsDeltaOnceThenZero) { + nixlTelemetryStagingQueue queue(1); + ASSERT_TRUE(queue.tryPush(makeEvent(1))); + EXPECT_FALSE(queue.tryPush(makeEvent(2))); + EXPECT_FALSE(queue.tryPush(makeEvent(3))); + EXPECT_EQ(queue.exchangeDropped(), 2u); + EXPECT_EQ(queue.exchangeDropped(), 0u); +} + +TEST(telemetryStagingQueueTest, ConcurrentProducersConserveEventSlots) { + constexpr size_t kProducers = 8; + constexpr size_t kEventsPerProducer = 5000; + constexpr size_t kBatchSize = 4; + constexpr size_t kProduced = kProducers * kEventsPerProducer * kBatchSize; + // Capacity below total production so both acceptance and drops occur, while + // a concurrent consumer drains to keep the queue from staying saturated. + nixlTelemetryStagingQueue queue(256); + + std::atomic done{false}; + std::vector drained; + std::thread consumer([&]() { + while (!done.load(std::memory_order_acquire)) { + auto pending = queue.takePending(); + drained.insert(drained.end(), pending.begin(), pending.end()); + std::this_thread::yield(); + } + auto pending = queue.takePending(); + drained.insert(drained.end(), pending.begin(), pending.end()); + }); + + std::vector producers; + producers.reserve(kProducers); + for (size_t p = 0; p < kProducers; ++p) { + producers.emplace_back([&, p]() { + for (size_t i = 0; i < kEventsPerProducer; ++i) { + const uint64_t base = (p * kEventsPerProducer + i) * kBatchSize; + std::array batch; + for (size_t b = 0; b < kBatchSize; ++b) { + batch[b] = makeEvent(base + b); + } + queue.tryPushBatch(batch.data(), batch.size()); + } + }); + } + for (auto &t : producers) { + t.join(); + } + done.store(true, std::memory_order_release); + consumer.join(); + + const uint64_t dropped = queue.exchangeDropped(); + EXPECT_EQ(drained.size() + dropped, kProduced); + + // Every accepted identifier appears exactly once, and acceptance is + // batch-atomic: each fully accepted batch contributes all kBatchSize members. + std::unordered_map counts; + counts.reserve(drained.size()); + for (const auto &event : drained) { + ++counts[event.value_]; + } + for (const auto &[value, seen] : counts) { + EXPECT_EQ(seen, 1u) << "duplicate identifier " << value; + } + + std::unordered_map batch_members; + batch_members.reserve(drained.size()); + for (const auto &event : drained) { + ++batch_members[event.value_ / kBatchSize]; + } + for (const auto &[batch_id, members] : batch_members) { + EXPECT_EQ(members, kBatchSize) + << "partial batch " << batch_id << " with " << members << " members"; + } +} + +} // namespace From a9db285e919e44e0630e1faf8f794ba2d2a89fc2 Mon Sep 17 00:00:00 2001 From: Efraim Eygin Date: Wed, 15 Jul 2026 09:35:45 +0000 Subject: [PATCH 2/4] telemetry: address CodeRabbit review on staging queue 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 --- src/core/telemetry/telemetry_staging_queue.cpp | 6 +++--- src/core/telemetry/telemetry_staging_queue.h | 9 +++++++++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/src/core/telemetry/telemetry_staging_queue.cpp b/src/core/telemetry/telemetry_staging_queue.cpp index e19fbab418..548db8a29c 100644 --- a/src/core/telemetry/telemetry_staging_queue.cpp +++ b/src/core/telemetry/telemetry_staging_queue.cpp @@ -22,7 +22,7 @@ nixlTelemetryStagingQueue::nixlTelemetryStagingQueue(size_t capacity) : capacity bool nixlTelemetryStagingQueue::tryPush(const nixlTelemetryEvent &event) { - std::lock_guard lock(mutex_); + const std::lock_guard lock(mutex_); if (events_.size() >= capacity_) { droppedEvents_.fetch_add(1, std::memory_order_relaxed); return false; @@ -36,7 +36,7 @@ nixlTelemetryStagingQueue::tryPushBatch(const nixlTelemetryEvent *events, size_t if (count == 0) { return true; } - std::lock_guard lock(mutex_); + const std::lock_guard lock(mutex_); if (events_.size() + count > capacity_) { droppedEvents_.fetch_add(count, std::memory_order_relaxed); return false; @@ -50,7 +50,7 @@ nixlTelemetryStagingQueue::takePending() { std::vector pending; pending.reserve(capacity_); { - std::lock_guard lock(mutex_); + const std::lock_guard lock(mutex_); events_.swap(pending); } return pending; diff --git a/src/core/telemetry/telemetry_staging_queue.h b/src/core/telemetry/telemetry_staging_queue.h index f2646fa839..3435582a70 100644 --- a/src/core/telemetry/telemetry_staging_queue.h +++ b/src/core/telemetry/telemetry_staging_queue.h @@ -41,6 +41,15 @@ */ 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); /** From b923ec4fc4277cd1fc4d18978261b15937007471 Mon Sep 17 00:00:00 2001 From: Efraim Eygin Date: Wed, 15 Jul 2026 11:28:18 +0000 Subject: [PATCH 3/4] telemetry: address review on staging queue - 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 and includes; neither std::vector nor std::mutex is referenced there after the extraction. No behavior change. Signed-off-by: Efraim Eygin --- src/core/telemetry/telemetry.h | 2 -- src/core/telemetry/telemetry_staging_queue.cpp | 2 +- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/src/core/telemetry/telemetry.h b/src/core/telemetry/telemetry.h index 056d122264..0de48e05db 100644 --- a/src/core/telemetry/telemetry.h +++ b/src/core/telemetry/telemetry.h @@ -24,8 +24,6 @@ #include "nixl_types.h" #include -#include -#include #include #include #include diff --git a/src/core/telemetry/telemetry_staging_queue.cpp b/src/core/telemetry/telemetry_staging_queue.cpp index 548db8a29c..24e7f2fdd1 100644 --- a/src/core/telemetry/telemetry_staging_queue.cpp +++ b/src/core/telemetry/telemetry_staging_queue.cpp @@ -37,7 +37,7 @@ nixlTelemetryStagingQueue::tryPushBatch(const nixlTelemetryEvent *events, size_t return true; } const std::lock_guard lock(mutex_); - if (events_.size() + count > capacity_) { + if (count > capacity_ - events_.size()) { droppedEvents_.fetch_add(count, std::memory_order_relaxed); return false; } From f3cd5f9684b98fbcca0c518c3fa2912c85930934 Mon Sep 17 00:00:00 2001 From: Efraim Eygin Date: Wed, 15 Jul 2026 13:17:45 +0000 Subject: [PATCH 4/4] telemetry: refine staging queue API per review - tryPushBatch now takes std::span 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 --- src/core/telemetry/telemetry.cpp | 12 ++++--- .../telemetry/telemetry_staging_queue.cpp | 16 +++++----- src/core/telemetry/telemetry_staging_queue.h | 26 ++++++++-------- .../telemetry_staging_queue_test.cpp | 31 ++++++++++--------- 4 files changed, 45 insertions(+), 40 deletions(-) diff --git a/src/core/telemetry/telemetry.cpp b/src/core/telemetry/telemetry.cpp index b7bec6ae8a..cba20e4a69 100644 --- a/src/core/telemetry/telemetry.cpp +++ b/src/core/telemetry/telemetry.cpp @@ -16,6 +16,7 @@ */ #include #include +#include #include #include #include @@ -292,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 = stagingQueue_.exchangeDropped(); + const uint64_t dropped = stagingQueue_.takeNumDropped(); if (dropped > 0 && isMetricEnabled(nixl_telemetry_event_type_t::AGENT_TELEMETRY_EVENTS_DROPPED)) { exporter_->exportEvent( @@ -324,8 +325,9 @@ nixlTelemetry::updateData(nixl_telemetry_event_type_t event_type, uint64_t value if (!isMetricEnabled(event_type)) { return; } - // agent can be multi-threaded; the queue performs its own drop accounting. - stagingQueue_.tryPush({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. @@ -408,5 +410,7 @@ nixlTelemetry::addXferStats(std::chrono::microseconds xfer_time, batch[count++] = {requests_type, 1}; } - stagingQueue_.tryPushBatch(batch.data(), count); + // 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)); } diff --git a/src/core/telemetry/telemetry_staging_queue.cpp b/src/core/telemetry/telemetry_staging_queue.cpp index 24e7f2fdd1..09d3b2c3d8 100644 --- a/src/core/telemetry/telemetry_staging_queue.cpp +++ b/src/core/telemetry/telemetry_staging_queue.cpp @@ -24,7 +24,7 @@ bool nixlTelemetryStagingQueue::tryPush(const nixlTelemetryEvent &event) { const std::lock_guard lock(mutex_); if (events_.size() >= capacity_) { - droppedEvents_.fetch_add(1, std::memory_order_relaxed); + numDroppedEvents_.fetch_add(1, std::memory_order_relaxed); return false; } events_.push_back(event); @@ -32,16 +32,16 @@ nixlTelemetryStagingQueue::tryPush(const nixlTelemetryEvent &event) { } bool -nixlTelemetryStagingQueue::tryPushBatch(const nixlTelemetryEvent *events, size_t count) { - if (count == 0) { +nixlTelemetryStagingQueue::tryPushBatch(std::span events) { + if (events.empty()) { return true; } const std::lock_guard lock(mutex_); - if (count > capacity_ - events_.size()) { - droppedEvents_.fetch_add(count, std::memory_order_relaxed); + if (events.size() > capacity_ - events_.size()) { + numDroppedEvents_.fetch_add(events.size(), std::memory_order_relaxed); return false; } - events_.insert(events_.end(), events, events + count); + events_.insert(events_.end(), events.begin(), events.end()); return true; } @@ -57,8 +57,8 @@ nixlTelemetryStagingQueue::takePending() { } uint64_t -nixlTelemetryStagingQueue::exchangeDropped() noexcept { - return droppedEvents_.exchange(0, std::memory_order_relaxed); +nixlTelemetryStagingQueue::takeNumDropped() noexcept { + return numDroppedEvents_.exchange(0, std::memory_order_relaxed); } size_t diff --git a/src/core/telemetry/telemetry_staging_queue.h b/src/core/telemetry/telemetry_staging_queue.h index 3435582a70..7feccc6010 100644 --- a/src/core/telemetry/telemetry_staging_queue.h +++ b/src/core/telemetry/telemetry_staging_queue.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include "telemetry_event.h" @@ -60,7 +61,7 @@ class nixlTelemetryStagingQueue { * by one. * @return true if the event was queued, false if rejected (dropped). */ - bool + [[nodiscard]] bool tryPush(const nixlTelemetryEvent &event); /** @@ -68,14 +69,13 @@ class nixlTelemetryStagingQueue { * * 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 @p count. A zero-length batch succeeds - * without locking or changing drop accounting. - * @param events Pointer to @p count contiguous events. - * @param count Number of events in the batch. + * 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. */ - bool - tryPushBatch(const nixlTelemetryEvent *events, size_t count); + [[nodiscard]] bool + tryPushBatch(std::span events); /** * @brief Swap the live queue with an empty capacity-reserved vector and @@ -84,15 +84,15 @@ class nixlTelemetryStagingQueue { * The swap happens under the mutex; producers can write to the fresh live * vector while the consumer processes the returned one. */ - std::vector + [[nodiscard]] std::vector takePending(); /** - * @brief Atomically take and reset the staging drops accumulated since the - * previous call. + * @brief Atomically take and reset the number of staging drops accumulated + * since the previous call. */ - uint64_t - exchangeDropped() noexcept; + [[nodiscard]] uint64_t + takeNumDropped() noexcept; [[nodiscard]] size_t capacity() const noexcept; @@ -101,7 +101,7 @@ class nixlTelemetryStagingQueue { const size_t capacity_; std::vector events_; std::mutex mutex_; - std::atomic droppedEvents_{0}; + std::atomic numDroppedEvents_{0}; }; #endif diff --git a/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp b/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp index 25d685129c..fdcc98a1e2 100644 --- a/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp +++ b/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -39,7 +40,7 @@ TEST(telemetryStagingQueueTest, ConstructorRetainsCapacityEmptyNoDrops) { nixlTelemetryStagingQueue queue(8); EXPECT_EQ(queue.capacity(), 8u); EXPECT_TRUE(queue.takePending().empty()); - EXPECT_EQ(queue.exchangeDropped(), 0u); + EXPECT_EQ(queue.takeNumDropped(), 0u); } TEST(telemetryStagingQueueTest, SinglePushPreservesTypeAndValue) { @@ -56,7 +57,7 @@ TEST(telemetryStagingQueueTest, ExactlyCapacitySinglePushesSucceed) { for (uint64_t i = 0; i < 3; ++i) { EXPECT_TRUE(queue.tryPush(makeEvent(i))); } - EXPECT_EQ(queue.exchangeDropped(), 0u); + EXPECT_EQ(queue.takeNumDropped(), 0u); } TEST(telemetryStagingQueueTest, PushBeyondCapacityFailsAndDropsByOne) { @@ -64,19 +65,19 @@ TEST(telemetryStagingQueueTest, PushBeyondCapacityFailsAndDropsByOne) { ASSERT_TRUE(queue.tryPush(makeEvent(0))); ASSERT_TRUE(queue.tryPush(makeEvent(1))); EXPECT_FALSE(queue.tryPush(makeEvent(2))); - EXPECT_EQ(queue.exchangeDropped(), 1u); + EXPECT_EQ(queue.takeNumDropped(), 1u); } TEST(telemetryStagingQueueTest, FittingBatchAppendedCompletelyInOrder) { nixlTelemetryStagingQueue queue(8); const std::vector batch = {makeEvent(10), makeEvent(20), makeEvent(30)}; - ASSERT_TRUE(queue.tryPushBatch(batch.data(), batch.size())); + ASSERT_TRUE(queue.tryPushBatch(batch)); auto pending = queue.takePending(); ASSERT_EQ(pending.size(), 3u); EXPECT_EQ(pending[0].value_, 10u); EXPECT_EQ(pending[1].value_, 20u); EXPECT_EQ(pending[2].value_, 30u); - EXPECT_EQ(queue.exchangeDropped(), 0u); + EXPECT_EQ(queue.takeNumDropped(), 0u); } TEST(telemetryStagingQueueTest, NonFittingBatchRejectedWithoutPartialInsertion) { @@ -84,7 +85,7 @@ TEST(telemetryStagingQueueTest, NonFittingBatchRejectedWithoutPartialInsertion) ASSERT_TRUE(queue.tryPush(makeEvent(1))); ASSERT_TRUE(queue.tryPush(makeEvent(2))); const std::vector batch = {makeEvent(3), makeEvent(4), makeEvent(5)}; - EXPECT_FALSE(queue.tryPushBatch(batch.data(), batch.size())); + EXPECT_FALSE(queue.tryPushBatch(batch)); auto pending = queue.takePending(); ASSERT_EQ(pending.size(), 2u); EXPECT_EQ(pending[0].value_, 1u); @@ -96,15 +97,15 @@ TEST(telemetryStagingQueueTest, RejectedBatchDropCountEqualsBatchLength) { ASSERT_TRUE(queue.tryPush(makeEvent(1))); ASSERT_TRUE(queue.tryPush(makeEvent(2))); const std::vector batch = {makeEvent(3), makeEvent(4), makeEvent(5)}; - EXPECT_FALSE(queue.tryPushBatch(batch.data(), batch.size())); - EXPECT_EQ(queue.exchangeDropped(), batch.size()); + EXPECT_FALSE(queue.tryPushBatch(batch)); + EXPECT_EQ(queue.takeNumDropped(), batch.size()); } TEST(telemetryStagingQueueTest, EmptyBatchSucceedsWithoutChangingState) { nixlTelemetryStagingQueue queue(4); ASSERT_TRUE(queue.tryPush(makeEvent(7))); - EXPECT_TRUE(queue.tryPushBatch(nullptr, 0)); - EXPECT_EQ(queue.exchangeDropped(), 0u); + EXPECT_TRUE(queue.tryPushBatch({})); + EXPECT_EQ(queue.takeNumDropped(), 0u); auto pending = queue.takePending(); ASSERT_EQ(pending.size(), 1u); EXPECT_EQ(pending[0].value_, 7u); @@ -143,13 +144,13 @@ TEST(telemetryStagingQueueTest, AcceptsEventsAfterDrain) { EXPECT_EQ(pending[1].value_, 4u); } -TEST(telemetryStagingQueueTest, ExchangeDroppedReturnsDeltaOnceThenZero) { +TEST(telemetryStagingQueueTest, TakeNumDroppedReturnsDeltaOnceThenZero) { nixlTelemetryStagingQueue queue(1); ASSERT_TRUE(queue.tryPush(makeEvent(1))); EXPECT_FALSE(queue.tryPush(makeEvent(2))); EXPECT_FALSE(queue.tryPush(makeEvent(3))); - EXPECT_EQ(queue.exchangeDropped(), 2u); - EXPECT_EQ(queue.exchangeDropped(), 0u); + EXPECT_EQ(queue.takeNumDropped(), 2u); + EXPECT_EQ(queue.takeNumDropped(), 0u); } TEST(telemetryStagingQueueTest, ConcurrentProducersConserveEventSlots) { @@ -183,7 +184,7 @@ TEST(telemetryStagingQueueTest, ConcurrentProducersConserveEventSlots) { for (size_t b = 0; b < kBatchSize; ++b) { batch[b] = makeEvent(base + b); } - queue.tryPushBatch(batch.data(), batch.size()); + (void)queue.tryPushBatch(batch); } }); } @@ -193,7 +194,7 @@ TEST(telemetryStagingQueueTest, ConcurrentProducersConserveEventSlots) { done.store(true, std::memory_order_release); consumer.join(); - const uint64_t dropped = queue.exchangeDropped(); + const uint64_t dropped = queue.takeNumDropped(); EXPECT_EQ(drained.size() + dropped, kProduced); // Every accepted identifier appears exactly once, and acceptance is