diff --git a/src/core/agent_data.h b/src/core/agent_data.h index 3949a50ca8..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,32 +39,62 @@ 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; -class nixlAgentData { +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; +}; + +// 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_; @@ -85,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 @@ -107,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, @@ -141,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 ee542bf94d..24240a81a2 100644 --- a/src/core/meson.build +++ b/src/core/meson.build @@ -64,8 +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/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 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', 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);