From bc4202cc0e8faa06f2009a27b9543d5cc4694c5f Mon Sep 17 00:00:00 2001 From: Asaf Schwartz Date: Mon, 13 Jul 2026 19:56:17 +0300 Subject: [PATCH] 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