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..cba20e4a69 100644 --- a/src/core/telemetry/telemetry.cpp +++ b/src/core/telemetry/telemetry.cpp @@ -14,7 +14,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +#include #include +#include #include #include #include @@ -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 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 +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 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 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}; } + + // 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.h b/src/core/telemetry/telemetry.h index acbbd1b536..0de48e05db 100644 --- a/src/core/telemetry/telemetry.h +++ b/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 -#include -#include #include #include #include @@ -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 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..09d3b2c3d8 --- /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) { + const std::lock_guard 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 events) { + if (events.empty()) { + return true; + } + const std::lock_guard 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 +nixlTelemetryStagingQueue::takePending() { + std::vector pending; + pending.reserve(capacity_); + { + const std::lock_guard 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_; +} diff --git a/src/core/telemetry/telemetry_staging_queue.h b/src/core/telemetry/telemetry_staging_queue.h new file mode 100644 index 0000000000..7feccc6010 --- /dev/null +++ b/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 +#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: + /** + * @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 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 + 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 events_; + std::mutex mutex_; + std::atomic numDroppedEvents_{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..fdcc98a1e2 --- /dev/null +++ b/test/gtest/unit/telemetry/telemetry_staging_queue_test.cpp @@ -0,0 +1,222 @@ +/* + * 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 + +#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.takeNumDropped(), 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.takeNumDropped(), 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.takeNumDropped(), 1u); +} + +TEST(telemetryStagingQueueTest, FittingBatchAppendedCompletelyInOrder) { + nixlTelemetryStagingQueue queue(8); + const std::vector batch = {makeEvent(10), makeEvent(20), makeEvent(30)}; + 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.takeNumDropped(), 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)); + 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)); + EXPECT_EQ(queue.takeNumDropped(), batch.size()); +} + +TEST(telemetryStagingQueueTest, EmptyBatchSucceedsWithoutChangingState) { + nixlTelemetryStagingQueue queue(4); + ASSERT_TRUE(queue.tryPush(makeEvent(7))); + EXPECT_TRUE(queue.tryPushBatch({})); + EXPECT_EQ(queue.takeNumDropped(), 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, 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.takeNumDropped(), 2u); + EXPECT_EQ(queue.takeNumDropped(), 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); + } + (void)queue.tryPushBatch(batch); + } + }); + } + for (auto &t : producers) { + t.join(); + } + done.store(true, std::memory_order_release); + consumer.join(); + + const uint64_t dropped = queue.takeNumDropped(); + 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