Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/BackendGuide.md
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ Note that plugin implementation is in C++ to achieve SOL, and the header files f

NIXL implements a modular plugin architecture where each backend is encapsulated within a library that exposes its features through the SB API, such as [UCX](https://github.com/ai-dynamo/nixl/tree/main/src/plugins/ucx), [GPUDirect Storage (GDS)](https://github.com/ai-dynamo/nixl/tree/main/src/plugins/cuda_gds), or any other custom implementation. The Plugin Manager component within NIXL handles discovery, loading, and instantiation of backend plugins, whether they are dynamically loaded or statically built into the NIXL library. Each plugin needs to implement the SB API, as well as a few methods for the plugin manager, which are discussed in the following sections.

Note that some of the methods in the SB API are not necessary to be implemented, for instance if a backend does not support notifications, it can indicate it through supportsNotif() method, and if it returns false, the agent will not send requests with notifications to this backend. There are 4 such capability indicators, which are detailed further and which APIs are required to be implemented for each of them.
Note that some of the methods in the SB API are not necessary to be implemented, for instance if a backend does not support notifications, it can indicate it through supportsNotif() method, and if it returns false, the agent will not send requests with notifications to this backend. There are 5 such capability indicators, which are detailed further and which APIs are required to be implemented for each of them.

![NIXL SB API](figures/nixl_sb_api.png)

Expand All @@ -52,6 +52,7 @@ The key/value parameters are a map of strings to byte arrays that are passed fro
* supportsRemote(): Indicates if the backend supports transfers across nodes
* supportsNotif(): Indicates if the backend supports notifications
* getSupportedMems(): Indicates memory types supported by the backend
* supportsTraceContext(): Indicates if the backend carries a request's trace context to its peer. Optional and `false` by default; only in-tree plugins can implement it, since the trace context type is internal

Based on the first 3 methods (supports*), the required methods to be implemented change. For instance, UCX backend implements all as it supports all scenarios, while GDS backend only has supportsLocal, detailed more in Example implementations. Note that a network backend must have supportsRemote and supportsNotif to be set to true, and preferably supportsLocal also to true, so another backend doesn't need to be involved for local transfers. For a storage backend, it should have supportsLocal and supportsNotif is optional.

Expand Down
18 changes: 14 additions & 4 deletions docs/tracing.md
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,16 @@ the value onto its own request handle. A carrier that puts it on the wire calls
`encodeTraceContext` itself. `TraceContext` is internal like the sink, so only
in-tree plugins can read it; none does yet.

A backend that carries the context to its peer says so by overriding
`supportsTraceContext()`, which defaults to `false`; core reads it once, when
the backend is created. Tracing never fails a transfer over it. A sampled
request on a backend that supports remote transfers but not trace contexts runs
as usual, its trace just does not reach the remote agent, and core logs a
warning once per backend in each agent; backends without remote transfers are
not reported. Likewise, if generating a request's context fails, the request
runs with the zeroed context and no correlation id, and each agent logs its
first such failure as an error.

## Profiling with NVTX / Nsight Systems

NVTX is a lazy, online API: when no profiler is attached, ranges are near-zero-cost
Expand Down Expand Up @@ -287,10 +297,10 @@ In NIXL tracing, "correlation" can mean two different things:
(e.g. `postXferReq` on the caller thread vs. the completion polled on another) to the
same request. Implemented via the backend-agnostic `pushCorrelationId()` /
`popCorrelationId()` API: NIXL wraps `postXferReq` and the `xfer.complete` marker in a
correlation scope keyed on the transfer-request handle's address, and the NVTX backend
records that id as the event's `uint64` payload -- so a post and its completion carry
the same id on the timeline regardless of which thread emitted each. (The id is a
process-local diagnostic handle, not a globally unique on-the-wire id.)
correlation scope keyed on the span id of the request's trace context, and the NVTX
backend records that id as the event's `uint64` payload -- so a post and its completion
carry the same id on the timeline regardless of which thread emitted each. A request
without a valid context carries no id. (The id is not propagated across processes yet.)

## Planned work

Expand Down
7 changes: 7 additions & 0 deletions src/api/cpp/backend/backend_engine.h
Original file line number Diff line number Diff line change
Expand Up @@ -247,5 +247,12 @@ class nixlBackendEngine {
const nixl_opt_args_t *extra_params = nullptr) const {
return NIXL_ERR_NOT_SUPPORTED;
}

// Determines if a backend carries a request's trace context to its peer. Core reads it
// once for remote backends and, if false, warns once when a sampled request goes untraced.
[[nodiscard]] virtual bool
supportsTraceContext() const noexcept {
return false;
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
};
#endif
2 changes: 1 addition & 1 deletion src/api/cpp/backend/backend_plugin.h
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class nixlUcxEngine;
// libstdc++ ABI as the NIXL core library. NIXL core currently requires C++20.

// Define the plugin API version
#define NIXL_PLUGIN_API_VERSION 2
#define NIXL_PLUGIN_API_VERSION 3

// Define the plugin interface class
class nixlBackendPlugin {
Expand Down
5 changes: 5 additions & 0 deletions src/core/agent_data.h
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
#include "nixl_metadata_context.h"
#include "telemetry.h"
#include "tracing/trace.h"
#include "tracing/trace_context.h"
#include "sync.h"

#include <atomic>
Expand Down Expand Up @@ -65,6 +66,7 @@ class nixlAgentData final : public nixlMetadataContext {
// backend is active.
const std::unique_ptr<nixl::trace::Tracer> tracer_;
std::unordered_map<nixl_backend_t, std::unique_ptr<nixlBackendTraceSink>> traceSinks_;
std::unordered_map<nixl_backend_t, std::atomic<bool>> traceContextUnsupported_;
// Bookkeeping for local connection metadata and user handles per backend
std::unordered_map<nixl_backend_t, std::unique_ptr<nixlBackendH>> backendHandles_;
std::unordered_map<nixl_backend_t, nixl_blob_t> connMd_;
Expand Down Expand Up @@ -102,6 +104,9 @@ class nixlAgentData final : public nixlMetadataContext {
nixl_mem_t mem_type);
void
warnAboutEfaHardwareMismatch();
void
warnIfTraceContextUnsupported(const nixlBackendEngine &backend,
const nixl::trace::TraceContext &context) noexcept;

template<class dlist_t>
nixl_status_t
Expand Down
23 changes: 23 additions & 0 deletions src/core/nixl_agent.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
#include "common/str_util.h"
#include "telemetry.h"
#include "telemetry_event.h"
#include "tracing/log_once.h"
#include "tracing/trace.h"
#include "tracing/trace_macros.h"
#include "tracing/trace_sink.h"
Expand Down Expand Up @@ -296,6 +297,23 @@ nixlAgentData::warnAboutEfaHardwareMismatch() {
}
}

void
nixlAgentData::warnIfTraceContextUnsupported(const nixlBackendEngine &backend,
const nixl::trace::TraceContext &context) noexcept {
if (!context.sampled()) {
return;
}
const auto it = traceContextUnsupported_.find(backend.getType());
if (it == traceContextUnsupported_.end()) {
return;
}
nixl::trace::logOnce(it->second, [&backend] {
NIXL_WARN << "nixl::trace: backend '" << backend.getType()
<< "' cannot carry trace contexts, so sampled transfers through it reach "
"remote agents untraced; logged once per backend and agent";
});
}

nixl_status_t
nixlAgent::createBackend(const nixl_backend_t &type,
const nixl_b_params_t &params,
Expand Down Expand Up @@ -395,6 +413,9 @@ nixlAgent::createBackend(const nixl_backend_t &type,
if (backend->supportsRemote()) {
data->notifEngines.push_back(backend.get());
data->connMd_[type] = conn_info;
if ((data->tracer_ != nullptr) && !backend->supportsTraceContext()) {
data->traceContextUnsupported_.try_emplace(type);
}
}

// TODO: Simplify, e.g. by making nixlBackendH's c'tor public?
Expand Down Expand Up @@ -924,6 +945,7 @@ nixlAgent::makeXferReq(nixl_xfer_op_t operation,
return ret;
}

data->warnIfTraceContextUnsupported(*backend, handle->traceContext());
req_hndl = handle.release();
return NIXL_SUCCESS;
}
Expand Down Expand Up @@ -1071,6 +1093,7 @@ nixlAgent::createXferReq(const nixl_xfer_op_t &operation,
return ret1;
}

data->warnIfTraceContextUnsupported(*handle->engine, handle->traceContext());
req_hndl = handle.release();
return NIXL_SUCCESS;
}
Expand Down
40 changes: 40 additions & 0 deletions src/core/tracing/log_once.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 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_TRACING_LOG_ONCE_H
#define NIXL_SRC_CORE_TRACING_LOG_ONCE_H

#include <atomic>

namespace nixl::trace {

template<typename logFn>
void
logOnce(std::atomic<bool> &logged, logFn &&log) noexcept {
if (logged.load(std::memory_order_relaxed) ||
logged.exchange(true, std::memory_order_relaxed)) {
return;
}
try {
log();
}
catch (...) {
}
}

} // namespace nixl::trace

#endif // NIXL_SRC_CORE_TRACING_LOG_ONCE_H
32 changes: 29 additions & 3 deletions src/core/tracing/trace.h
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@
* a public header later without churn if an external consumer appears.
*/

#include <atomic>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <string>
Expand Down Expand Up @@ -152,14 +154,22 @@ class Span {
std::vector<std::unique_ptr<SpanBackend>> backends_;
};

/**
* @brief Fills its buffer with random bytes for trace-context ids, or throws a
* std::exception. An all-zero id is drawn again; null selects the default
* getentropy-backed source.
*/
using random_bytes_fn_t = void (*)(std::uint8_t *, std::size_t);

/**
* @brief Composite tracer owned by the Agent. A single beginSpan()/mark() call
* fans out to every enabled backend. No singleton; injected explicitly.
*/
class Tracer {
public:
explicit Tracer(std::vector<std::unique_ptr<TraceBackend>> backends,
double sample_ratio = 0.0) noexcept;
double sample_ratio = 0.0,
random_bytes_fn_t random_bytes = nullptr) noexcept;
~Tracer();
Tracer(const Tracer &) = delete;
Tracer &
Expand All @@ -180,6 +190,18 @@ class Tracer {
return sampleRatio_;
}

/** @brief Random-byte source for this agent's trace contexts; null means the
* default. Carried here for the same reason as sampleRatio(). */
[[nodiscard]] random_bytes_fn_t
randomBytes() const noexcept {
return randomBytes_;
}

/** @brief Logs this tracer's first trace-context generation failure at ERROR;
* later ones are not logged. */
void
reportContextFailure(std::string_view reason) const noexcept;

[[nodiscard]] Span
beginSpan(std::string_view name, Kind kind = Kind::Generic);

Expand All @@ -194,18 +216,22 @@ class Tracer {
private:
std::vector<std::unique_ptr<TraceBackend>> backends_;
const double sampleRatio_;
const random_bytes_fn_t randomBytes_;
mutable std::atomic<bool> contextFailureLogged_ = false;
};

/**
* @brief RAII guard that keeps a correlation id active on the tracer for its
* lifetime, so every span/mark created while it is in scope is tagged
* with that id. Null-tracer-safe. Used to link spans emitted on different
* with that id. Null-tracer-safe; an id of 0, which an invalid context
* projects to, pushes nothing. Used to link spans emitted on different
* threads (e.g. postXferReq on the caller thread vs. the completion on
* the polling thread) to the same logical request.
*/
class CorrelationScope {
public:
[[nodiscard]] CorrelationScope(Tracer *tracer, std::uint64_t id) : tracer_(tracer) {
[[nodiscard]] CorrelationScope(Tracer *tracer, std::uint64_t id)
: tracer_(id == 0 ? nullptr : tracer) {
if (tracer_ != nullptr) {
tracer_->pushCorrelationId(id);
}
Expand Down
22 changes: 16 additions & 6 deletions src/core/tracing/trace_context.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -100,13 +100,16 @@ traceIdRandomness(const nixl::trace::TraceContext &context) noexcept {
}

void
generateInto(nixl::trace::TraceContext &context, double sample_ratio) {
generateInto(nixl::trace::TraceContext &context,
double sample_ratio,
nixl::trace::random_bytes_fn_t random_bytes) {
const auto fill = (random_bytes != nullptr) ? random_bytes : nixl::generateRandomBytes;
do {
nixl::generateRandomBytes(context.traceId.data(), context.traceId.size());
fill(context.traceId.data(), context.traceId.size());
Comment thread
coderabbitai[bot] marked this conversation as resolved.
} while (isAllZero(context.traceId));

do {
nixl::generateRandomBytes(context.spanId.data(), context.spanId.size());
fill(context.spanId.data(), context.spanId.size());
} while (isAllZero(context.spanId));

context.flags = 0x02;
Expand All @@ -125,8 +128,15 @@ appendByte(std::string &result, std::uint8_t value) {
} // namespace

nixl::trace::TraceContext::TraceContext(const nixl::trace::Tracer *tracer) {
if (tracer != nullptr) {
generateInto(*this, tracer->sampleRatio());
if (tracer == nullptr) {
return;
}
try {
generateInto(*this, tracer->sampleRatio(), tracer->randomBytes());
}
catch (const std::exception &e) {
*this = TraceContext{};
tracer->reportContextFailure(e.what());
}
}

Expand Down Expand Up @@ -287,6 +297,6 @@ nixl::trace::resolveTraceSampleRatio() {
nixl::trace::TraceContext
nixl::trace::generateTraceContext(double sample_ratio) {
nixl::trace::TraceContext context;
generateInto(context, sample_ratio);
generateInto(context, sample_ratio, nullptr);
return context;
}
16 changes: 14 additions & 2 deletions src/core/tracing/tracer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

#include "common/nixl_log.h"
#include "plugin_manager.h"
#include "tracing/log_once.h"

namespace nixl::trace {

Expand Down Expand Up @@ -76,12 +77,23 @@ Span::id() const noexcept {

/*** Tracer ***/

Tracer::Tracer(std::vector<std::unique_ptr<TraceBackend>> backends, double sample_ratio) noexcept
Tracer::Tracer(std::vector<std::unique_ptr<TraceBackend>> backends,
double sample_ratio,
random_bytes_fn_t random_bytes) noexcept
: backends_(std::move(backends)),
sampleRatio_(sample_ratio) {}
sampleRatio_(sample_ratio),
randomBytes_(random_bytes) {}

Tracer::~Tracer() = default;

void
Tracer::reportContextFailure(std::string_view reason) const noexcept {
logOnce(contextFailureLogged_, [reason] {
NIXL_ERROR << "nixl::trace: could not generate a trace context (" << reason
<< "); the request continues untraced and further failures are not logged";
});
}

Span
Tracer::beginSpan(std::string_view name, Kind kind) {
std::vector<std::unique_ptr<SpanBackend>> spans;
Expand Down
25 changes: 25 additions & 0 deletions test/gtest/common.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
#include <sys/socket.h>
#include <unistd.h>
#include <random>
#include "absl/log/globals.h"
#include "absl/log/log_sink_registry.h"

namespace gtest {
Expand Down Expand Up @@ -212,4 +213,28 @@ LogProblemCounter::Send(const absl::LogEntry &entry) {
std::cerr << "ATTENTION: Message is '" << msg << '\'' << std::endl;
}

LogMessageCounter::LogMessageCounter(std::string_view needle)
: needle_(needle),
minLogLevel_(absl::MinLogLevel()) {
absl::SetMinLogLevel(absl::LogSeverityAtLeast::kInfo);
absl::AddLogSink(static_cast<absl::LogSink *>(this));
}

LogMessageCounter::~LogMessageCounter() {
absl::RemoveLogSink(static_cast<absl::LogSink *>(this));
absl::SetMinLogLevel(minLogLevel_);
}

size_t
LogMessageCounter::count() const noexcept {
return count_.load(std::memory_order_relaxed);
}

void
LogMessageCounter::Send(const absl::LogEntry &entry) {
if (entry.text_message().find(needle_) != std::string_view::npos) {
count_.fetch_add(1, std::memory_order_relaxed);
}
}

} // namespace gtest
Loading
Loading