From bc4202cc0e8faa06f2009a27b9543d5cc4694c5f Mon Sep 17 00:00:00 2001 From: Asaf Schwartz Date: Mon, 13 Jul 2026 19:56:17 +0300 Subject: [PATCH 1/3] refactor(metadata): add backend contract and P2P backend Introduce the metadata-exchange backend layer without wiring it in yet: the nixlMetadataBackend contract, the nixlMetadataContext interface (the agent-side operations a backend needs), and the self-contained P2P (socket) backend that implements the contract against the context. Nothing constructs these yet; the agent still uses its inline path. The manager that owns and dispatches to backends, and the cutover that removes the inline path, come in the following commit. Splitting the introduction out keeps that cutover reviewable on its own. --- src/core/agent_data.h | 46 +++ src/core/meson.build | 1 + src/core/nixl_metadata_backend.h | 109 +++++++ src/core/nixl_p2p_metadata_backend.cpp | 392 +++++++++++++++++++++++++ src/core/nixl_p2p_metadata_backend.h | 91 ++++++ 5 files changed, 639 insertions(+) create mode 100644 src/core/nixl_metadata_backend.h create mode 100644 src/core/nixl_p2p_metadata_backend.cpp create mode 100644 src/core/nixl_p2p_metadata_backend.h diff --git a/src/core/agent_data.h b/src/core/agent_data.h index 3949a50ca8..52161191f0 100644 --- a/src/core/agent_data.h +++ b/src/core/agent_data.h @@ -63,6 +63,52 @@ using nixl_socket_peer_t = std::pair; using nixl_socket_map_t = std::map; +class nixlMDManager; + +/** + * @class nixlMetadataContext + * @brief Core-internal interface: the agent-side operations a metadata backend + * needs. + * + * Implemented by nixlAgentData. Backends hold a reference to this interface + * instead of the concrete agent, so they never reference nixlAgent (no cycle) + * and need no friendship. + */ +class nixlMetadataContext { +public: + virtual ~nixlMetadataContext() = default; + + /** Serialize this agent's full local metadata blob. */ + [[nodiscard]] virtual nixl_status_t + getLocalMD(nixl_blob_t &blob) = 0; + + /** Serialize a partial local metadata blob for the given descriptors. */ + [[nodiscard]] virtual nixl_status_t + getLocalPartialMD(const nixl_reg_dlist_t &descs, + nixl_blob_t &blob, + const nixl_opt_args_t *extra_params) = 0; + + /** This agent's name; used by centralized backends to build their KV keys. */ + [[nodiscard]] virtual const std::string & + getName() const = 0; + + /** Agent configuration a backend needs (listen port/thread, watch timeout). */ + [[nodiscard]] virtual const nixlAgentConfig & + getConfig() const = 0; + + /** + * Deserialize a received metadata blob into the remote-section cache, + * returning the embedded remote agent name. Used by every backend to load a + * fetched/received blob. + */ + [[nodiscard]] virtual nixl_status_t + loadRemoteMD(const nixl_blob_t &blob, std::string &out_name) = 0; + + /** Evict a remote agent's cached metadata (used by inbound INVL / watches). */ + virtual nixl_status_t + invalidateRemoteMD(const std::string &remote_name) = 0; +}; + class nixlAgentData { private: const std::string name_; diff --git a/src/core/meson.build b/src/core/meson.build index ee542bf94d..90b24f50a1 100644 --- a/src/core/meson.build +++ b/src/core/meson.build @@ -64,6 +64,7 @@ endif nixl_lib_sources = [ 'nixl_agent.cpp', 'nixl_enum_strings.cpp', + 'nixl_p2p_metadata_backend.cpp', 'nixl_plugin_manager.cpp', 'nixl_listener.cpp', 'telemetry/telemetry.cpp', diff --git a/src/core/nixl_metadata_backend.h b/src/core/nixl_metadata_backend.h new file mode 100644 index 0000000000..0daa048b91 --- /dev/null +++ b/src/core/nixl_metadata_backend.h @@ -0,0 +1,109 @@ +/* + * 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. + */ +/** + * @file nixl_metadata_backend.h + * @brief Core-internal contract for nixlMDManager metadata backends. + */ +#ifndef NIXL_SRC_CORE_NIXL_METADATA_BACKEND_H +#define NIXL_SRC_CORE_NIXL_METADATA_BACKEND_H + +#include "nixl_descriptors.h" +#include "nixl_types.h" + +#include +#include +#include + +/** A unit of transport I/O produced on the caller thread, run on the worker. */ +using nixlWorkerTask = std::function; + +/** + * @struct nixlPreparedOp + * @brief Result of a backend's caller-thread prepare step. + * + * @var status Synchronous validation/serialization result, returned to the + * caller. Anything other than NIXL_SUCCESS means the op was rejected + * and no task should run. + * @var task The transport work to run on the manager's worker thread. Empty + * when there is nothing to schedule. + */ +struct nixlPreparedOp { + nixl_status_t status = NIXL_SUCCESS; + nixlWorkerTask task; +}; + +/** + * @class nixlMetadataBackend + * @brief Metadata-exchange contract that nixlMDManager dispatches to. + * + * Each transport implements this contract (P2P, ETCD). Core-internal: + * not part of the installed public headers, so backend dependencies never leak + * into the public API. Operational addressing (`ipAddr`/`port`, `metadataLabel`) + * is carried in `nixl_opt_args_t`. + * + * Thread contract, encoded in the interface: + * - the `prepare*` methods run on the CALLER thread: they validate and + * serialize, return a synchronous status, and hand back the transport work as + * a nixlWorkerTask. They must not block on I/O. + * - the returned nixlWorkerTask and `serviceEvents()` run on the manager's + * WORKER thread: that is where all blocking transport I/O belongs. + * The manager owns scheduling; backends never touch a queue or a thread. + */ +class nixlMetadataBackend { +public: + virtual ~nixlMetadataBackend() = default; + + /** Stable transport name reported by nixlMDManager::backendName(). */ + [[nodiscard]] virtual std::string_view + name() const = 0; + + /** Caller thread: prepare a full-metadata publish. */ + [[nodiscard]] virtual nixlPreparedOp + prepareSendLocal(const nixl_opt_args_t *extra_params) = 0; + + /** Caller thread: prepare a partial-metadata publish. */ + [[nodiscard]] virtual nixlPreparedOp + prepareSendLocalPartial(const nixl_reg_dlist_t &descs, const nixl_opt_args_t *extra_params) = 0; + + /** Caller thread: prepare retrieval of a remote agent's metadata. */ + [[nodiscard]] virtual nixlPreparedOp + prepareFetchRemote(const std::string &remote_name, const nixl_opt_args_t *extra_params) = 0; + + /** Caller thread: prepare withdrawal of our metadata. */ + [[nodiscard]] virtual nixlPreparedOp + prepareInvalidateLocal(const nixl_opt_args_t *extra_params) = 0; + + /** + * @brief Whether this backend needs the manager's worker thread running + * (for background servicing and/or to execute its tasks). Default + * false (a backend that does nothing off-thread). + */ + [[nodiscard]] virtual bool + needsWorker() const { + return false; + } + + /** + * @brief Worker thread: one pass of background servicing, called repeatedly + * (e.g. accept peers / read replies for P2P, drain watch invalidations + * for ETCD). Default no-op. + */ + virtual void + serviceEvents() {} +}; + +#endif // NIXL_SRC_CORE_NIXL_METADATA_BACKEND_H diff --git a/src/core/nixl_p2p_metadata_backend.cpp b/src/core/nixl_p2p_metadata_backend.cpp new file mode 100644 index 0000000000..f4a0ceb30d --- /dev/null +++ b/src/core/nixl_p2p_metadata_backend.cpp @@ -0,0 +1,392 @@ +/* + * 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. + */ +#include "nixl_p2p_metadata_backend.h" + +#include "agent_data.h" +#include "stream/metadata_stream.h" +#include "common/nixl_log.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +namespace { + +// Socket helpers, moved verbatim from the former nixl_listener.cpp. They are the +// wire mechanics of the P2P transport and belong with this backend. + +int +connectToIP(const std::string &ip_addr, int port) { + struct sockaddr_in listenerAddr; + listenerAddr.sin_port = htons(port); + listenerAddr.sin_family = AF_INET; + + if (inet_pton(AF_INET, ip_addr.c_str(), &listenerAddr.sin_addr) <= 0) { + NIXL_ERROR << "inet_pton failed for ip_addr: " << ip_addr; + return -1; + } + + int ret_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0); + if (ret_fd == -1) { + NIXL_ERROR << "socket creation failed for ip_addr: " << ip_addr << " and port: " << port; + return -1; + } + + int ret = connect(ret_fd, (struct sockaddr *)&listenerAddr, sizeof(listenerAddr)); + if (ret < 0 && errno != EINPROGRESS) { + close(ret_fd); + return -1; + } + + struct pollfd pfd; + pfd.fd = ret_fd; + pfd.events = POLLOUT; + pfd.revents = 0; + + ret = poll(&pfd, 1, 1000); // 1000ms timeout + if (ret <= 0) { + if (ret < 0) { + NIXL_PERROR << "poll failed for ip_addr: " << ip_addr << " and port: " << port; + } else { + NIXL_ERROR << "poll timed out for ip_addr: " << ip_addr << " and port: " << port; + } + close(ret_fd); + return -1; + } + + if (!(pfd.revents & POLLOUT)) { + NIXL_ERROR << "poll returned but socket not ready for write for ip_addr: " << ip_addr + << " and port: " << port; + close(ret_fd); + return -1; + } + + int error = 0; + socklen_t len = sizeof(error); + if (getsockopt(ret_fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0) { + NIXL_PERROR << "getsockopt failed for ip_addr: " << ip_addr << " and port: " << port; + close(ret_fd); + return -1; + } + + if (error != 0) { + errno = error; // For the 'PERROR'. + NIXL_PERROR << "getsockopt gave error for ip_addr: " << ip_addr << " and port: " << port; + close(ret_fd); + return -1; + } + + return ret_fd; +} + +void +sendCommMessage(int fd, const std::string &msg) { + size_t size = msg.size(); + constexpr size_t iov_size = 2; + struct iovec iov[iov_size] = {{&size, sizeof(size)}, + {const_cast(msg.data()), msg.size()}}; + + for (size_t i = 0, offset = 0, sent = 0; i < iov_size;) { + auto bytes = send(fd, + static_cast(iov[i].iov_base) + offset, + iov[i].iov_len - offset, + MSG_NOSIGNAL); + if (bytes < 0) { + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { + continue; + } + throw std::runtime_error( + absl::StrFormat("sendCommMessage(fd=%d, msg=%s) %zu/%zu bytes failed, errno=%d", + fd, + msg.c_str(), + sent, + size + sizeof(size), + errno)); + } + offset += bytes; + sent += bytes; + if (offset == iov[i].iov_len) { + offset = 0; + ++i; + } + } +} + +bool +recvCommMessageType(int fd, void *data, size_t size, bool force = false) { + for (size_t received = 0; received < size;) { + auto bytes = recv(fd, static_cast(data) + received, size - received, 0); + if (bytes > 0) { + received += bytes; + continue; + } + if (bytes == 0 && received == 0 && !force) { + return false; + } + if (bytes < 0) { + if (errno == EINTR) { + continue; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) { + if (!force && received == 0) { + return false; // nothing to read yet + } + continue; + } + } + throw std::runtime_error( + absl::StrFormat("recvCommMessage(fd=%d) %zu/%zu bytes failed ret=%d errno=%d", + fd, + received, + size, + bytes, + errno)); + } + return true; +} + +bool +recvCommMessage(int fd, std::string &msg) { + size_t size; + if (!recvCommMessageType(fd, &size, sizeof(size))) { + return false; + } + msg.resize(size); + return recvCommMessageType(fd, msg.data(), size, true); +} + +} // namespace + +nixlP2PMetadataBackend::nixlP2PMetadataBackend(nixlMetadataContext &ctx) : ctx_(ctx) { + if (ctx_.getConfig().useListenThread) { + listener_ = std::make_unique(ctx_.getConfig().listenPort); + listener_->setupListener(); // throws on bind/listen failure + } +} + +nixlP2PMetadataBackend::~nixlP2PMetadataBackend() { + // The worker is already stopped by the time backends are destroyed, so no + // other thread touches remoteSockets_ here. + for (auto &[peer, fd] : remoteSockets_) { + shutdown(fd, SHUT_RDWR); + close(fd); + } +} + +std::string_view +nixlP2PMetadataBackend::name() const { + return "P2P"; +} + +bool +nixlP2PMetadataBackend::needsWorker() const { + return ctx_.getConfig().useListenThread; +} + +nixlPreparedOp +nixlP2PMetadataBackend::prepareSendLocal(const nixl_opt_args_t *extra_params) { + if (!extra_params || extra_params->ipAddr.empty()) { + return {NIXL_ERR_INVALID_PARAM, {}}; + } + nixl_blob_t blob; + const nixl_status_t ret = ctx_.getLocalMD(blob); + if (ret < 0) { + return {ret, {}}; + } + const std::string ip = extra_params->ipAddr; + const int port = extra_params->port; + return {NIXL_SUCCESS, [this, ip, port, blob = std::move(blob)]() { + sendToPeer(ip, port, "NIXLCOMM:LOAD" + blob); + }}; +} + +nixlPreparedOp +nixlP2PMetadataBackend::prepareSendLocalPartial(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params) { + if (!extra_params || extra_params->ipAddr.empty()) { + return {NIXL_ERR_INVALID_PARAM, {}}; + } + nixl_blob_t blob; + const nixl_status_t ret = ctx_.getLocalPartialMD(descs, blob, extra_params); + if (ret < 0) { + return {ret, {}}; + } + const std::string ip = extra_params->ipAddr; + const int port = extra_params->port; + return {NIXL_SUCCESS, [this, ip, port, blob = std::move(blob)]() { + sendToPeer(ip, port, "NIXLCOMM:LOAD" + blob); + }}; +} + +nixlPreparedOp +nixlP2PMetadataBackend::prepareFetchRemote(const std::string & /*remote_name*/, + const nixl_opt_args_t *extra_params) { + if (!extra_params || extra_params->ipAddr.empty()) { + return {NIXL_ERR_INVALID_PARAM, {}}; + } + // Socket fetch is keyed by address, not name; the reply is loaded into the + // remote-section cache by serviceEvents() when the peer answers. + const std::string ip = extra_params->ipAddr; + const int port = extra_params->port; + return {NIXL_SUCCESS, [this, ip, port]() { sendToPeer(ip, port, "NIXLCOMM:SEND"); }}; +} + +nixlPreparedOp +nixlP2PMetadataBackend::prepareInvalidateLocal(const nixl_opt_args_t *extra_params) { + if (!extra_params || extra_params->ipAddr.empty()) { + return {NIXL_ERR_INVALID_PARAM, {}}; + } + const std::string ip = extra_params->ipAddr; + const int port = extra_params->port; + return {NIXL_SUCCESS, + [this, ip, port]() { sendToPeer(ip, port, "NIXLCOMM:INVL" + ctx_.getName()); }}; +} + +void +nixlP2PMetadataBackend::serviceEvents() { + if (listener_) { + acceptPeers(); + } + readIncoming(); +} + +void +nixlP2PMetadataBackend::sendToPeer(const std::string &ip, int port, const std::string &msg) { + const auto key = std::make_pair(ip, port); + auto client = remoteSockets_.find(key); + if (client == remoteSockets_.end()) { + const int new_client = connectToIP(ip, port); + if (new_client == -1) { + NIXL_ERROR << "P2P backend could not connect to IP " << ip << " and port " << port; + return; + } + client = remoteSockets_.emplace(key, new_client).first; + } + try { + sendCommMessage(client->second, msg); + } + catch (const std::runtime_error &e) { + NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); + close(client->second); + remoteSockets_.erase(client); + } +} + +void +nixlP2PMetadataBackend::acceptPeers() { + int new_fd = 0; + while (new_fd != -1) { + new_fd = listener_->acceptClient(); + if (new_fd == -1) { + break; + } + sockaddr_in client_address; + socklen_t client_addrlen = sizeof(client_address); + if (getpeername(new_fd, (sockaddr *)&client_address, &client_addrlen) != 0) { + NIXL_PERROR << "getpeername failed for accepted client"; + close(new_fd); + continue; + } + char client_ip[INET_ADDRSTRLEN]; + if (inet_ntop(AF_INET, &client_address.sin_addr, client_ip, INET_ADDRSTRLEN) == nullptr) { + NIXL_PERROR << "inet_ntop failed for client address"; + close(new_fd); + continue; + } + remoteSockets_[std::make_pair(std::string(client_ip), (int)client_address.sin_port)] = + new_fd; + const int flags = fcntl(new_fd, F_GETFL, 0); + if (flags == -1 || fcntl(new_fd, F_SETFL, flags | O_NONBLOCK) == -1) { + NIXL_PERROR << "fcntl failed for accepted client"; + } + } +} + +void +nixlP2PMetadataBackend::readIncoming() { + auto socket_iter = remoteSockets_.begin(); + while (socket_iter != remoteSockets_.end()) { + std::string commands; + bool disconnected = false; + + try { + if (!recvCommMessage(socket_iter->second, commands)) { + ++socket_iter; + continue; + } + } + catch (const std::runtime_error &e) { + NIXL_ERROR << "Failed to receive message from peer, disconnecting: " << e.what(); + close(socket_iter->second); + socket_iter = remoteSockets_.erase(socket_iter); + continue; + } + + for (const auto &command : absl::StrSplit(commands, "NIXLCOMM:")) { + if (command.size() < 4) { + continue; + } + const std::string header = std::string(command.substr(0, 4)); + + if (header == "LOAD") { + std::string remote_agent; + const nixl_status_t ret = + ctx_.loadRemoteMD(std::string(command.substr(4)), remote_agent); + if (ret != NIXL_SUCCESS) { + NIXL_ERROR << "loadRemoteMD in P2P backend failed from peer " + << socket_iter->first.first << ":" << socket_iter->first.second + << " with error " << ret; + } + } else if (header == "SEND") { + nixl_blob_t blob; + (void)ctx_.getLocalMD(blob); + try { + sendCommMessage(socket_iter->second, "NIXLCOMM:LOAD" + blob); + } + catch (const std::runtime_error &e) { + NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); + disconnected = true; + break; + } + } else if (header == "INVL") { + (void)ctx_.invalidateRemoteMD(std::string(command.substr(4))); + break; + } else { + NIXL_ERROR << "Received socket message with bad header " << header << " from peer " + << socket_iter->first.first << ":" << socket_iter->first.second; + } + } + + if (disconnected) { + close(socket_iter->second); + socket_iter = remoteSockets_.erase(socket_iter); + } else { + ++socket_iter; + } + } +} diff --git a/src/core/nixl_p2p_metadata_backend.h b/src/core/nixl_p2p_metadata_backend.h new file mode 100644 index 0000000000..4a56d137a2 --- /dev/null +++ b/src/core/nixl_p2p_metadata_backend.h @@ -0,0 +1,91 @@ +/* + * 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. + */ +/** + * @file nixl_p2p_metadata_backend.h + * @brief Point-to-point (socket) metadata backend. + */ +#ifndef NIXL_SRC_CORE_NIXL_P2P_METADATA_BACKEND_H +#define NIXL_SRC_CORE_NIXL_P2P_METADATA_BACKEND_H + +#include "nixl_metadata_backend.h" + +#include +#include +#include +#include +#include + +class nixlMetadataContext; +class nixlMDStreamListener; + +/** + * @class nixlP2PMetadataBackend + * @brief Self-contained socket-based metadata backend. + * + * Owns its transport state: the open peer connections and (when the agent + * enables listening) the accept socket. Outbound ops validate and serialize + * synchronously, then submit the socket send as a task on the manager's worker + * thread; inbound work (accepting peers, reading LOAD/SEND/INVL replies) happens + * in serviceEvents(), also on that worker thread, so the connection map is only + * ever touched by one thread. Depends only on nixlMetadataContext, not nixlAgent. + */ +class nixlP2PMetadataBackend : public nixlMetadataBackend { +public: + explicit nixlP2PMetadataBackend(nixlMetadataContext &ctx); + ~nixlP2PMetadataBackend() override; + + [[nodiscard]] std::string_view + name() const override; + + [[nodiscard]] nixlPreparedOp + prepareSendLocal(const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareSendLocalPartial(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareFetchRemote(const std::string &remote_name, + const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareInvalidateLocal(const nixl_opt_args_t *extra_params) override; + + // The worker is needed for inbound servicing when listening is enabled. + [[nodiscard]] bool + needsWorker() const override; + + // Accept new peers (if listening) and read/dispatch incoming messages. + void + serviceEvents() override; + +private: + // Connect-on-demand to (ip, port) and send msg; disconnect on error. + // Runs only on the worker thread (submitted task). + void + sendToPeer(const std::string &ip, int port, const std::string &msg); + void + acceptPeers(); + void + readIncoming(); + + nixlMetadataContext &ctx_; + std::map, int> remoteSockets_; + std::unique_ptr listener_; +}; + +#endif // NIXL_SRC_CORE_NIXL_P2P_METADATA_BACKEND_H From 8cb653109a3dfa6e3326f9d407a82899e5ba24f0 Mon Sep 17 00:00:00 2001 From: Asaf Schwartz Date: Mon, 13 Jul 2026 19:58:33 +0300 Subject: [PATCH 2/3] refactor(metadata): make the manager the single exchange path Wire the backend layer in and remove the inline path. nixlAgentData now owns a nixlMDManager (built unconditionally) that owns the worker thread and dispatches each call to a backend by precedence: a peer address selects P2P, otherwise the configured centralized store (ETCD). Add the self-contained ETCD backend (owns its nixlEtcdClient). nixlAgent's exchange methods (sendLocalMD, sendLocalPartialMD, fetchRemoteMD, invalidateLocalMD) become thin wrappers over the manager; pure-cache operations are consolidated on nixlAgentData, which implements nixlMetadataContext for the backends. nixl_listener.cpp is deleted: its worker loop moves into the manager, its socket helpers into the P2P backend, and its etcd client into the ETCD backend. Public API and observable P2P/ETCD behavior are unchanged; gtest log expectations are updated to the new backend messages. --- src/core/agent_data.h | 69 +- src/core/meson.build | 3 +- src/core/nixl_agent.cpp | 411 ++++++------ src/core/nixl_etcd_metadata_backend.cpp | 379 +++++++++++ src/core/nixl_etcd_metadata_backend.h | 82 +++ src/core/nixl_listener.cpp | 820 ------------------------ src/core/nixl_md_manager.cpp | 276 ++++++++ src/core/nixl_md_manager.h | 183 ++++++ test/gtest/metadata_exchange.cpp | 12 +- 9 files changed, 1179 insertions(+), 1056 deletions(-) create mode 100644 src/core/nixl_etcd_metadata_backend.cpp create mode 100644 src/core/nixl_etcd_metadata_backend.h delete mode 100644 src/core/nixl_listener.cpp create mode 100644 src/core/nixl_md_manager.cpp create mode 100644 src/core/nixl_md_manager.h diff --git a/src/core/agent_data.h b/src/core/agent_data.h index 52161191f0..9d1a830425 100644 --- a/src/core/agent_data.h +++ b/src/core/agent_data.h @@ -24,6 +24,7 @@ #include "sync.h" #include +#include #include #if HAVE_ETCD @@ -38,27 +39,7 @@ class SyncClient; using backend_list_t = std::vector; -//Internal typedef to define metadata communication request types -//To be extended with ETCD operations -enum nixl_comm_t { - SOCK_SEND, - SOCK_FETCH, - SOCK_INVAL, - SOCK_MAX, -#if HAVE_ETCD - ETCD_SEND, - ETCD_FETCH, - ETCD_INVAL -#endif // HAVE_ETCD -}; - -//Command to be sent to listener thread from NIXL API -// 1) Command type -// 2) IP Address -// 3) Port -// 4) Metadata to send (for sendLocalMD calls) -using nixl_comm_req_t = std::tuple; - +// A peer socket (ip, port) and the map of open connections the P2P backend owns. using nixl_socket_peer_t = std::pair; using nixl_socket_map_t = std::map; @@ -109,7 +90,11 @@ class nixlMetadataContext { invalidateRemoteMD(const std::string &remote_name) = 0; }; -class nixlAgentData { +// Implements nixlMetadataContext so metadata backends reach the serialization +// primitives and the comm thread without referencing nixlAgent. +// Preserve the grandfathered 8-space class layout below. +// clang-format off +class nixlAgentData : public nixlMetadataContext { private: const std::string name_; const nixlAgentConfig config_; @@ -131,15 +116,10 @@ class nixlAgentData { std::unordered_map> remoteBackends_; - // State/methods for listener thread - std::unique_ptr listener; - nixl_socket_map_t remoteSockets; - std::thread commThread; - std::vector commQueue; - std::mutex commLock; - std::atomic commThreadStop; - std::atomic agentShutdown; - std::exception_ptr commThreadException_; + // Agent-owned metadata manager; always built (single metadata path). + // It owns the worker thread and the pluggable backends (which now own + // their own transport state: sockets/listener for P2P, client for ETCD). + const std::unique_ptr md_; // The order of the following data members is crucial for destruction. // Bookkeeping for local connection metadata and user handles per backend @@ -153,12 +133,25 @@ class nixlAgentData { const std::unique_ptr tracer_; nixlLocalSection localSection_; - void - commWorker(nixlAgent &myAgent) noexcept; - void - commWorkerInternal(nixlAgent *myAgent); - void enqueueCommWork(nixl_comm_req_t request); - void getCommWork(std::vector &req_list); + // nixlMetadataContext impl; private as before (backends call via the interface). + [[nodiscard]] nixl_status_t + getLocalMD(nixl_blob_t &blob) override; + [[nodiscard]] nixl_status_t + getLocalPartialMD(const nixl_reg_dlist_t &descs, + nixl_blob_t &blob, + const nixl_opt_args_t *extra_params) override; + [[nodiscard]] const std::string & + getName() const override { + return name_; + } + [[nodiscard]] const nixlAgentConfig & + getConfig() const override { + return config_; + } + [[nodiscard]] nixl_status_t + loadRemoteMD(const nixl_blob_t &blob, std::string &out_name) override; + nixl_status_t + invalidateRemoteMD(const std::string &remote_name) override; nixl_status_t loadConnInfo(const std::string &remote_name, const nixl_backend_t &backend, @@ -187,6 +180,8 @@ class nixlAgentData { friend class nixlAgent; }; +// clang-format on + class nixlBackendEngine; // This class hides away the nixlBackendEngine from user of the Agent API diff --git a/src/core/meson.build b/src/core/meson.build index 90b24f50a1..24240a81a2 100644 --- a/src/core/meson.build +++ b/src/core/meson.build @@ -64,9 +64,10 @@ endif nixl_lib_sources = [ 'nixl_agent.cpp', 'nixl_enum_strings.cpp', + 'nixl_md_manager.cpp', 'nixl_p2p_metadata_backend.cpp', + 'nixl_etcd_metadata_backend.cpp', 'nixl_plugin_manager.cpp', - 'nixl_listener.cpp', 'telemetry/telemetry.cpp', 'telemetry/buffer_exporter.cpp', 'telemetry/buffer_plugin.cpp', diff --git a/src/core/nixl_agent.cpp b/src/core/nixl_agent.cpp index 5e123ce896..b57c85e73b 100644 --- a/src/core/nixl_agent.cpp +++ b/src/core/nixl_agent.cpp @@ -30,6 +30,7 @@ #include "backend/backend_engine.h" #include "transfer_request.h" #include "agent_data.h" +#include "nixl_md_manager.h" #include "plugin_manager.h" #include "common/configuration.h" #include "common/nixl_log.h" @@ -152,11 +153,8 @@ namespace { [[nodiscard]] bool detectEtcd() { -#if HAVE_ETCD - return nixl::config::checkExistence("NIXL_ETCD_ENDPOINTS"); -#else - return false; -#endif + // Single source of truth (shared with the manager's backend selection). + return nixlMDManager::etcdConfigured(); } // The comm thread (used for etcd or listen-based metadata exchange) shares @@ -188,14 +186,26 @@ makeAgentTracer(const std::string &name) { return nixl::trace::makeTracer(nixl::trace::TracerConfig{name, std::move(requested_backends)}); } +// The metadata manager is the single path for metadata exchange; it selects its +// backends from the environment (P2P plus an optional name-addressed backend). +// Built unconditionally so the public methods can delegate without a fallback. +[[nodiscard]] std::unique_ptr +makeMDManager(nixlMetadataContext &ctx) { + return std::make_unique(ctx); +} + } // namespace nixlAgentData::nixlAgentData(const std::string &name, const nixlAgentConfig &config) : name_(name), config_(config), useEtcd_(detectEtcd()), + // The manager runs a worker thread when P2P listens or a centralized store + // (etcd) is configured; that thread shares agent state, so the + // sync mode is upgraded to STRICT in those cases. needsCommThread_(useEtcd_ || config.useListenThread), lock(effectiveSyncMode(config.syncMode, needsCommThread_)), + md_(makeMDManager(*this)), tracer_(makeAgentTracer(name)) { #if HAVE_ETCD NIXL_DEBUG << "NIXL ETCD is " << (useEtcd_ ? "enabled" : "disabled"); @@ -223,56 +233,20 @@ nixlAgentData::nixlAgentData(const std::string &name, const nixlAgentConfig &con } /*** nixlAgent implementation ***/ -nixlAgent::nixlAgent(const std::string &name, const nixlAgentConfig &cfg) : - data(std::make_unique(name, cfg)) -{ - if(cfg.useListenThread) { - data->listener = std::make_unique(cfg.listenPort); - data->listener->setupListener(); // throws on bind/listen failure - } - - if (data->needsCommThread_) { - data->commThreadStop = false; - data->agentShutdown = false; - data->commThread = std::thread(&nixlAgentData::commWorker, data.get(), std::ref(*this)); - } +nixlAgent::nixlAgent(const std::string &name, const nixlAgentConfig &cfg) + : data(std::make_unique(name, cfg)) { + // The metadata manager owns the worker thread and the pluggable backends + // (which own their transport state, e.g. the P2P listener bound during + // nixlAgentData construction). Start the worker now that the agent is fully + // constructed, so it never touches partially-built state. + data->md_->start(); } nixlAgent::~nixlAgent() { - if (data->needsCommThread_) { - data->agentShutdown = true; - // commQueue is guarded by commLock (see enqueueCommWork/getCommWork); - // take the lock for the drain check to avoid racing the comm thread. - while (true) { - { - const std::lock_guard lock(data->commLock); - if (data->commQueue.empty()) { - break; - } - } - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - } - - data->commThreadStop = true; - if(data->commThread.joinable()) data->commThread.join(); - - try { - if (data->commThreadException_) { - std::rethrow_exception(data->commThreadException_); - } - } - catch (const std::exception &e) { - NIXL_WARN << "Communication thread has thrown an exception: " << e.what(); - } - - // Close remaining connections from comm thread - for (auto &[remote, fd] : data->remoteSockets) { - shutdown(fd, SHUT_RDWR); - close(fd); - } - - data->listener.reset(); - } + // Stop and join the manager's worker while the backends are still alive: + // they own sockets / etcd watchers that must be torn down only after the + // worker (which uses them) has joined. + data->md_->stop(); } nixl_status_t @@ -1470,14 +1444,13 @@ nixlAgent::genNotif(const std::string &remote_agent, } nixl_status_t -nixlAgent::getLocalMD (nixl_blob_t &str) const { - size_t conn_cnt; +nixlAgentData::getLocalMD(nixl_blob_t &str) { nixl_backend_t nixl_backend; nixl_status_t ret; - NIXL_LOCK_GUARD(data->lock); - // data->connMd_ was populated when the backend was created - conn_cnt = data->connMd_.size(); + NIXL_LOCK_GUARD(lock); + // connMd_ was populated when the backend was created + const size_t conn_cnt = connMd_.size(); if (conn_cnt == 0) { // Error, no backend supports remote NIXL_ERROR_FUNC << "no backends support remote operations"; @@ -1485,27 +1458,39 @@ nixlAgent::getLocalMD (nixl_blob_t &str) const { } nixlSerDes sd; - ret = sd.addStr("Agent", data->name_); + ret = sd.addStr("Agent", name_); // Always returns SUCCESS, serdes class logs errors if necessary - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } ret = sd.addBuf("Conns", &conn_cnt, sizeof(conn_cnt)); - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } - for (auto &c : data->connMd_) { + for (auto &c : connMd_) { nixl_backend = c.first; ret = sd.addStr("t", nixl_backend); - if (ret) break; + if (ret != NIXL_SUCCESS) { + break; + } ret = sd.addStr("c", c.second); - if (ret) break; + if (ret != NIXL_SUCCESS) { + break; + } + } + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; } - if (ret) return NIXL_ERR_UNKNOWN; ret = sd.addStr("", "MemSection"); - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } - ret = data->localSection_.serialize(&sd); - if (ret) { + ret = localSection_.serialize(&sd); + if (ret != NIXL_SUCCESS) { NIXL_ERROR_FUNC << "serialization failed"; return ret; } @@ -1515,19 +1500,26 @@ nixlAgent::getLocalMD (nixl_blob_t &str) const { } nixl_status_t -nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, - nixl_blob_t &str, - const nixl_opt_args_t* extra_params) const { +nixlAgent::getLocalMD(nixl_blob_t &str) const { + // Single serialization implementation lives in nixlAgentData (the metadata + // context the manager also uses); this public entry point delegates to it. + return data->getLocalMD(str); +} + +nixl_status_t +nixlAgentData::getLocalPartialMD(const nixl_reg_dlist_t &descs, + nixl_blob_t &str, + const nixl_opt_args_t *extra_params) { backend_list_t tmp_list; backend_list_t *backend_list; nixl_status_t ret; - NIXL_LOCK_GUARD(data->lock); + NIXL_LOCK_GUARD(lock); if (!extra_params || extra_params->backends.size() == 0) { - if (descs.descCount() != 0) { + if (!descs.isEmpty()) { // Non-empty dlist, return backends that support the memory type - backend_list = &data->memToBackend[descs.getType()]; + backend_list = &memToBackend[descs.getType()]; if (backend_list->empty()) { NIXL_ERROR_FUNC << "no available backends for mem type '" << descs.getType() << "'"; return NIXL_ERR_NOT_FOUND; @@ -1535,7 +1527,7 @@ nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, } else { // Empty dlist, return all backends backend_list = &tmp_list; - for (const auto &elm : data->backendEngines_) { + for (const auto &elm : backendEngines_) { backend_list->push_back(elm.second.get()); } } @@ -1549,10 +1541,12 @@ nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, // First find all relevant engines and their conn info. // Best effort, ignore if no conn info (meaning backend doesn't support remote). backend_set_t selected_engines; - std::vectorconnMd_)::iterator> found_iters; + std::vector found_iters; for (const auto &backend : *backend_list) { - auto it = data->connMd_.find(backend->getType()); - if (it == data->connMd_.end()) continue; + auto it = connMd_.find(backend->getType()); + if (it == connMd_.end()) { + continue; + } found_iters.push_back(it); selected_engines.insert(backend); } @@ -1563,29 +1557,42 @@ nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, } nixlSerDes sd; - ret = sd.addStr("Agent", data->name_); + ret = sd.addStr("Agent", name_); // Always returns SUCCESS, serdes class logs errors if necessary - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } // Only add connection info if requested via extra_params or empty dlist size_t conn_cnt = ((extra_params && extra_params->includeConnInfo) || descs.descCount() == 0) ? - found_iters.size() : 0; + found_iters.size() : + 0; ret = sd.addBuf("Conns", &conn_cnt, sizeof(conn_cnt)); - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } for (size_t i = 0; i < conn_cnt; i++) { ret = sd.addStr("t", found_iters[i]->first); - if (ret) break; + if (ret != NIXL_SUCCESS) { + break; + } ret = sd.addStr("c", found_iters[i]->second); - if (ret) break; + if (ret != NIXL_SUCCESS) { + break; + } + } + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; } - if (ret) return NIXL_ERR_UNKNOWN; ret = sd.addStr("", "MemSection"); - if (ret) return NIXL_ERR_UNKNOWN; + if (ret != NIXL_SUCCESS) { + return NIXL_ERR_UNKNOWN; + } - ret = data->localSection_.serializePartial(&sd, selected_engines, descs); - if (ret) { + ret = localSection_.serializePartial(&sd, selected_engines, descs); + if (ret != NIXL_SUCCESS) { NIXL_ERROR_FUNC << "serialization failed"; return ret; } @@ -1594,38 +1601,36 @@ nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, return NIXL_SUCCESS; } +// Single deserialization implementation, exposed via nixlMetadataContext so the +// manager/backends can load a fetched blob into the cache. nixlAgent::loadRemoteMD +// delegates here. nixl_status_t -nixlAgent::loadRemoteMD (const nixl_blob_t &remote_metadata, - std::string &agent_name) { - NIXL_TRACE_SCOPE( - trace_span, data->tracer_.get(), "nixl::loadRemoteMD", nixl::trace::Kind::Metadata); - +nixlAgentData::loadRemoteMD(const nixl_blob_t &remote_metadata, std::string &out_name) { nixlSerDes sd; nixl_blob_t conn_info; nixl_backend_t nixl_backend; nixl_status_t ret; - NIXL_LOCK_GUARD(data->lock); + NIXL_LOCK_GUARD(lock); ret = sd.importStr(remote_metadata); if (ret != NIXL_SUCCESS) { NIXL_ERROR_FUNC << "failed to deserialize remote metadata"; return NIXL_ERR_MISMATCH; } - std::string remote_agent = sd.getStr("Agent"); + const std::string remote_agent = sd.getStr("Agent"); if (remote_agent.empty()) { NIXL_ERROR_FUNC << "error in deserializing remote agent name"; return NIXL_ERR_MISMATCH; } - if (remote_agent == data->name_) { + if (remote_agent == name_) { NIXL_ERROR_FUNC << "remote agent name same as local agent, " "no need to load metadata"; return NIXL_ERR_INVALID_PARAM; } NIXL_DEBUG << "Loading remote metadata for agent: " << remote_agent; - NIXL_TRACE_ATTR(trace_span, "remote_agent", std::string_view{remote_agent}); size_t conn_cnt; ret = sd.getBuf("Conns", &conn_cnt, sizeof(conn_cnt)); @@ -1644,7 +1649,7 @@ nixlAgent::loadRemoteMD (const nixl_blob_t &remote_metadata, return NIXL_ERR_MISMATCH; } - ret = data->loadConnInfo(remote_agent, nixl_backend, conn_info); + ret = loadConnInfo(remote_agent, nixl_backend, conn_info); if (ret == NIXL_SUCCESS) { count++; } else if (ret != NIXL_ERR_NOT_SUPPORTED) { @@ -1664,37 +1669,62 @@ nixlAgent::loadRemoteMD (const nixl_blob_t &remote_metadata, return NIXL_ERR_MISMATCH; } - ret = data->loadRemoteSections(remote_agent, sd); + ret = loadRemoteSections(remote_agent, sd); if (ret != NIXL_SUCCESS) { NIXL_ERROR_FUNC << "error loading remote metadata for agent '" << remote_agent << "' with status " << ret; return ret; } - agent_name = remote_agent; + out_name = remote_agent; return NIXL_SUCCESS; } nixl_status_t -nixlAgent::invalidateRemoteMD(const std::string &remote_agent) { - NIXL_LOCK_GUARD(data->lock); +nixlAgent::getLocalPartialMD(const nixl_reg_dlist_t &descs, + nixl_blob_t &str, + const nixl_opt_args_t *extra_params) const { + // Single serialization implementation lives in nixlAgentData (the metadata + // context the manager also uses); this public entry point delegates to it. + return data->getLocalPartialMD(descs, str, extra_params); +} + +nixl_status_t +nixlAgent::loadRemoteMD(const nixl_blob_t &remote_metadata, std::string &agent_name) { + NIXL_TRACE_SCOPE( + trace_span, data->tracer_.get(), "nixl::loadRemoteMD", nixl::trace::Kind::Metadata); - if (remote_agent == data->name_) { + // Single deserialization implementation lives in nixlAgentData (the metadata + // context the manager also uses); this public entry point delegates to it. + const nixl_status_t ret = data->loadRemoteMD(remote_metadata, agent_name); + if (ret == NIXL_SUCCESS) { + NIXL_TRACE_ATTR(trace_span, "remote_agent", std::string_view{agent_name}); + } + return ret; +} + +// Single cache-eviction implementation, shared by the public method and by +// backends (P2P inbound INVL, etcd watch) through the metadata context. +nixl_status_t +nixlAgentData::invalidateRemoteMD(const std::string &remote_agent) { + NIXL_LOCK_GUARD(lock); + + if (remote_agent == name_) { NIXL_ERROR_FUNC << "remote agent same as local agent, cannot invalidate local metadata"; return NIXL_ERR_INVALID_PARAM; } nixl_status_t ret = NIXL_ERR_NOT_FOUND; - if (data->remoteSections_.erase(remote_agent) > 0) { + if (remoteSections_.erase(remote_agent) > 0) { ret = NIXL_SUCCESS; } - if (data->remoteBackends_.count(remote_agent) != 0) { - for (auto &it : data->remoteBackends_[remote_agent]) { - data->backendEngines_[it.first]->disconnect(remote_agent); + if (remoteBackends_.count(remote_agent) != 0) { + for (auto &it : remoteBackends_[remote_agent]) { + backendEngines_[it.first]->disconnect(remote_agent); } - data->remoteBackends_.erase(remote_agent); + remoteBackends_.erase(remote_agent); ret = NIXL_SUCCESS; } @@ -1707,68 +1737,102 @@ nixlAgent::invalidateRemoteMD(const std::string &remote_agent) { return ret; } +// Relocated from the former nixl_listener.cpp (shared cache helpers used by the +// load/invalidate paths). nixl_status_t -nixlAgent::sendLocalMD (const nixl_opt_args_t* extra_params) const { - nixl_blob_t myMD; - nixl_status_t ret = getLocalMD(myMD); - if (ret < 0) { - NIXL_ERROR_FUNC << "error getting local metadata with status " << ret; - return ret; +nixlAgentData::loadConnInfo(const std::string &remote_name, + const nixl_backend_t &backend, + const nixl_blob_t &conn_info) { + if (backendEngines_.count(backend) == 0) { + NIXL_DEBUG << "Agent " << name_ << " does not support a remote backend: " << backend; + return NIXL_ERR_NOT_SUPPORTED; + } + + // No need to reload same conn info, error if it changed + const auto r_it = remoteBackends_.find(remote_name); + if (r_it != remoteBackends_.end()) { + const auto rb_it = r_it->second.find(backend); + if (rb_it != r_it->second.end()) { + if (rb_it->second != conn_info) { + return NIXL_ERR_NOT_ALLOWED; + } + return NIXL_SUCCESS; + } } - // If IP is provided, use socket-based communication - if (extra_params && !extra_params->ipAddr.empty()) { - data->enqueueCommWork(std::make_tuple(SOCK_SEND, extra_params->ipAddr, extra_params->port, std::move(myMD))); - return NIXL_SUCCESS; + nixlBackendEngine *eng = backendEngines_[backend].get(); + if (!eng->supportsRemote()) { + NIXL_DEBUG << backend << " does not support remote operations"; + return NIXL_ERR_NOT_SUPPORTED; } -#if HAVE_ETCD - // If no IP is provided, use etcd (now via thread) - if (data->useEtcd_) { - data->enqueueCommWork(std::make_tuple(ETCD_SEND, default_metadata_label, 0, std::move(myMD))); - return NIXL_SUCCESS; + const nixl_status_t ret = eng->loadRemoteConnInfo(remote_name, conn_info); + if (ret != NIXL_SUCCESS) { + return ret; } - NIXL_ERROR_FUNC << "invalid parameters to be used for either socket or ETCD"; - return NIXL_ERR_INVALID_PARAM; -#else - NIXL_ERROR_FUNC - << "sendLocalMD: ETCD is not supported and socket information was not provided either"; - return NIXL_ERR_NOT_SUPPORTED; -#endif // HAVE_ETCD + + remoteBackends_[remote_name].emplace(backend, conn_info); + return NIXL_SUCCESS; } nixl_status_t -nixlAgent::sendLocalPartialMD(const nixl_reg_dlist_t &descs, - const nixl_opt_args_t* extra_params) const { - nixl_blob_t myMD; - nixl_status_t ret = getLocalPartialMD(descs, myMD, extra_params); - if (ret < 0) { - NIXL_ERROR_FUNC << "error getting local partial metadata with status " << ret; +nixlAgentData::loadRemoteSections(const std::string &remote_name, nixlSerDes &sd) { + const auto [it, inserted] = remoteSections_.try_emplace(remote_name, remote_name); + const nixl_status_t ret = it->second.loadRemoteData(&sd, backendEngines_); + // TODO: can be more graceful, if just the new MD blob was improper + if (ret != NIXL_SUCCESS) { + remoteSections_.erase(it); + remoteBackends_.erase(remote_name); return ret; } - // If IP is provided, use socket-based communication - if (extra_params && !extra_params->ipAddr.empty()) { - data->enqueueCommWork(std::make_tuple(SOCK_SEND, extra_params->ipAddr, extra_params->port, std::move(myMD))); - return NIXL_SUCCESS; + return NIXL_SUCCESS; +} + +nixl_status_t +nixlAgentData::invalidateRemoteData(const std::string &remote_name, uint64_t generation) { + lock.assertHeld(); + + if (remote_name == name_) { + NIXL_ERROR << "Agent " << name_ << " cannot invalidate itself"; + return NIXL_ERR_INVALID_PARAM; } -#if HAVE_ETCD - // If no IP is provided, use etcd (now via thread) - if (data->useEtcd_) { - if (!extra_params || extra_params->metadataLabel.empty()) { - NIXL_ERROR_FUNC << "metadata label is required for etcd send of local partial metadata"; - return NIXL_ERR_INVALID_PARAM; + const auto sec_it = remoteSections_.find(remote_name); + if (sec_it == remoteSections_.end() || sec_it->second.getGeneration() != generation) { + return NIXL_ERR_NOT_FOUND; + } + + remoteSections_.erase(sec_it); + + auto it_backends = remoteBackends_.find(remote_name); + if (it_backends != remoteBackends_.end()) { + for (auto &it : it_backends->second) { + backendEngines_[it.first]->disconnect(remote_name); } - data->enqueueCommWork(std::make_tuple(ETCD_SEND, extra_params->metadataLabel, 0, std::move(myMD))); - return NIXL_SUCCESS; + + remoteBackends_.erase(it_backends); } - NIXL_ERROR_FUNC << "invalid parameters to be used for either socket or ETCD"; - return NIXL_ERR_INVALID_PARAM; -#else - NIXL_ERROR_FUNC << "ETCD is not supported and socket information was not provided either"; - return NIXL_ERR_NOT_SUPPORTED; -#endif // HAVE_ETCD + + return NIXL_SUCCESS; +} + +nixl_status_t +nixlAgent::invalidateRemoteMD(const std::string &remote_agent) { + return data->invalidateRemoteMD(remote_agent); +} + +nixl_status_t +nixlAgent::sendLocalMD (const nixl_opt_args_t* extra_params) const { + // Metadata exchange is owned by the manager, which routes to P2P when a peer + // address is given, otherwise to the configured name-addressed backend. + return data->md_->sendLocalMD(extra_params); +} + +nixl_status_t +nixlAgent::sendLocalPartialMD(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t* extra_params) const { + return data->md_->sendLocalPartialMD(descs, extra_params); } nixl_status_t @@ -1778,49 +1842,12 @@ nixlAgent::fetchRemoteMD (const std::string remote_name, trace_span, data->tracer_.get(), "nixl::fetchRemoteMD", nixl::trace::Kind::Metadata); NIXL_TRACE_ATTR(trace_span, "remote_agent", std::string_view{remote_name}); - // If IP is provided, use socket-based communication - if (extra_params && !extra_params->ipAddr.empty()) { - data->enqueueCommWork(std::make_tuple(SOCK_FETCH, extra_params->ipAddr, extra_params->port, "")); - return NIXL_SUCCESS; - } - -#if HAVE_ETCD - // If no IP is provided, use etcd via thread with watch capability - if (data->useEtcd_) { - std::string metadata_label = extra_params && !extra_params->metadataLabel.empty() ? - extra_params->metadataLabel : - default_metadata_label; - data->enqueueCommWork(std::make_tuple(ETCD_FETCH, std::move(metadata_label), 0, remote_name)); - return NIXL_SUCCESS; - } - NIXL_ERROR_FUNC << "invalid parameters to be used for either socket or ETCD"; - return NIXL_ERR_INVALID_PARAM; -#else - NIXL_ERROR_FUNC << "ETCD is not supported and socket information was not provided either"; - return NIXL_ERR_NOT_SUPPORTED; -#endif // HAVE_ETCD + return data->md_->fetchRemoteMD(remote_name, extra_params); } nixl_status_t nixlAgent::invalidateLocalMD (const nixl_opt_args_t* extra_params) const { - // If IP is provided, use socket-based communication - if (extra_params && !extra_params->ipAddr.empty()) { - data->enqueueCommWork(std::make_tuple(SOCK_INVAL, extra_params->ipAddr, extra_params->port, "")); - return NIXL_SUCCESS; - } - -#if HAVE_ETCD - // If no IP is provided, use etcd via thread - if (data->useEtcd_) { - data->enqueueCommWork(std::make_tuple(ETCD_INVAL, "", 0, "")); - return NIXL_SUCCESS; - } - NIXL_ERROR_FUNC << "invalid parameters to be used for either socket or ETCD"; - return NIXL_ERR_INVALID_PARAM; -#else - NIXL_ERROR_FUNC << "ETCD is not supported and socket information was not provided either"; - return NIXL_ERR_NOT_SUPPORTED; -#endif // HAVE_ETCD + return data->md_->invalidateLocalMD(extra_params); } nixl_status_t diff --git a/src/core/nixl_etcd_metadata_backend.cpp b/src/core/nixl_etcd_metadata_backend.cpp new file mode 100644 index 0000000000..73eab9705a --- /dev/null +++ b/src/core/nixl_etcd_metadata_backend.cpp @@ -0,0 +1,379 @@ +/* + * 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. + */ +#if HAVE_ETCD + +#include "nixl_etcd_metadata_backend.h" + +#include "agent_data.h" +#include "nixl_types.h" +#include "common/configuration.h" +#include "common/nixl_log.h" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// Hand-rolled etcd client (connection + watchers). Moved verbatim from the +// former nixl_listener.cpp, with processInvalidatedAgents now driving the cache +// through nixlMetadataContext instead of a nixlAgent pointer. +class nixlEtcdClient { +private: + std::unique_ptr etcd; + const std::string namespace_prefix; + std::vector invalidated_agents; + std::mutex invalidated_agents_mutex; + std::unordered_map> agentWatchers; + std::chrono::microseconds watchTimeout_; + + std::string + makeKey(const std::string &agent_name, const std::string &metadata_type) { + std::stringstream ss; + ss << namespace_prefix << "/" << agent_name << "/" << metadata_type; + return ss.str(); + } + +public: + explicit nixlEtcdClient( + const std::string &my_agent_name, + const std::chrono::microseconds &timeout = std::chrono::microseconds(5000000)) + : namespace_prefix( + nixl::config::getValueDefaulted("NIXL_ETCD_NAMESPACE", + NIXL_ETCD_NAMESPACE_DEFAULT)), + watchTimeout_(timeout) { + const auto etcd_endpoints = nixl::config::getNonEmptyString("NIXL_ETCD_ENDPOINTS"); + + try { + etcd = std::make_unique(etcd_endpoints); + } + catch (const std::exception &e) { + NIXL_ERROR << "Error creating etcd client: " << e.what(); + return; + } + NIXL_DEBUG << "Created etcd client to endpoints: " << etcd_endpoints; + NIXL_DEBUG << "Using etcd namespace for agents: " << namespace_prefix; + + std::string agent_prefix = makeKey(my_agent_name, ""); + etcd::Response response = etcd->put(agent_prefix, ""); + if (!response.is_ok()) { + throw std::runtime_error("Failed to store agent " + my_agent_name + + " prefix key in etcd: " + response.error_message()); + } + } + + nixl_status_t + storeMetadataInEtcd(const std::string &agent_name, + const std::string &metadata_type, + const nixl_blob_t &metadata) { + if (!etcd) { + NIXL_ERROR << "ETCD client not available"; + return NIXL_ERR_NOT_SUPPORTED; + } + try { + std::string metadata_key = makeKey(agent_name, metadata_type); + etcd::Response response = etcd->put(metadata_key, metadata); + if (response.is_ok()) { + NIXL_DEBUG << "Successfully stored " << metadata_type + << " in etcd with key: " << metadata_key << " (rev " + << response.value().modified_index() << ")"; + return NIXL_SUCCESS; + } + NIXL_ERROR << "Failed to store " << metadata_type + << " in etcd: " << response.error_message(); + return NIXL_ERR_BACKEND; + } + catch (const std::exception &e) { + NIXL_ERROR << "Error sending " << metadata_type << " to etcd: " << e.what(); + return NIXL_ERR_BACKEND; + } + } + + nixl_status_t + removeMetadataFromEtcd(const std::string &agent_name) { + if (!etcd) { + NIXL_ERROR << "ETCD client not available"; + return NIXL_ERR_NOT_SUPPORTED; + } + try { + std::string agent_prefix = makeKey(agent_name, ""); + etcd::Response response = etcd->rmdir(agent_prefix, true); + if (response.is_ok()) { + NIXL_DEBUG << "Successfully removed " << response.values().size() + << " etcd keys for agent: " << agent_name; + return NIXL_SUCCESS; + } + NIXL_ERROR << "Warning: Failed to remove etcd keys for agent: " << agent_name << " : " + << response.error_message(); + return NIXL_ERR_BACKEND; + } + catch (const std::exception &e) { + NIXL_ERROR << "Exception removing etcd keys for agent: " << agent_name << " : " + << e.what(); + return NIXL_ERR_BACKEND; + } + } + + nixl_status_t + fetchMetadataFromEtcd(const std::string &agent_name, + const std::string &metadata_type, + nixl_blob_t &metadata) { + if (!etcd) { + NIXL_ERROR << "ETCD client not available"; + return NIXL_ERR_NOT_SUPPORTED; + } + std::string metadata_key = makeKey(agent_name, metadata_type); + try { + etcd::Response response = etcd->get(metadata_key); + if (response.is_ok()) { + metadata = response.value().as_string(); + if (metadata.empty()) { + // A present-but-empty value is never a valid MD blob (e.g. the + // agent-prefix anchor, or a key observed mid-write). Treat as + // not-found so the caller watches/retries instead of handing an + // empty blob to loadRemoteMD. + NIXL_INFO << "Fetched empty value for key: " << metadata_key + << "; treating as not found"; + return NIXL_ERR_NOT_FOUND; + } + NIXL_DEBUG << "Successfully fetched key: " << metadata_key << " (rev " + << response.value().modified_index() << ")"; + return NIXL_SUCCESS; + } + NIXL_INFO << "Failed to fetch key: " << metadata_key + << " from etcd: " << response.error_message(); + return NIXL_ERR_NOT_FOUND; + } + catch (const std::exception &e) { + NIXL_ERROR << "Error fetching key: " << metadata_key << " from etcd: " << e.what(); + return NIXL_ERR_UNKNOWN; + } + } + + nixl_status_t + waitForMetadataFromEtcd(const std::string &metadata_key, nixl_blob_t &remote_metadata) { + try { + etcd::Response response = etcd->get(metadata_key); + int64_t watch_index = response.index(); + std::promise ret_prom; + auto future = ret_prom.get_future(); + std::atomic promise_set{false}; + + auto watcher_callback = [&](etcd::Response response) -> void { + if (promise_set.exchange(true)) { + NIXL_DEBUG << "Ignoring subsequent watch event for key: " << metadata_key; + return; + } + if (!response.is_ok()) { + NIXL_ERROR << "Watch failed for key: " << metadata_key << " : " + << response.error_message(); + ret_prom.set_value(NIXL_ERR_BACKEND); + return; + } + if (response.action() == "delete") { + NIXL_ERROR << "Watch response: metadata key deleted: " << metadata_key; + ret_prom.set_value(NIXL_ERR_INVALID_PARAM); + return; + } + remote_metadata = response.value().as_string(); + NIXL_DEBUG << "Watch response: metadata key fetched: " << metadata_key; + ret_prom.set_value(NIXL_SUCCESS); + }; + + auto watcher = etcd::Watcher(*etcd, metadata_key, watch_index, watcher_callback); + + auto status = future.wait_for(watchTimeout_); + if (status == std::future_status::timeout) { + NIXL_ERROR << "Watch timed out for key: " << metadata_key; + // Cancel before returning so the callback cannot fire after the + // stack locals it captures by reference go out of scope. + watcher.Cancel(); + return NIXL_ERR_BACKEND; + } + watcher.Cancel(); + return future.get(); + } + catch (const std::exception &e) { + NIXL_ERROR << "Error watching etcd for key: " << metadata_key << " : " << e.what(); + return NIXL_ERR_BACKEND; + } + } + + nixl_status_t + fetchOrWaitForMetadataFromEtcd(const std::string &remote_agent, + const std::string &metadata_label, + nixl_blob_t &remote_metadata) { + nixl_status_t ret = fetchMetadataFromEtcd(remote_agent, metadata_label, remote_metadata); + if (ret == NIXL_SUCCESS) { + return NIXL_SUCCESS; + } + std::string metadata_key = makeKey(remote_agent, metadata_label); + NIXL_DEBUG << "Metadata not found, setting up watch for: " << metadata_key; + return waitForMetadataFromEtcd(metadata_key, remote_metadata); + } + + void + setupAgentWatcher(const std::string &agent_name) { + if (agentWatchers.find(agent_name) != agentWatchers.end()) { + return; + } + // DELETE events are enqueued to be processed in serviceEvents (can't be + // done inside the Watcher callback). + auto process_response = [this, agent_name](etcd::Response response) -> void { + if (!response.is_ok()) { + NIXL_ERROR << "Watcher failed to watch agent " << agent_name + << " from etcd: " << response.error_message(); + return; + } + NIXL_DEBUG << "Watcher received " << response.events().size() << " events from etcd"; + if (response.events().size() != 1) { + NIXL_ERROR << "Watcher agent " << agent_name + << " received unexpected number of events from etcd: " + << response.events().size(); + return; + } + const auto &event = response.events()[0]; + if (event.event_type() == etcd::Event::EventType::DELETE_) { + NIXL_DEBUG << "Watcher DELETE: " << event.kv().key() << " (rev " + << event.kv().modified_index() << ")"; + std::lock_guard lock(invalidated_agents_mutex); + invalidated_agents.push_back(agent_name); + } else { + NIXL_ERROR << "Watcher for " << event.kv().key() + << " received unexpected event from etcd: " + << static_cast(event.event_type()); + } + }; + std::string agent_prefix = makeKey(agent_name, ""); + agentWatchers[agent_name] = + std::make_unique(*etcd, agent_prefix, process_response); + } + + void + processInvalidatedAgents(nixlMetadataContext &ctx) { + std::vector tmp_invalidated_agents; + { + std::lock_guard lock(invalidated_agents_mutex); + tmp_invalidated_agents = std::move(invalidated_agents); + } + for (const auto &agent : tmp_invalidated_agents) { + NIXL_DEBUG << "Invalidated agent: " << agent; + agentWatchers.erase(agent); + nixl_status_t ret = ctx.invalidateRemoteMD(agent); + if (ret != NIXL_SUCCESS) { + NIXL_ERROR << "Failed to invalidate remote metadata for agent: " << agent << ": " + << ret; + } else { + NIXL_DEBUG << "Successfully invalidated remote metadata for agent: " << agent; + } + } + } +}; + +nixlEtcdMetadataBackend::nixlEtcdMetadataBackend(nixlMetadataContext &ctx) : ctx_(ctx) { + client_ = std::make_unique(ctx_.getName(), ctx_.getConfig().etcdWatchTimeout); +} + +nixlEtcdMetadataBackend::~nixlEtcdMetadataBackend() = default; + +std::string_view +nixlEtcdMetadataBackend::name() const { + return "ETCD"; +} + +void +nixlEtcdMetadataBackend::serviceEvents() { + client_->processInvalidatedAgents(ctx_); +} + +nixlPreparedOp +nixlEtcdMetadataBackend::prepareSendLocal(const nixl_opt_args_t * /*extra_params*/) { + nixl_blob_t blob; + const nixl_status_t ret = ctx_.getLocalMD(blob); + if (ret < 0) { + return {ret, {}}; + } + const std::string agent = ctx_.getName(); + return {NIXL_SUCCESS, [this, agent, blob = std::move(blob)]() { + (void)client_->storeMetadataInEtcd(agent, default_metadata_label, blob); + }}; +} + +nixlPreparedOp +nixlEtcdMetadataBackend::prepareSendLocalPartial(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params) { + if (!extra_params || extra_params->metadataLabel.empty()) { + NIXL_ERROR_FUNC << "metadata label is required for etcd send of local partial metadata"; + return {NIXL_ERR_INVALID_PARAM, {}}; + } + nixl_blob_t blob; + const nixl_status_t ret = ctx_.getLocalPartialMD(descs, blob, extra_params); + if (ret < 0) { + return {ret, {}}; + } + const std::string agent = ctx_.getName(); + const std::string label = extra_params->metadataLabel; + return {NIXL_SUCCESS, [this, agent, label, blob = std::move(blob)]() { + (void)client_->storeMetadataInEtcd(agent, label, blob); + }}; +} + +nixlPreparedOp +nixlEtcdMetadataBackend::prepareFetchRemote(const std::string &remote_name, + const nixl_opt_args_t *extra_params) { + const std::string label = (extra_params && !extra_params->metadataLabel.empty()) ? + extra_params->metadataLabel : + default_metadata_label; + return {NIXL_SUCCESS, [this, remote_name, label]() { + nixl_blob_t blob; + if (client_->fetchOrWaitForMetadataFromEtcd(remote_name, label, blob) != + NIXL_SUCCESS) { + NIXL_ERROR << "Failed to fetch metadata from etcd for agent: " << remote_name; + return; + } + std::string loaded_name; + const nixl_status_t ret = ctx_.loadRemoteMD(blob, loaded_name); + if (ret != NIXL_SUCCESS) { + NIXL_ERROR << "Failed to load remote metadata for agent: " << remote_name + << ": " << ret; + return; + } + if (loaded_name != remote_name) { + NIXL_ERROR << "Metadata mismatch for agent: " << remote_name + << " from md: " << loaded_name; + return; + } + NIXL_DEBUG << "Successfully loaded metadata for agent: " << remote_name; + client_->setupAgentWatcher(remote_name); + }}; +} + +nixlPreparedOp +nixlEtcdMetadataBackend::prepareInvalidateLocal(const nixl_opt_args_t * /*extra_params*/) { + const std::string agent = ctx_.getName(); + return {NIXL_SUCCESS, [this, agent]() { (void)client_->removeMetadataFromEtcd(agent); }}; +} + +#endif // HAVE_ETCD diff --git a/src/core/nixl_etcd_metadata_backend.h b/src/core/nixl_etcd_metadata_backend.h new file mode 100644 index 0000000000..3bfc80f68a --- /dev/null +++ b/src/core/nixl_etcd_metadata_backend.h @@ -0,0 +1,82 @@ +/* + * 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. + */ +/** + * @file nixl_etcd_metadata_backend.h + * @brief ETCD (centralized key/value) metadata backend. + */ +#ifndef NIXL_SRC_CORE_NIXL_ETCD_METADATA_BACKEND_H +#define NIXL_SRC_CORE_NIXL_ETCD_METADATA_BACKEND_H + +#if HAVE_ETCD + +#include "nixl_metadata_backend.h" + +#include +#include +#include + +class nixlMetadataContext; +class nixlEtcdClient; + +/** + * @class nixlEtcdMetadataBackend + * @brief Self-contained centralized-store metadata backend (etcd). + * + * Owns its own nixlEtcdClient (connection + watchers). Outbound ops reuse the + * context's serialization (getLocalMD / getLocalPartialMD) and submit the etcd + * I/O as tasks on the manager's worker thread; watch-driven invalidations are + * drained in serviceEvents(). Depends only on nixlMetadataContext, not nixlAgent. + * Selected by nixlMDManager when NIXL_ETCD_ENDPOINTS is set. + */ +class nixlEtcdMetadataBackend : public nixlMetadataBackend { +public: + explicit nixlEtcdMetadataBackend(nixlMetadataContext &ctx); + ~nixlEtcdMetadataBackend() override; + + [[nodiscard]] std::string_view + name() const override; + + [[nodiscard]] nixlPreparedOp + prepareSendLocal(const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareSendLocalPartial(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareFetchRemote(const std::string &remote_name, + const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] nixlPreparedOp + prepareInvalidateLocal(const nixl_opt_args_t *extra_params) override; + + [[nodiscard]] bool + needsWorker() const override { + return true; + } + + void + serviceEvents() override; + +private: + nixlMetadataContext &ctx_; + std::unique_ptr client_; +}; + +#endif // HAVE_ETCD + +#endif // NIXL_SRC_CORE_NIXL_ETCD_METADATA_BACKEND_H diff --git a/src/core/nixl_listener.cpp b/src/core/nixl_listener.cpp deleted file mode 100644 index 8a8efa22b9..0000000000 --- a/src/core/nixl_listener.cpp +++ /dev/null @@ -1,820 +0,0 @@ -/* - * 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 "nixl.h" -#include "common/configuration.h" -#include "common/nixl_time.h" -#include "agent_data.h" -#include "common/nixl_log.h" -#if HAVE_ETCD -#include -#include -#include -#endif // HAVE_ETCD -#include -#include -#include - -const std::string default_metadata_label = "metadata"; - -namespace { - -static const std::string invalid_label = "invalid"; - -int connectToIP(std::string ip_addr, int port) { - - struct sockaddr_in listenerAddr; - listenerAddr.sin_port = htons(port); - listenerAddr.sin_family = AF_INET; - - if (inet_pton(AF_INET, ip_addr.c_str(), &listenerAddr.sin_addr) <= 0) { - NIXL_ERROR << "inet_pton failed for ip_addr: " << ip_addr; - return -1; - } - - // Create a non-blocking socket - int ret_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0); - if (ret_fd == -1) { - NIXL_ERROR << "socket creation failed for ip_addr: " << ip_addr << " and port: " << port; - return -1; - } - - // Connect will return immediately with EINPROGRESS - int ret = connect(ret_fd, (struct sockaddr*)&listenerAddr, sizeof(listenerAddr)); - if (ret < 0 && errno != EINPROGRESS) { - close(ret_fd); - return -1; - } - - // Use poll to wait for connection with timeout - struct pollfd pfd; - pfd.fd = ret_fd; - pfd.events = POLLOUT; - pfd.revents = 0; - - ret = poll(&pfd, 1, 1000); // 1000ms timeout - if (ret <= 0) { - if (ret < 0) { - NIXL_PERROR << "poll failed for ip_addr: " << ip_addr << " and port: " << port; - } else { - NIXL_ERROR << "poll timed out for ip_addr: " << ip_addr << " and port: " << port; - } - close(ret_fd); - return -1; - } - - if (!(pfd.revents & POLLOUT)) { - NIXL_ERROR << "poll returned but socket not ready for write for ip_addr: " << ip_addr - << " and port: " << port; - close(ret_fd); - return -1; - } - - // Check if connection was successful - int error = 0; - socklen_t len = sizeof(error); - if (getsockopt(ret_fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0) { - NIXL_PERROR << "getsockopt failed for ip_addr: " << ip_addr << " and port: " << port; - close(ret_fd); - return -1; - } - - if (error != 0) { - errno = error; // For the 'PERROR'. - NIXL_PERROR << "getsockopt gave error for ip_addr: " << ip_addr << " and port: " << port; - close(ret_fd); - return -1; - } - - return ret_fd; -} - -void -sendCommMessage(int fd, const std::string& msg) { - size_t size = msg.size(); - constexpr size_t iov_size = 2; - struct iovec iov[iov_size] = { - {&size, sizeof(size)}, - {const_cast(msg.data()), msg.size()} - }; - - for (size_t i = 0, offset = 0, sent = 0; i < iov_size;) { - auto bytes = send(fd, - static_cast(iov[i].iov_base) + offset, - iov[i].iov_len - offset, - MSG_NOSIGNAL); - if (bytes < 0) { - if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { - continue; - } - - throw std::runtime_error( - absl::StrFormat("sendCommMessage(fd=%d, msg=%s) %zu/%zu bytes failed, errno=%d", - fd, - msg.c_str(), - sent, - size + sizeof(size), - errno)); - } - - offset += bytes; - sent += bytes; - if (offset == iov[i].iov_len) { - offset = 0; - ++i; - } - } -} - -bool -recvCommMessageType(int fd, void *data, size_t size, bool force = false) { - for (size_t received = 0; received < size;) { - auto bytes = recv(fd, static_cast(data) + received, size - received, 0); - if (bytes > 0) { - received += bytes; - continue; - } - if (bytes == 0 && received == 0 && !force) { - return false; - } - - if (bytes < 0) { - if (errno == EINTR) { - continue; - } - - if (errno == EAGAIN || errno == EWOULDBLOCK) { - if (!force && received == 0) { - return false; // nothing to read yet - } - - continue; - } - } - - throw std::runtime_error( - absl::StrFormat("recvCommMessage(fd=%d) %zu/%zu bytes failed ret=%d errno=%d", - fd, - received, - size, - bytes, - errno)); - } - - return true; -} - -bool -recvCommMessage(int fd, std::string &msg) { - size_t size; - if (!recvCommMessageType(fd, &size, sizeof(size))) { - return false; - } - - msg.resize(size); - return recvCommMessageType(fd, msg.data(), size, true); -} - -#if HAVE_ETCD -class nixlEtcdClient { -private: - std::unique_ptr etcd; - const std::string namespace_prefix; - std::vector invalidated_agents; - std::mutex invalidated_agents_mutex; - std::unordered_map> agentWatchers; - std::chrono::microseconds watchTimeout_; - - // Helper function to create etcd key - std::string makeKey(const std::string& agent_name, - const std::string& metadata_type) { - std::stringstream ss; - ss << namespace_prefix << "/" << agent_name << "/" << metadata_type; - return ss.str(); - } - -public: - explicit nixlEtcdClient( - const std::string &my_agent_name, - const std::chrono::microseconds &timeout = std::chrono::microseconds(5000000)) - : namespace_prefix( - nixl::config::getValueDefaulted("NIXL_ETCD_NAMESPACE", - NIXL_ETCD_NAMESPACE_DEFAULT)), - watchTimeout_(timeout) { - const auto etcd_endpoints = nixl::config::getNonEmptyString("NIXL_ETCD_ENDPOINTS"); - - try { - etcd = std::make_unique(etcd_endpoints); - } - catch (const std::exception &e) { - NIXL_ERROR << "Error creating etcd client: " << e.what(); - return; - } - NIXL_DEBUG << "Created etcd client to endpoints: " << etcd_endpoints; - NIXL_DEBUG << "Using etcd namespace for agents: " << namespace_prefix; - - std::string agent_prefix = makeKey(my_agent_name, ""); - etcd::Response response = etcd->put(agent_prefix, ""); - if (!response.is_ok()) { - throw std::runtime_error("Failed to store agent " + my_agent_name + - " prefix key in etcd: " + response.error_message()); - } - } - - // Store metadata in etcd - nixl_status_t storeMetadataInEtcd(const std::string& agent_name, - const std::string& metadata_type, - const nixl_blob_t& metadata) { - if (!etcd) { - NIXL_ERROR << "ETCD client not available"; - return NIXL_ERR_NOT_SUPPORTED; - } - - try { - std::string metadata_key = makeKey(agent_name, metadata_type); - etcd::Response response = etcd->put(metadata_key, metadata); - - if (response.is_ok()) { - NIXL_DEBUG << "Successfully stored " << metadata_type - << " in etcd with key: " << metadata_key << " (rev " - << response.value().modified_index() << ")"; - return NIXL_SUCCESS; - } else { - NIXL_ERROR << "Failed to store " << metadata_type << " in etcd: " << response.error_message(); - return NIXL_ERR_BACKEND; - } - } - catch (const std::exception &e) { - NIXL_ERROR << "Error sending " << metadata_type << " to etcd: " << e.what(); - return NIXL_ERR_BACKEND; - } - } - - // Remove all agent's metadata from etcd - nixl_status_t removeMetadataFromEtcd(const std::string& agent_name) { - if (!etcd) { - NIXL_ERROR << "ETCD client not available"; - return NIXL_ERR_NOT_SUPPORTED; - } - - try { - std::string agent_prefix = makeKey(agent_name, ""); - etcd::Response response = etcd->rmdir(agent_prefix, true); - - if (response.is_ok()) { - NIXL_DEBUG << "Successfully removed " << response.values().size() - << " etcd keys for agent: " << agent_name; - return NIXL_SUCCESS; - } else { - NIXL_ERROR << "Warning: Failed to remove etcd keys for agent: " - << agent_name << " : " << response.error_message(); - return NIXL_ERR_BACKEND; - } - } - catch (const std::exception &e) { - NIXL_ERROR << "Exception removing etcd keys for agent: " << agent_name << " : " << e.what(); - return NIXL_ERR_BACKEND; - } - } - - // Fetch metadata from etcd - nixl_status_t fetchMetadataFromEtcd(const std::string& agent_name, - const std::string& metadata_type, - nixl_blob_t& metadata) { - if (!etcd) { - NIXL_ERROR << "ETCD client not available"; - return NIXL_ERR_NOT_SUPPORTED; - } - - std::string metadata_key = makeKey(agent_name, metadata_type); - try { - etcd::Response response = etcd->get(metadata_key); - - if (response.is_ok()) { - metadata = response.value().as_string(); - NIXL_DEBUG << "Successfully fetched key: " << metadata_key - << " (rev " << response.value().modified_index() << ")"; - return NIXL_SUCCESS; - } else { - NIXL_INFO << "Failed to fetch key: " << metadata_key - << " from etcd: " << response.error_message(); - return NIXL_ERR_NOT_FOUND; - } - } - catch (const std::exception &e) { - NIXL_ERROR << "Error fetching key: " << metadata_key << " from etcd: " << e.what(); - return NIXL_ERR_UNKNOWN; - } - } - - nixl_status_t waitForMetadataFromEtcd(const std::string& metadata_key, - nixl_blob_t& remote_metadata) { - try { - - // Get current index to watch from - etcd::Response response = etcd->get(metadata_key); - int64_t watch_index = response.index(); - std::promise ret_prom; - auto future = ret_prom.get_future(); - std::atomic promise_set{false}; - - // This lambda assumes lifetime only inside this method - auto watcher_callback = [&](etcd::Response response) -> void { - if (promise_set.exchange(true)) { - NIXL_DEBUG << "Ignoring subsequent watch event for key: " << metadata_key; - return; - } - - if (!response.is_ok()) { - NIXL_ERROR << "Watch failed for key: " << metadata_key << " : " - << response.error_message(); - ret_prom.set_value(NIXL_ERR_BACKEND); - return; - } - if (response.action() == "delete") { - NIXL_ERROR << "Watch response: metadata key deleted: " << metadata_key; - ret_prom.set_value(NIXL_ERR_INVALID_PARAM); - return; - } - remote_metadata = response.value().as_string(); - NIXL_DEBUG << "Watch response: metadata key fetched: " << metadata_key; - ret_prom.set_value(NIXL_SUCCESS); - }; - - auto watcher = etcd::Watcher(*etcd, metadata_key, watch_index, watcher_callback); - - auto status = future.wait_for(watchTimeout_); - if (status == std::future_status::timeout) { - NIXL_ERROR << "Watch timed out for key: " << metadata_key; - return NIXL_ERR_BACKEND; - } - watcher.Cancel(); - return future.get(); - } - catch (const std::exception &e) { - NIXL_ERROR << "Error watching etcd for key: " << metadata_key << " : " << e.what(); - return NIXL_ERR_BACKEND; - } - } - - // Fetch metadata from etcd or wait for it to be available - nixl_status_t fetchOrWaitForMetadataFromEtcd(const std::string& remote_agent, - const std::string& metadata_label, - nixl_blob_t& remote_metadata) { - nixl_status_t ret = fetchMetadataFromEtcd(remote_agent, metadata_label, remote_metadata); - if (ret == NIXL_SUCCESS) { - return NIXL_SUCCESS; - } - - std::string metadata_key = makeKey(remote_agent, metadata_label); - NIXL_DEBUG << "Metadata not found, setting up watch for: " << metadata_key; - - return waitForMetadataFromEtcd(metadata_key, remote_metadata); - } - - // Setup a watcher for an agent's metadata invalidation if it doesn't already exist - void setupAgentWatcher(const std::string &agent_name) { - if (agentWatchers.find(agent_name) != agentWatchers.end()) { - return; - } - - // DELETE events are enqueued to be deleted in commWorker (can't be done inside the Watcher callback) - auto process_response = [this, agent_name](etcd::Response response) -> void { - if (!response.is_ok()) { - NIXL_ERROR << "Watcher failed to watch agent " << agent_name - << " from etcd: " << response.error_message(); - return; - } - NIXL_DEBUG << "Watcher received " << response.events().size() << " events from etcd"; - if (response.events().size() != 1) { - NIXL_ERROR << "Watcher agent " << agent_name << " received unexpected number of events from etcd: " - << response.events().size(); - return; - } - const auto &event = response.events()[0]; - if (event.event_type() == etcd::Event::EventType::DELETE_) { - NIXL_DEBUG << "Watcher DELETE: " << event.kv().key() - << " (rev " << event.kv().modified_index() << ")"; - std::lock_guard lock(invalidated_agents_mutex); - invalidated_agents.push_back(agent_name); - } else { - NIXL_ERROR << "Watcher for " << event.kv().key() - << " received unexpected event from etcd: " - << static_cast(event.event_type()); - } - }; - - std::string agent_prefix = makeKey(agent_name, ""); - agentWatchers[agent_name] = std::make_unique(*etcd, agent_prefix, process_response); - } - - // Process invalidated agents from watchers - void processInvalidatedAgents(nixlAgent* my_agent) { - std::vector tmp_invalidated_agents; - { - std::lock_guard lock(invalidated_agents_mutex); - tmp_invalidated_agents = std::move(invalidated_agents); - } - for (const auto &agent : tmp_invalidated_agents) { - NIXL_DEBUG << "Invalidated agent: " << agent; - agentWatchers.erase(agent); - nixl_status_t ret = my_agent->invalidateRemoteMD(agent); - if (ret != NIXL_SUCCESS) - NIXL_ERROR << "Failed to invalidate remote metadata for agent: " << agent << ": " << ret; - else - NIXL_DEBUG << "Successfully invalidated remote metadata for agent: " << agent; - } - } -}; -#endif // HAVE_ETCD - -} // unnamed namespace - -void -nixlAgentData::commWorker(nixlAgent &myAgent) noexcept { - try { - commWorkerInternal(&myAgent); - } - catch (...) { - commThreadException_ = std::current_exception(); - } -} - -void -nixlAgentData::commWorkerInternal(nixlAgent *myAgent) { -#if HAVE_ETCD - std::unique_ptr etcdClient = nullptr; - // useEtcd_ is set in nixlAgent constructor and is true if NIXL_ETCD_ENDPOINTS is set - if (useEtcd_) { - etcdClient = std::make_unique(name_, config_.etcdWatchTimeout); - } -#endif // HAVE_ETCD - - while(!(commThreadStop)) { - std::vector work_queue; - - // first, accept new connections - int new_fd = 0; - - while (new_fd != -1 && config_.useListenThread) { - new_fd = listener->acceptClient(); - nixl_socket_peer_t accepted_client; - - if(new_fd != -1){ - // need to convert fd to IP address and add to client map - sockaddr_in client_address; - socklen_t client_addrlen = sizeof(client_address); - if (getpeername(new_fd, (sockaddr*)&client_address, &client_addrlen) == 0) { - char client_ip[INET_ADDRSTRLEN]; - if (inet_ntop(AF_INET, &client_address.sin_addr, client_ip, INET_ADDRSTRLEN) == - nullptr) { - NIXL_PERROR << "inet_ntop failed for client address"; - close(new_fd); - throw std::runtime_error("inet_ntop failed for client address"); - } - accepted_client.first = std::string(client_ip); - accepted_client.second = client_address.sin_port; - } else { - throw std::runtime_error("getpeername failed"); - } - remoteSockets[accepted_client] = new_fd; - - // make new socket nonblocking - int new_flags = fcntl(new_fd, F_GETFL, 0) | O_NONBLOCK; - - if (fcntl(new_fd, F_SETFL, new_flags) == -1) - throw std::runtime_error("fcntl accept"); - - } - } - - // second, do agent commands - getCommWork(work_queue); - - for(const auto &request: work_queue) { - - // TODO: req_ip and req_port are relevant only for SOCK_*, need different request structure for ETCD_* - const auto &[req_command, req_ip, req_port, my_MD] = request; - - nixl_socket_peer_t req_sock = std::make_pair(req_ip, req_port); - - // use remote IP for socket lookup - auto client = remoteSockets.find(req_sock); - - // not connected - if (req_command < SOCK_MAX) { - if (client == remoteSockets.end()) { - int new_client = connectToIP(req_ip, req_port); - if (new_client == -1) { - NIXL_ERROR << "Listener thread could not connect to IP " << req_ip - << " and port " << req_port; - continue; - } - remoteSockets[req_sock] = new_client; - client = remoteSockets.find(req_sock); - } - } - - bool needs_disconnect = false; - switch(req_command) { - case SOCK_SEND: { - try { - sendCommMessage(client->second, "NIXLCOMM:LOAD" + my_MD); - } - catch (const std::runtime_error &e) { - NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); - needs_disconnect = true; - } - break; - } - case SOCK_FETCH: { - try { - sendCommMessage(client->second, "NIXLCOMM:SEND"); - } - catch (const std::runtime_error &e) { - NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); - needs_disconnect = true; - } - break; - } - case SOCK_INVAL: { - try { - sendCommMessage(client->second, "NIXLCOMM:INVL" + name_); - } - catch (const std::runtime_error &e) { - NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); - needs_disconnect = true; - } - break; - } -#if HAVE_ETCD - // ETCD operations using existing methods - case ETCD_SEND: - { - if (!useEtcd_) { - throw std::runtime_error("ETCD is not enabled"); - } - - // Parse request parameters - const std::string &metadata_label = req_ip; - - // Use local storeMetadataInEtcd function - const nixl_status_t ret = - etcdClient->storeMetadataInEtcd(name_, metadata_label, my_MD); - if (ret != NIXL_SUCCESS) { - NIXL_ERROR << "Failed to store metadata in etcd: " << ret; - } - break; - } - case ETCD_FETCH: - { - if (!useEtcd_) { - throw std::runtime_error("ETCD is not enabled"); - } - - const std::string &metadata_label = req_ip; - const std::string &remote_agent = my_MD; - - // First try a direct get - nixl_blob_t remote_metadata; - nixl_status_t ret = etcdClient->fetchOrWaitForMetadataFromEtcd(remote_agent, metadata_label, - remote_metadata); - if (ret != NIXL_SUCCESS) { - NIXL_ERROR << "Failed to fetch metadata from etcd: " << ret; - break; - } - - std::string remote_agent_from_md; - ret = myAgent->loadRemoteMD(remote_metadata, remote_agent_from_md); - if (ret != NIXL_SUCCESS) { - NIXL_ERROR << "Failed to load remote metadata: " << ret; - break; - } else if (remote_agent_from_md != remote_agent) { - NIXL_ERROR << "Metadata mismatch for agent: " << remote_agent - << " from md: " << remote_agent_from_md; - break; - } - NIXL_DEBUG << "Successfully loaded metadata for agent: " << remote_agent; - - etcdClient->setupAgentWatcher(remote_agent); - break; - } - case ETCD_INVAL: - { - if (!useEtcd_) { - throw std::runtime_error("ETCD is not enabled"); - } - - const nixl_status_t ret = etcdClient->removeMetadataFromEtcd(name_); - if (ret != NIXL_SUCCESS) { - NIXL_ERROR << "Failed to invalidate metadata in etcd: " << ret; - } - break; - } -#endif // HAVE_ETCD - default: - { - throw std::runtime_error("Impossible command\n"); - break; - } - } - if (needs_disconnect) { - close(client->second); - client = remoteSockets.erase(client); - } - } - - // third, do remote commands - auto socket_iter = remoteSockets.begin(); - while (socket_iter != remoteSockets.end()) { - std::string commands; - std::vector command_list; - nixl_status_t ret; - bool disconnected = false; - - try { - const bool received = recvCommMessage(socket_iter->second, commands); - if (!received) { - // No message received, but without error condition. - // Skip to the next peer - socket_iter++; - continue; - } - } - catch (const std::runtime_error &e) { - NIXL_ERROR << "Failed to receive message from peer, disconnecting: " << e.what(); - close(socket_iter->second); - socket_iter = remoteSockets.erase(socket_iter); - continue; - } - - command_list = absl::StrSplit(commands, "NIXLCOMM:"); - - for(const auto &command : command_list) { - - if(command.size() < 4) continue; - - // always just 4 chars: - std::string header = command.substr(0, 4); - - if(header == "LOAD") { - std::string remote_md = command.substr(4); - std::string remote_agent; - ret = myAgent->loadRemoteMD(remote_md, remote_agent); - if(ret != NIXL_SUCCESS) { - NIXL_ERROR << "loadRemoteMD in listener thread failed for md from peer " - << socket_iter->first.first << ":" << socket_iter->first.second - << " with error " << ret; - continue; - } - // not sure what to do with remote_agent - } else if(header == "SEND") { - nixl_blob_t my_MD; - myAgent->getLocalMD(my_MD); - - try { - sendCommMessage(socket_iter->second, std::string("NIXLCOMM:LOAD" + my_MD)); - } - catch (const std::runtime_error &e) { - NIXL_ERROR << "Failed to send message to peer, disconnecting: " << e.what(); - disconnected = true; - break; - } - } else if(header == "INVL") { - std::string remote_agent = command.substr(4); - myAgent->invalidateRemoteMD(remote_agent); - break; - } else { - NIXL_ERROR << "Received socket message with bad header" + header + " from peer " - << socket_iter->first.first << ":" << socket_iter->first.second; - } - } - - if (disconnected) { - close(socket_iter->second); - socket_iter = remoteSockets.erase(socket_iter); - } else { - socket_iter++; - } - } - -#if HAVE_ETCD - if (etcdClient) { - etcdClient->processInvalidatedAgents(myAgent); - } -#endif // HAVE_ETCD - - nixlTime::us_t start = nixlTime::getUs(); - while ((start + config_.lthrDelay) > nixlTime::getUs()) { - std::this_thread::yield(); - } - } -} - -void nixlAgentData::enqueueCommWork(nixl_comm_req_t request){ - if (agentShutdown) { - NIXL_WARN << "Agent shutting down, unable to accept new requests"; - return; - } - const std::lock_guard lock(commLock); - commQueue.push_back(std::move(request)); -} - -void nixlAgentData::getCommWork(std::vector &req_list){ - const std::lock_guard lock(commLock); - req_list = std::move(commQueue); - commQueue.clear(); -} - -nixl_status_t -nixlAgentData::loadConnInfo(const std::string &remote_name, - const nixl_backend_t &backend, - const nixl_blob_t &conn_info) { - if (backendEngines_.count(backend) == 0) { - NIXL_DEBUG << "Agent " << name_ << " does not support a remote backend: " << backend; - return NIXL_ERR_NOT_SUPPORTED; - } - - // No need to reload same conn info, error if it changed - const auto r_it = remoteBackends_.find(remote_name); - if (r_it != remoteBackends_.end()) { - const auto rb_it = r_it->second.find(backend); - if (rb_it != r_it->second.end()) { - if (rb_it->second != conn_info) { - return NIXL_ERR_NOT_ALLOWED; - } - return NIXL_SUCCESS; - } - } - - nixlBackendEngine *eng = backendEngines_[backend].get(); - if (!eng->supportsRemote()) { - NIXL_DEBUG << backend << " does not support remote operations"; - return NIXL_ERR_NOT_SUPPORTED; - } - - const nixl_status_t ret = eng->loadRemoteConnInfo(remote_name, conn_info); - if (ret != NIXL_SUCCESS) { - return ret; - } - - remoteBackends_[remote_name].emplace(backend, conn_info); - return NIXL_SUCCESS; -} - -nixl_status_t -nixlAgentData::loadRemoteSections(const std::string &remote_name, nixlSerDes &sd) { - const auto [it, inserted] = remoteSections_.try_emplace(remote_name, remote_name); - const nixl_status_t ret = it->second.loadRemoteData(&sd, backendEngines_); - // TODO: can be more graceful, if just the new MD blob was improper - if (ret != NIXL_SUCCESS) { - remoteSections_.erase(it); - remoteBackends_.erase(remote_name); - return ret; - } - - return NIXL_SUCCESS; -} - -nixl_status_t -nixlAgentData::invalidateRemoteData(const std::string &remote_name, uint64_t generation) { - lock.assertHeld(); - - if (remote_name == name_) { - NIXL_ERROR << "Agent " << name_ << " cannot invalidate itself"; - return NIXL_ERR_INVALID_PARAM; - } - - const auto sec_it = remoteSections_.find(remote_name); - if (sec_it == remoteSections_.end() || sec_it->second.getGeneration() != generation) { - return NIXL_ERR_NOT_FOUND; - } - - remoteSections_.erase(sec_it); - - auto it_backends = remoteBackends_.find(remote_name); - if (it_backends != remoteBackends_.end()) { - for (auto &it : it_backends->second) { - backendEngines_[it.first]->disconnect(remote_name); - } - - remoteBackends_.erase(it_backends); - } - - return NIXL_SUCCESS; -} diff --git a/src/core/nixl_md_manager.cpp b/src/core/nixl_md_manager.cpp new file mode 100644 index 0000000000..db2e1f9430 --- /dev/null +++ b/src/core/nixl_md_manager.cpp @@ -0,0 +1,276 @@ +/* + * 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. + */ +#include "nixl_md_manager.h" + +#include "agent_data.h" +#include "nixl_p2p_metadata_backend.h" + +#if HAVE_ETCD +#include "nixl_etcd_metadata_backend.h" +#endif + +#include "common/configuration.h" +#include "common/nixl_log.h" +#include "common/nixl_time.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// Definition of the default metadata label (declared in nixl_types.h). It used +// to live in the now-deleted nixl_listener.cpp; the manager is a natural home. +const std::string default_metadata_label = "metadata"; + +namespace { + +// The name-addressed backend for this run, chosen from the environment (null +// when none is configured, i.e. address-only / P2P). Adding a name-addressed +// transport is one class plus a branch here; it is not tied to any storage kind. +[[nodiscard]] std::unique_ptr +makeBackend([[maybe_unused]] nixlMetadataContext &ctx) { +#if HAVE_ETCD + if (nixlMDManager::etcdConfigured()) { + return std::make_unique(ctx); + } +#endif + return nullptr; +} + +// A call is address-routed (P2P) when it carries a peer address; otherwise it is +// name-addressed and handled by the configured backend. +[[nodiscard]] bool +hasAddress(const nixl_opt_args_t *extra_params) { + return extra_params && !extra_params->ipAddr.empty(); +} + +// Error when a call carries no peer address and no name-addressed backend +// (etcd) is configured. +[[nodiscard]] nixl_status_t +noTransport() { +#if HAVE_ETCD + NIXL_ERROR_FUNC << "no peer address provided and no centralized store configured " + "(set NIXL_ETCD_ENDPOINTS)"; + return NIXL_ERR_INVALID_PARAM; +#else + NIXL_ERROR_FUNC << "no peer address provided and no centralized store configured " + "(this build has no ETCD)"; + return NIXL_ERR_NOT_SUPPORTED; +#endif +} + +} // namespace + +bool +nixlMDManager::etcdConfigured() { +#if HAVE_ETCD + return nixl::config::checkExistence("NIXL_ETCD_ENDPOINTS"); +#else + return false; +#endif +} + +// The manager holds the P2P backend (always) plus an optional name-addressed +// backend, and routes each call by precedence: a peer address selects P2P, +// otherwise the configured backend. This preserves the agent's original per-call +// precedence (address wins over a configured backend). +nixlMDManager::nixlMDManager(nixlMetadataContext &ctx) + : ctx_(ctx), + p2pBackend_(std::make_unique(ctx)), + backend_(makeBackend(ctx)) {} + +nixlMDManager::~nixlMDManager() { + // Safety net: stop the worker before the backends (members) are destroyed, + // even if the agent did not call stop() explicitly. + stop(); +} + +template +nixl_status_t +nixlMDManager::route(const nixl_opt_args_t *extra_params, Prepare prepare) { + // Address wins per call: a peer address selects P2P, otherwise the configured + // name backend (which may be null when none is configured). + nixlMetadataBackend *b = hasAddress(extra_params) ? + static_cast(p2pBackend_.get()) : + backend_.get(); + if (!b) { + return noTransport(); + } + const nixlPreparedOp op = prepare(*b); // caller thread: validate + serialize + if (op.status != NIXL_SUCCESS) { + return op.status; + } + if (op.task) { + worker_.submit(std::move(op.task)); // worker thread: the transport I/O + } + return NIXL_SUCCESS; +} + +nixl_status_t +nixlMDManager::sendLocalMD(const nixl_opt_args_t *extra_params) { + return route(extra_params, + [&](nixlMetadataBackend &b) { return b.prepareSendLocal(extra_params); }); +} + +nixl_status_t +nixlMDManager::sendLocalPartialMD(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params) { + return route(extra_params, [&](nixlMetadataBackend &b) { + return b.prepareSendLocalPartial(descs, extra_params); + }); +} + +nixl_status_t +nixlMDManager::fetchRemoteMD(const std::string &remote_name, const nixl_opt_args_t *extra_params) { + return route(extra_params, [&](nixlMetadataBackend &b) { + return b.prepareFetchRemote(remote_name, extra_params); + }); +} + +nixl_status_t +nixlMDManager::invalidateLocalMD(const nixl_opt_args_t *extra_params) { + return route(extra_params, + [&](nixlMetadataBackend &b) { return b.prepareInvalidateLocal(extra_params); }); +} + +std::string_view +nixlMDManager::backendName() const noexcept { + return backend_ ? backend_->name() : p2pBackend_->name(); +} + +void +nixlMDManager::pollBackends() { + p2pBackend_->serviceEvents(); + if (backend_) { + backend_->serviceEvents(); + } +} + +void +nixlMDManager::start() { + const bool need = p2pBackend_->needsWorker() || (backend_ && backend_->needsWorker()); + if (need) { + worker_.start([this] { pollBackends(); }, ctx_.getConfig().lthrDelay); + } +} + +void +nixlMDManager::stop() { + worker_.stop(); +} + +// ---- nixlMetadataWorker ---- + +nixlMetadataWorker::~nixlMetadataWorker() { + stop(); +} + +void +nixlMetadataWorker::start(Poll poll, nixlTime::us_t delay) { + if (thread_.joinable()) { + return; + } + poll_ = std::move(poll); + delay_ = delay; + stop_.store(false); + accepting_.store(true); + thread_ = std::thread([this] { + try { + loop(); + } + catch (...) { + exception_ = std::current_exception(); + } + }); +} + +void +nixlMetadataWorker::stop() { + if (!thread_.joinable()) { + return; + } + // Stop accepting new work, then let the loop drain what is queued so a + // send/invalidate issued just before shutdown still reaches the peer/store. + accepting_.store(false); + while (true) { + { + const std::lock_guard lk(mutex_); + if (tasks_.empty()) { + break; + } + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + stop_.store(true); + thread_.join(); + if (exception_) { + try { + std::rethrow_exception(exception_); + } + catch (const std::exception &e) { + NIXL_WARN << "Metadata worker thread threw an exception: " << e.what(); + } + exception_ = nullptr; + } +} + +void +nixlMetadataWorker::submit(nixlWorkerTask task) { + if (!accepting_.load()) { + return; + } + const std::lock_guard lk(mutex_); + tasks_.push_back(std::move(task)); +} + +void +nixlMetadataWorker::loop() { + while (!stop_.load()) { + std::deque batch; + { + const std::lock_guard lk(mutex_); + batch.swap(tasks_); + } + // Isolate each unit of work: one throwing task or poll is logged and the + // worker keeps running, rather than tearing down all metadata processing. + for (auto &task : batch) { + try { + task(); + } + catch (const std::exception &e) { + NIXL_ERROR << "Metadata worker task threw an exception: " << e.what(); + } + } + try { + if (poll_) { + poll_(); + } + } + catch (const std::exception &e) { + NIXL_ERROR << "Metadata worker poll threw an exception: " << e.what(); + } + const nixlTime::us_t begin = nixlTime::getUs(); + while ((begin + delay_) > nixlTime::getUs()) { + std::this_thread::yield(); + } + } +} diff --git a/src/core/nixl_md_manager.h b/src/core/nixl_md_manager.h new file mode 100644 index 0000000000..22fd66b3f3 --- /dev/null +++ b/src/core/nixl_md_manager.h @@ -0,0 +1,183 @@ +/* + * 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. + */ +/** + * @file nixl_md_manager.h + * @brief Core-internal, agent-owned metadata manager that routes metadata + * exchange to a pluggable backend. + */ +#ifndef NIXL_SRC_CORE_NIXL_MD_MANAGER_H +#define NIXL_SRC_CORE_NIXL_MD_MANAGER_H + +#include "nixl_descriptors.h" +#include "nixl_metadata_backend.h" +#include "nixl_types.h" +#include "common/nixl_time.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +class nixlMetadataContext; + +/** + * @class nixlMetadataWorker + * @brief A single background thread that drains a task queue and calls a poll + * callback each iteration. + * + * This is where all blocking metadata I/O and background servicing runs, off the + * caller thread. It knows nothing about backends or routing; nixlMDManager gives + * it tasks to run and a poll callback to invoke. + */ +class nixlMetadataWorker { +public: + using Poll = std::function; + + nixlMetadataWorker() = default; + ~nixlMetadataWorker(); + + nixlMetadataWorker(const nixlMetadataWorker &) = delete; + nixlMetadataWorker & + operator=(const nixlMetadataWorker &) = delete; + + /** + * @brief Launch the loop (no-op if already running). Each pass runs the + * queued tasks, calls @p poll, then yields for @p delay. + */ + void + start(Poll poll, nixlTime::us_t delay); + + /** @brief Drain queued tasks, then stop and join. Idempotent. */ + void + stop(); + + /** @brief Enqueue a task; ignored once the worker is stopping/stopped. */ + void + submit(nixlWorkerTask task); + +private: + void + loop(); + + Poll poll_; + nixlTime::us_t delay_ = 0; + std::deque tasks_; + std::mutex mutex_; + std::thread thread_; + std::atomic stop_{false}; + std::atomic accepting_{true}; + std::exception_ptr exception_; +}; + +/** + * @class nixlMDManager + * @brief Core-internal: owns the metadata backends and routes each call. + * + * Built and owned by nixlAgentData (constructed unconditionally). Depends + * only on the nixlMetadataContext interface (not nixlAgent), so there is no cycle. + * Holds the address-routed backend (P2P) plus an optional name-addressed backend + * chosen from the environment; a peer address selects P2P, otherwise the name + * backend (address wins per call). Backend I/O runs on an owned nixlMetadataWorker. + */ +class nixlMDManager { +public: + explicit nixlMDManager(nixlMetadataContext &ctx); + ~nixlMDManager(); + + nixlMDManager(const nixlMDManager &) = delete; + nixlMDManager(nixlMDManager &&) = delete; + nixlMDManager & + operator=(const nixlMDManager &) = delete; + nixlMDManager & + operator=(nixlMDManager &&) = delete; + + /** + * @brief Whether the ETCD backend is selected (NIXL_ETCD_ENDPOINTS set and + * this build has ETCD support). Single source of truth: the manager + * uses it to pick the name backend, the agent to gate the comm thread. + */ + [[nodiscard]] static bool + etcdConfigured(); + + /** + * @brief Publish the full local metadata blob through the active backend. + * + * Routes to a backend's prepareSendLocal on the caller thread; any resulting + * transport work is scheduled on the worker thread. + */ + [[nodiscard]] nixl_status_t + sendLocalMD(const nixl_opt_args_t *extra_params = nullptr); + + /** @brief Publish a partial local metadata blob through the active backend. */ + [[nodiscard]] nixl_status_t + sendLocalPartialMD(const nixl_reg_dlist_t &descs, + const nixl_opt_args_t *extra_params = nullptr); + + /** @brief Initiate retrieval of a remote agent's metadata. */ + [[nodiscard]] nixl_status_t + fetchRemoteMD(const std::string &remote_name, const nixl_opt_args_t *extra_params = nullptr); + + /** @brief Withdraw our metadata through the active backend. */ + [[nodiscard]] nixl_status_t + invalidateLocalMD(const nixl_opt_args_t *extra_params = nullptr); + + /** + * @brief Name of the active metadata backend: the name backend when one is + * configured, otherwise "P2P". + */ + [[nodiscard]] std::string_view + backendName() const noexcept; + + /** + * @brief Start the worker thread if any backend needs it. Called by the + * agent once construction is complete (not during it). + */ + void + start(); + + /** @brief Drain pending tasks, stop and join the worker. Idempotent. */ + void + stop(); + +private: + // Route a call: select the backend (address wins), run its caller-thread + // prepare step, and enqueue the resulting task on the worker. `prepare` is + // invoked as `prepare(nixlMetadataBackend&) -> nixlPreparedOp`. + template + [[nodiscard]] nixl_status_t + route(const nixl_opt_args_t *extra_params, Prepare prepare); + + // Worker callback: poll each backend for background work (P2P accept/read, + // ETCD watch). Runs on the worker thread. + void + pollBackends(); + + nixlMetadataContext &ctx_; + // P2P (address-routed), always present. + const std::unique_ptr p2pBackend_; + // Name-addressed backend (etcd/future), or null when none configured. + const std::unique_ptr backend_; + // Runs backend I/O and background servicing off the caller thread. + nixlMetadataWorker worker_; +}; + +#endif // NIXL_SRC_CORE_NIXL_MD_MANAGER_H diff --git a/test/gtest/metadata_exchange.cpp b/test/gtest/metadata_exchange.cpp index c381101147..7d1e3f568f 100644 --- a/test/gtest/metadata_exchange.cpp +++ b/test/gtest/metadata_exchange.cpp @@ -398,8 +398,8 @@ TEST_F(MetadataExchangeTestFixture, SocketSendLocalAndInvalidateLocal) { { const LogIgnoreGuard lig1("poll timed out for ip_addr: " + ip_str + " and port: " + port_str); - const LogIgnoreGuard lig2("Listener thread could not connect to IP " + ip_str + - " and port " + port_str); + const LogIgnoreGuard lig2("P2P backend could not connect to IP " + ip_str + " and port " + + port_str); const LogIgnoreGuard lig3("getsockopt gave error for ip_addr: " + ip_str + " and port: " + port_str + ": No route to host"); const LogIgnoreGuard lig4("poll returned but socket not ready for write for ip_addr: " + @@ -517,20 +517,20 @@ TEST_F(MetadataExchangeTestFixture, SocketSendLocalPartialWithErrors) { unregistered_descs.addDesc(unregistered_buffer.getBlobDesc()); { + // The serialization failure is still surfaced by getLocalPartialMD; the + // public method now delegates to the manager, so the redundant wrapper + // log ("error getting local partial metadata") no longer applies. const LogIgnoreGuard lig1("getLocalPartialMD: serialization failed"); - const LogIgnoreGuard lig2("sendLocalPartialMD: error getting local partial metadata with " - "status NIXL_ERR_NOT_FOUND"); ASSERT_NE(src.agent->sendLocalPartialMD(unregistered_descs, &send_args), NIXL_SUCCESS); EXPECT_EQ(lig1.getIgnoredCount(), 1); - EXPECT_EQ(lig2.getIgnoredCount(), 1); } // Case 2: Attempt to load connection info on agent without backend { const LogIgnoreGuard lig1("loadRemoteMD: no common backend found"); - const LogIgnoreGuard lig2(std::regex("loadRemoteMD in listener thread failed for md from " + const LogIgnoreGuard lig2(std::regex("loadRemoteMD in P2P backend failed from " "peer 127.0.0.1:[0-9]+ with error NIXL_ERR_BACKEND")); ASSERT_EQ(src.agent->sendLocalPartialMD({DRAM_SEG}, &send_args), NIXL_SUCCESS); From 8cfa5674d10b0995fcdcfdb934160c246a15f161 Mon Sep 17 00:00:00 2001 From: Asaf Schwartz Date: Mon, 13 Jul 2026 19:59:08 +0300 Subject: [PATCH 3/3] test(metadata): add nixlMDManager gtests for P2P and ETCD Add md_manager.cpp covering the manager's routing and the P2P and ETCD backends (send/fetch/invalidate by peer address and by name). ETCD cases are gated on a live NIXL_ETCD_ENDPOINTS endpoint. --- test/gtest/md_manager.cpp | 272 ++++++++++++++++++++++++++++++++++++++ test/gtest/meson.build | 1 + 2 files changed, 273 insertions(+) create mode 100644 test/gtest/md_manager.cpp diff --git a/test/gtest/md_manager.cpp b/test/gtest/md_manager.cpp new file mode 100644 index 0000000000..86c1c2ca33 --- /dev/null +++ b/test/gtest/md_manager.cpp @@ -0,0 +1,272 @@ +/* + * 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. + */ +#include + +#include +#include +#include +#include +#include +#include + +#include "common.h" +#include "nixl.h" + +// Exercises the agent-owned metadata manager: with NIXL_USE_MD_MANAGER set, the +// agent's public metadata methods route through nixlMDManager + a backend (P2P +// when an address is given, else the KV/ETCD backend) instead of the inline +// path. The observable behavior must match the inline path. The ETCD cases are +// gated on NIXL_ETCD_ENDPOINTS and skip when no live store is configured. +namespace gtest::md_manager { + +class MemBuffer { +public: + explicit MemBuffer(size_t size) : vec_(size) {} + + operator uintptr_t() const { + return reinterpret_cast(vec_.data()); + } + + nixlBasicDesc + getBasicDesc() const { + return nixlBasicDesc(static_cast(*this), vec_.size(), 0); + } + + nixlBlobDesc + getBlobDesc() const { + return nixlBlobDesc(getBasicDesc(), ""); + } + +private: + std::vector vec_; +}; + +namespace { + + nixl_opt_args_t + peerArgs(const std::string &ip, int port) { + nixl_opt_args_t args; + args.ipAddr = ip; + args.port = port; + return args; + } + + // Bounded polling around checkRemoteMD: avoids fixed sleeps that make + // async assertions slow and timing-sensitive. + nixl_status_t + waitForRemoteMD(nixlAgent *agent, + const std::string &remote_name, + const nixl_xfer_dlist_t &descs, + nixl_status_t expected, + std::chrono::milliseconds timeout = std::chrono::seconds(3), + std::chrono::milliseconds interval = std::chrono::milliseconds(25)) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + nixl_status_t last = agent->checkRemoteMD(remote_name, descs); + while (last != expected && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(interval); + last = agent->checkRemoteMD(remote_name, descs); + } + return last; + } + +} // namespace + +class MDManagerFixture : public testing::Test { +protected: + struct AgentContext { + std::string name; + std::string ip = "127.0.0.1"; + int port; + nixlBackendH *backend_handle = nullptr; + std::vector buffers; + std::unique_ptr agent; + + void + createBackend() { + ASSERT_EQ(agent->createBackend("UCX", {}, backend_handle), NIXL_SUCCESS); + ASSERT_NE(backend_handle, nullptr); + } + + void + initAndRegisterBuffers(size_t count, size_t size) { + for (size_t i = 0; i < count; i++) { + buffers.emplace_back(size); + } + nixl_reg_dlist_t dlist(DRAM_SEG); + for (const auto &buf : buffers) { + dlist.addDesc(buf.getBlobDesc()); + } + const LogIgnoreGuard lig_efa_warn( + "Amazon EFA\\(s\\) were detected, but the UCX backend was configured"); + ASSERT_EQ(agent->registerMem(dlist), NIXL_SUCCESS); + } + }; + + void + SetUp() override { + // Route the agent's P2P metadata methods through the manager. + setenv("NIXL_USE_MD_MANAGER", "1", 1); + + for (int i = 0; i < AGENT_COUNT_; i++) { + AgentContext ctx; + ctx.port = PortAllocator::next_tcp_port(); + ctx.name = "mdm_agent_" + std::to_string(i); + + nixlAgentConfig cfg; + cfg.useListenThread = true; + cfg.listenPort = ctx.port; + cfg.syncMode = nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT; + ctx.agent = std::make_unique(ctx.name, cfg); + + ctx.createBackend(); + ctx.initAndRegisterBuffers(BUFF_COUNT_, BUFF_SIZE_); + + agents_.push_back(std::move(ctx)); + } + } + + void + TearDown() override { + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + agents_.clear(); + unsetenv("NIXL_USE_MD_MANAGER"); + } + + static constexpr int AGENT_COUNT_ = 2; + static constexpr size_t BUFF_COUNT_ = 4; + static constexpr size_t BUFF_SIZE_ = 1024; + + std::vector agents_; +}; + +TEST_F(MDManagerFixture, SendLocalAndInvalidateLocal) { + auto &src = agents_[0]; + auto &dst = agents_[1]; + + const nixl_opt_args_t args = peerArgs(dst.ip, dst.port); + + ASSERT_EQ(src.agent->sendLocalMD(&args), NIXL_SUCCESS); + EXPECT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_SUCCESS), NIXL_SUCCESS); + + ASSERT_EQ(src.agent->invalidateLocalMD(&args), NIXL_SUCCESS); + EXPECT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_ERR_NOT_FOUND), + NIXL_ERR_NOT_FOUND); +} + +TEST_F(MDManagerFixture, FetchRemote) { + auto &src = agents_[0]; + auto &dst = agents_[1]; + + const nixl_opt_args_t args = peerArgs(src.ip, src.port); + + ASSERT_EQ(dst.agent->fetchRemoteMD(src.name, &args), NIXL_SUCCESS); + EXPECT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_SUCCESS), NIXL_SUCCESS); +} + +// ETCD (KV) backend: a no-address metadata call routes through the manager's KV +// backend, which reuses the same ETCD comm-thread work items as the inline path. +// Gated on a live store via NIXL_ETCD_ENDPOINTS. +class MDManagerEtcdFixture : public testing::Test { +protected: + struct AgentContext { + std::string name; + nixlBackendH *backend_handle = nullptr; + std::vector buffers; + std::unique_ptr agent; + }; + + void + SetUp() override { + if (std::getenv("NIXL_ETCD_ENDPOINTS") == nullptr) { + GTEST_SKIP() << "NIXL_ETCD_ENDPOINTS not set; skipping ETCD backend tests"; + } + // No-address metadata routes through the manager (to the KV backend). + setenv("NIXL_USE_MD_MANAGER", "1", 1); + + // Unique per-run names so stale ETCD keys from earlier runs cannot leak in. + const std::string suffix = + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()); + + for (int i = 0; i < AGENT_COUNT_; i++) { + AgentContext ctx; + ctx.name = "mdm_etcd_agent_" + std::to_string(i) + "_" + suffix; + + // No listen thread: the ETCD path drives the comm thread on its own. + nixlAgentConfig cfg; + cfg.syncMode = nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT; + ctx.agent = std::make_unique(ctx.name, cfg); + + ASSERT_EQ(ctx.agent->createBackend("UCX", {}, ctx.backend_handle), NIXL_SUCCESS); + ASSERT_NE(ctx.backend_handle, nullptr); + + for (size_t b = 0; b < BUFF_COUNT_; b++) { + ctx.buffers.emplace_back(BUFF_SIZE_); + } + nixl_reg_dlist_t dlist(DRAM_SEG); + for (const auto &buf : ctx.buffers) { + dlist.addDesc(buf.getBlobDesc()); + } + const LogIgnoreGuard lig_efa_warn( + "Amazon EFA\\(s\\) were detected, but the UCX backend was configured"); + ASSERT_EQ(ctx.agent->registerMem(dlist), NIXL_SUCCESS); + + agents_.push_back(std::move(ctx)); + } + } + + void + TearDown() override { + // Drop each agent's published metadata so the shared store does not + // accumulate orphaned keys across CI runs (names are unique per run). + for (auto &ctx : agents_) { + ctx.agent->invalidateLocalMD(nullptr); + } + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + agents_.clear(); + unsetenv("NIXL_USE_MD_MANAGER"); + } + + static constexpr int AGENT_COUNT_ = 2; + static constexpr size_t BUFF_COUNT_ = 4; + static constexpr size_t BUFF_SIZE_ = 1024; + + std::vector agents_; +}; + +TEST_F(MDManagerEtcdFixture, SendAndFetchByName) { + auto &src = agents_[0]; + auto &dst = agents_[1]; + + ASSERT_EQ(src.agent->sendLocalMD(nullptr), NIXL_SUCCESS); + ASSERT_EQ(dst.agent->fetchRemoteMD(src.name, nullptr), NIXL_SUCCESS); + EXPECT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_SUCCESS), NIXL_SUCCESS); +} + +TEST_F(MDManagerEtcdFixture, InvalidateLocalRemovesRemote) { + auto &src = agents_[0]; + auto &dst = agents_[1]; + + ASSERT_EQ(src.agent->sendLocalMD(nullptr), NIXL_SUCCESS); + ASSERT_EQ(dst.agent->fetchRemoteMD(src.name, nullptr), NIXL_SUCCESS); + ASSERT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_SUCCESS), NIXL_SUCCESS); + + ASSERT_EQ(src.agent->invalidateLocalMD(nullptr), NIXL_SUCCESS); + EXPECT_EQ(waitForRemoteMD(dst.agent.get(), src.name, {DRAM_SEG}, NIXL_ERR_NOT_FOUND), + NIXL_ERR_NOT_FOUND); +} + +} // namespace gtest::md_manager diff --git a/test/gtest/meson.build b/test/gtest/meson.build index b2be5f7bb1..430ebde47f 100644 --- a/test/gtest/meson.build +++ b/test/gtest/meson.build @@ -85,6 +85,7 @@ gtest_sources = [ 'error_handling.cpp', 'test_transfer.cpp', 'metadata_exchange.cpp', + 'md_manager.cpp', 'common.cpp', 'query_mem.cpp', 'telemetry_test.cpp',