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