From d0b9d02c78045f462e5b23c8fb463a95be540281 Mon Sep 17 00:00:00 2001 From: Ilia Yastrebov Date: Mon, 5 Oct 2026 13:06:42 +0300 Subject: [PATCH 1/4] [5/N] PLUGINS/UCX: Simplify notification --- src/plugins/ucx/ucx_backend.cpp | 62 ++++++---------------- src/plugins/ucx/ucx_backend.h | 4 +- src/plugins/ucx/ucx_backend_req.h | 54 ++++++++++--------- src/plugins/ucx/ucx_thread_pool_engine.cpp | 6 +++ src/plugins/ucx/ucx_utils.cpp | 48 +++++++---------- src/plugins/ucx/ucx_utils.h | 19 +++---- 6 files changed, 79 insertions(+), 114 deletions(-) diff --git a/src/plugins/ucx/ucx_backend.cpp b/src/plugins/ucx/ucx_backend.cpp index bb08139075..e35eaa0b73 100644 --- a/src/plugins/ucx/ucx_backend.cpp +++ b/src/plugins/ucx/ucx_backend.cpp @@ -46,10 +46,6 @@ sglEnabledFromConfig() { } } // namespace -// A transfer to a single endpoint posts at most three requests: -// one data request, one flush request, and one notification request. -constexpr size_t single_ep_request_count = 3; - /**************************************** * Constructor/Destructor *****************************************/ @@ -456,18 +452,17 @@ nixlUcxEngine::sendXferSgl(nixlBackendReqH *handle) const { const ucx_connection_ptr_t &conn = sgl.conn(); auto &ep = conn->getEp(int_handle->getWorkerId()); - - int_handle->reserve(single_ep_request_count); + int_handle->init(conn, *ep); nixlUcxReq req; const nixl_status_t post_ret = sgl.post(*ep, req); - if (int_handle->append(post_ret, req, conn) != NIXL_SUCCESS) { + if (int_handle->append(post_ret, req) != NIXL_SUCCESS) { return post_ret; } nixlUcxReq flush_req; const nixl_status_t flush_ret = ep->flushEp(flush_req); - if (int_handle->append(flush_ret, flush_req, conn) != NIXL_SUCCESS) { + if (int_handle->append(flush_ret, flush_req) != NIXL_SUCCESS) { return flush_ret; } @@ -499,11 +494,10 @@ nixlUcxEngine::sendXferRange(const nixl_xfer_op_t &operation, } #endif - int_handle->reserve(single_ep_request_count); - const ucx_connection_ptr_t &conn = static_cast(remote[start_idx].metadataP)->conn; auto &ep = conn->getEp(worker_id); + int_handle->init(conn, *ep); nixl_status_t status = NIXL_SUCCESS; nixlUcxReq pending_req = nullptr; @@ -542,7 +536,7 @@ nixlUcxEngine::sendXferRange(const nixl_xfer_op_t &operation, status = NIXL_IN_PROG; } - if (int_handle->append(status, pending_req, conn) != NIXL_SUCCESS) { + if (int_handle->append(status, pending_req) != NIXL_SUCCESS) { return status; } @@ -552,7 +546,7 @@ nixlUcxEngine::sendXferRange(const nixl_xfer_op_t &operation, */ nixlUcxReq flush_req; const nixl_status_t flush_ret = ep->flushEp(flush_req); - if (int_handle->append(flush_ret, flush_req, conn) != NIXL_SUCCESS) { + if (int_handle->append(flush_ret, flush_req) != NIXL_SUCCESS) { return flush_ret; } @@ -588,16 +582,14 @@ nixlUcxEngine::postXfer(const nixl_xfer_op_t &operation, if (opt_args && opt_args->hasNotif) { if (ret == NIXL_SUCCESS) { nixlUcxReq req; - const auto rmd = static_cast(remote[0].metadataP); - const nixlUcxEp &ep = *rmd->conn->getEp(int_handle->getWorkerId()); - ret = notifSendPriv(remote_agent, opt_args->notifMsg, ep, &req); - if (int_handle->append(ret, req, rmd->conn) != NIXL_SUCCESS) { + ret = notifSendPriv(remote_agent, opt_args->notifMsg, int_handle->getEp(), &req); + if (int_handle->append(ret, req) != NIXL_SUCCESS) { return ret; } ret = int_handle->status(); } else if (ret == NIXL_IN_PROG) { - int_handle->notif.emplace(remote_agent, buildNotif(opt_args->notifMsg)); + int_handle->notif = buildNotif(opt_args->notifMsg); } } @@ -613,23 +605,17 @@ nixl_status_t nixlUcxEngine::checkXfer (nixlBackendReqH* handle) const return handle_status; } - nixlUcxBackendReqH::Notif notif(std::move(int_handle->notif).value()); + std::string msg = std::move(*int_handle->notif); int_handle->notif.reset(); if (handle_status != NIXL_SUCCESS) [[unlikely]] { return handle_status; } - const ucx_connection_ptr_t conn = getConnection(notif.agent); - if (!conn) [[unlikely]] { - return NIXL_ERR_NOT_FOUND; - } - nixlUcxReq req; - const nixlUcxEp &ep = *conn->getEp(int_handle->getWorkerId()); - const nixl_status_t status = sendNotif(std::move(notif.msg), ep, &req); + const nixl_status_t status = sendNotif(std::move(msg), int_handle->getEp(), &req); - if (int_handle->append(status, req, conn) != NIXL_SUCCESS) { + if (int_handle->append(status, req) != NIXL_SUCCESS) { return status; } @@ -667,35 +653,19 @@ nixlUcxEngine::progressLoop() { * Notifications *****************************************/ -std::unique_ptr +std::string nixlUcxEngine::buildNotif(const std::string &msg) const { nixlSerDes ser_des; ser_des.addStr("name", localAgent); ser_des.addStr("msg", msg); // TODO: replace with mpool for performance - return std::make_unique(ser_des.exportStr()); + return ser_des.exportStr(); } nixl_status_t -nixlUcxEngine::sendNotif(std::unique_ptr &&msg, const nixlUcxEp &ep, nixlUcxReq *req) { - std::string *buffer = msg.release(); - auto cleanup = [buffer, req](void *completed_request, void *ptr) { - delete buffer; - if ((req == nullptr) && (completed_request != nullptr)) { - /* Caller is not interested in the request, free it */ - ucp_request_free(completed_request); - } - }; - - return ep.sendAm(nixl::ucx::am_cb_op_t::NOTIF_STR, - nullptr, - 0, - buffer->data(), - buffer->size(), - UCP_AM_SEND_FLAG_EAGER, - req, - std::move(cleanup)); +nixlUcxEngine::sendNotif(std::string &&msg, const nixlUcxEp &ep, nixlUcxReq *req) { + return ep.sendAm(nixl::ucx::am_cb_op_t::NOTIF_STR, std::move(msg), UCP_AM_SEND_FLAG_EAGER, req); } nixl_status_t diff --git a/src/plugins/ucx/ucx_backend.h b/src/plugins/ucx/ucx_backend.h index 4de273e5f6..3c45427690 100644 --- a/src/plugins/ucx/ucx_backend.h +++ b/src/plugins/ucx/ucx_backend.h @@ -268,11 +268,11 @@ class nixlUcxEngine : public nixlBackendEngine { size_t length, const ucp_am_recv_param_t *param); - [[nodiscard]] std::unique_ptr + [[nodiscard]] std::string buildNotif(const std::string &msg) const; [[nodiscard]] static nixl_status_t - sendNotif(std::unique_ptr &&msg, const nixlUcxEp &ep, nixlUcxReq *req); + sendNotif(std::string &&msg, const nixlUcxEp &ep, nixlUcxReq *req); nixl_status_t notifSendPriv(const std::string &remote_agent, diff --git a/src/plugins/ucx/ucx_backend_req.h b/src/plugins/ucx/ucx_backend_req.h index d3586be499..693e388a93 100644 --- a/src/plugins/ucx/ucx_backend_req.h +++ b/src/plugins/ucx/ucx_backend_req.h @@ -31,17 +31,8 @@ class nixlUcxBackendReqH : public nixlBackendReqH { public: - // Notification to be sent after completion of all requests - struct Notif { - const std::string agent; - std::unique_ptr msg; - - Notif(const std::string &remote_agent, std::unique_ptr msg) - : agent(remote_agent), - msg(std::move(msg)) {} - }; - - std::optional notif; + // Notification to be sent over the bound connection after completion of all requests + std::optional notif; #ifdef HAVE_UCX_SGL_API std::optional sgl; @@ -52,13 +43,21 @@ class nixlUcxBackendReqH : public nixlBackendReqH { } void - reserve(size_t size) { - requests_.reserve(size); - NIXL_ASSERT(conn_ == nullptr); + init(const ucx_connection_ptr_t &conn, const nixlUcxEp &ep) { + NIXL_ASSERT(requests_.empty()); + requests_.reserve(max_requests); + conn_ = conn; + ep_ = &ep; + } + + [[nodiscard]] const nixlUcxEp & + getEp() const { + NIXL_ASSERT(ep_ != nullptr); + return *ep_; } [[nodiscard]] nixl_status_t - append(nixl_status_t status, nixlUcxReq req, const ucx_connection_ptr_t &conn) { + append(nixl_status_t status, nixlUcxReq req) { if (status == NIXL_IN_PROG) [[likely]] { requests_.push_back(req); } else if (status != NIXL_SUCCESS) { @@ -67,8 +66,6 @@ class nixlUcxBackendReqH : public nixlBackendReqH { return status; } - NIXL_ASSERT(conn_ == nullptr || conn_ == conn); - conn_ = conn; return NIXL_SUCCESS; } @@ -89,15 +86,13 @@ class nixlUcxBackendReqH : public nixlBackendReqH { } worker_->reqRelease(req); } - requests_.clear(); - conn_.reset(); + reset(); } [[nodiscard]] virtual nixl_status_t status() { if (requests_.empty()) { /* No pending transmissions */ - conn_.reset(); return NIXL_SUCCESS; } @@ -133,9 +128,6 @@ class nixlUcxBackendReqH : public nixlBackendReqH { } requests_.resize(incomplete_reqs); - if (requests_.empty()) { - conn_.reset(); - } return out_ret; } @@ -157,14 +149,26 @@ class nixlUcxBackendReqH : public nixlBackendReqH { } private: + // A post to a single endpoint issues at most three requests: + // one data request, one flush request, and one notification request. + static constexpr size_t max_requests = 3; + + void + reset() noexcept { + requests_.clear(); + conn_.reset(); + ep_ = nullptr; + } + [[nodiscard]] nixl_status_t checkConnection(const nixl_status_t status = NIXL_SUCCESS) const { - NIXL_ASSERT(conn_ != nullptr); - const nixl_status_t conn_status = conn_->getEp(getWorkerId())->checkTxState(); + NIXL_ASSERT(ep_ != nullptr); + const nixl_status_t conn_status = ep_->checkTxState(); return (conn_status != NIXL_SUCCESS) ? conn_status : status; } ucx_connection_ptr_t conn_; + const nixlUcxEp *ep_ = nullptr; std::vector requests_; nixlUcxWorker *worker_ = nullptr; }; diff --git a/src/plugins/ucx/ucx_thread_pool_engine.cpp b/src/plugins/ucx/ucx_thread_pool_engine.cpp index bf7a07938d..57f8686533 100644 --- a/src/plugins/ucx/ucx_thread_pool_engine.cpp +++ b/src/plugins/ucx/ucx_thread_pool_engine.cpp @@ -454,6 +454,12 @@ nixlUcxThreadPoolEngine::sendXferRange(const nixl_xfer_op_t &operation, const auto comp_handle = static_cast(int_handle); comp_handle->startXfer(); + + // Notifications of the composite request are sent over its shared worker + const ucx_connection_ptr_t &conn = + static_cast(remote[start_idx].metadataP)->conn; + comp_handle->init(conn, *conn->getEp(comp_handle->getWorkerId())); + const size_t batch_size = local.descCount(); const size_t num_chunks = comp_handle->getNumChunks(); NIXL_TRACE << "sending " << *comp_handle; diff --git a/src/plugins/ucx/ucx_utils.cpp b/src/plugins/ucx/ucx_utils.cpp index dded8524b3..2b21c941fd 100644 --- a/src/plugins/ucx/ucx_utils.cpp +++ b/src/plugins/ucx/ucx_utils.cpp @@ -17,7 +17,6 @@ #include "ucx_utils.h" -#include #include #include #include @@ -201,8 +200,8 @@ nixlUcxEp::disconnect_nb() { * Active message handling * =========================================== */ -using nixl_ucx_am_cb_ctx_t = std::pair; -using nixl_ucx_am_cb_ctx_ptr_t = std::unique_ptr; +// Payload of an in-flight AM send and whether to release the request on completion +using nixl_ucx_am_send_ctx_t = std::pair; void nixlUcxEp::sendAmCallback(void *request, ucs_status_t status, void *user_data) { @@ -210,54 +209,43 @@ nixlUcxEp::sendAmCallback(void *request, ucs_status_t status, void *user_data) { NIXL_ERROR << "UCX AM send failed with status " << status << " (" << ucs_status_string(status) << ")"; } - auto ctx = static_cast(user_data); - ctx->second(request, ctx->first); - delete ctx; + const std::unique_ptr ctx( + static_cast(user_data)); + if (ctx->second) { + ucp_request_free(request); + } } nixl_status_t nixlUcxEp::sendAm(nixl::ucx::am_cb_op_t msg_id, - void *hdr, - size_t hdr_len, - void *buffer, - size_t len, + std::string &&payload, uint32_t flags, - nixlUcxReq *req, - am_cleanup_t &&cleanup) const { + nixlUcxReq *req) const { const nixl_status_t status = checkTxState(); if (status != NIXL_SUCCESS) { // The endpoint is already in a failed state (e.g. the peer disconnected), - // so no request will be issued. Invoke the cleanup -- as the inline - // completion path below does -- so the caller's buffer is not leaked. - if (cleanup) { - cleanup(nullptr, buffer); - } + // so no request will be issued. return status; } + auto ctx = std::make_unique(std::move(payload), req == nullptr); + ucp_request_param_t param; - param.op_attr_mask = UCP_OP_ATTR_FIELD_FLAGS | UCP_OP_ATTR_FIELD_MEMORY_TYPE; + param.op_attr_mask = UCP_OP_ATTR_FIELD_FLAGS | UCP_OP_ATTR_FIELD_MEMORY_TYPE | + UCP_OP_ATTR_FIELD_CALLBACK | UCP_OP_ATTR_FIELD_USER_DATA; param.flags = flags; param.memory_type = UCS_MEMORY_TYPE_HOST; + param.cb.send = sendAmCallback; + param.user_data = ctx.get(); - nixl_ucx_am_cb_ctx_ptr_t ctx; - if (cleanup) { - ctx = std::make_unique(buffer, std::move(cleanup)); - param.op_attr_mask |= UCP_OP_ATTR_FIELD_CALLBACK | UCP_OP_ATTR_FIELD_USER_DATA; - param.cb.send = sendAmCallback; - param.user_data = ctx.get(); - } - - const ucs_status_ptr_t request = - ucp_am_send_nbx(eph, unsigned(msg_id), hdr, hdr_len, buffer, len, ¶m); + const ucs_status_ptr_t request = ucp_am_send_nbx( + eph, unsigned(msg_id), nullptr, 0, ctx->first.data(), ctx->first.size(), ¶m); if (UCS_PTR_IS_PTR(request)) { ctx.release(); if (req != nullptr) { *req = static_cast(request); } return NIXL_IN_PROG; - } else if (ctx) { - ctx->second(nullptr, ctx->first); } return nixl::ucx::ucsToNixlStatus(UCS_PTR_STATUS(request)); diff --git a/src/plugins/ucx/ucx_utils.h b/src/plugins/ucx/ucx_utils.h index 72a8ba5a0f..f417cd3336 100644 --- a/src/plugins/ucx/ucx_utils.h +++ b/src/plugins/ucx/ucx_utils.h @@ -30,8 +30,6 @@ extern "C" { #include "rkey.h" #include "ucx_enums.h" -#include "absl/strings/numbers.h" - inline constexpr std::string_view nixl_ucx_err_handling_param_name = "ucx_error_handling_mode"; // The API `ucp_context_query(ctx, &attr)` sets `UCS_MEMORY_TYPE_RDMA` in `attr.memory_types` @@ -74,18 +72,17 @@ class nixlUcxEp { nixlUcxEp & operator=(const nixlUcxEp &) = delete; - using am_cleanup_t = std::function; - - /* Active message handling */ + /** + * @brief Send an active message that owns its payload + * @param payload Data released once the send completes + * @param req Receives the UCX request of an in-progress send; if null, the request is + * released on completion + */ nixl_status_t sendAm(nixl::ucx::am_cb_op_t msg_id, - void *hdr, - size_t hdr_len, - void *buffer, - size_t len, + std::string &&payload, uint32_t flags, - nixlUcxReq *req = nullptr, - am_cleanup_t &&cleanup = nullptr) const; + nixlUcxReq *req = nullptr) const; /* Data access */ [[nodiscard]] nixl_status_t From f422d7b19054d4daffab7898212d958171dabfa5 Mon Sep 17 00:00:00 2001 From: Ilia Yastrebov Date: Mon, 5 Oct 2026 16:33:01 +0300 Subject: [PATCH 2/4] CR fixes from #2293 --- src/plugins/ucx/ucx_backend.cpp | 16 ++++++++-------- src/plugins/ucx/ucx_backend.h | 2 +- src/plugins/ucx/ucx_thread_pool_engine.cpp | 8 ++------ 3 files changed, 11 insertions(+), 15 deletions(-) diff --git a/src/plugins/ucx/ucx_backend.cpp b/src/plugins/ucx/ucx_backend.cpp index e35eaa0b73..8560d1d284 100644 --- a/src/plugins/ucx/ucx_backend.cpp +++ b/src/plugins/ucx/ucx_backend.cpp @@ -44,6 +44,13 @@ sglEnabledFromConfig() { return false; #endif } + +[[nodiscard]] ucp_err_handling_mode_t +errHandlingModeFromParams(const nixl_b_params_t *custom_params) { + const auto opt = nixl::getBackendParamOptional( + custom_params, std::string(nixl_ucx_err_handling_param_name)); + return opt ? ucx_err_mode_from_string(*opt) : UCP_ERR_HANDLING_MODE_PEER; +} } // namespace /**************************************** @@ -67,6 +74,7 @@ nixlUcxEngine::create(const nixlBackendInitParams &init_params) { nixlUcxEngine::nixlUcxEngine(const nixlBackendInitParams &init_params, size_t num_dedicated_workers) : nixlBackendEngine(&init_params), + errHandlingMode_(errHandlingModeFromParams(init_params.customParams)), sharedWorkerIndex_(1), sglEnabled_(sglEnabledFromConfig()) { std::vector devs; /* Empty vector */ @@ -85,14 +93,6 @@ nixlUcxEngine::nixlUcxEngine(const nixlBackendInitParams &init_params, size_t nu const size_t num_device_channels = nixl::getBackendParamDefaulted(custom_params, "ucx_num_device_channels", 4u); - - if (const auto opt = nixl::getBackendParamOptional( - custom_params, std::string(nixl_ucx_err_handling_param_name))) { - errHandlingMode_ = ucx_err_mode_from_string(*opt); - } else { - errHandlingMode_ = UCP_ERR_HANDLING_MODE_PEER; - } - const auto engine_config = nixl::getBackendParamDefaulted(custom_params, "engine_config", std::string()); diff --git a/src/plugins/ucx/ucx_backend.h b/src/plugins/ucx/ucx_backend.h index 3c45427690..69ed72c5b3 100644 --- a/src/plugins/ucx/ucx_backend.h +++ b/src/plugins/ucx/ucx_backend.h @@ -304,7 +304,7 @@ class nixlUcxEngine : public nixlBackendEngine { std::unique_ptr uc; std::vector> workers_; size_t numSharedWorkers_; - ucp_err_handling_mode_t errHandlingMode_; + const ucp_err_handling_mode_t errHandlingMode_; std::string workerAddr; mutable std::atomic sharedWorkerIndex_; const bool sglEnabled_; diff --git a/src/plugins/ucx/ucx_thread_pool_engine.cpp b/src/plugins/ucx/ucx_thread_pool_engine.cpp index 57f8686533..e0a056a280 100644 --- a/src/plugins/ucx/ucx_thread_pool_engine.cpp +++ b/src/plugins/ucx/ucx_thread_pool_engine.cpp @@ -271,8 +271,6 @@ class nixlUcxCompositeBackendReqH : public nixlUcxBackendReqH { std::shared_ptr sharedState_; }; -} // namespace - class nixlUcxDedicatedThread final : public nixlUcxThread { public: nixlUcxDedicatedThread(nixlUcxEngine *engine, nixlUcxDedicatedWorker &worker); @@ -348,8 +346,6 @@ class nixlUcxDedicatedThread final : public nixlUcxThread { std::jthread thread_; }; -namespace { - /** * @brief UCX worker that owns its dedicated thread. */ @@ -376,14 +372,14 @@ class nixlUcxDedicatedWorker : public nixlUcxWorker { std::optional thread_; }; -} // namespace - nixlUcxDedicatedThread::nixlUcxDedicatedThread(nixlUcxEngine *engine, nixlUcxDedicatedWorker &worker) : nixlUcxThread(engine, {&worker}), worker_(worker), thread_(startThread()) {} +} // namespace + nixlUcxThreadPoolEngine::nixlUcxThreadPoolEngine(const nixlBackendInitParams &init_params, size_t num_threads) : nixlUcxThreadEngine(init_params, num_threads), From 2755a803fd1802e21cccb8ce70cf1183783d2960 Mon Sep 17 00:00:00 2001 From: Ilia Yastrebov Date: Mon, 5 Oct 2026 17:30:49 +0200 Subject: [PATCH 3/4] CR fix --- src/plugins/ucx/ucx_backend_req.h | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/plugins/ucx/ucx_backend_req.h b/src/plugins/ucx/ucx_backend_req.h index 693e388a93..9258decf31 100644 --- a/src/plugins/ucx/ucx_backend_req.h +++ b/src/plugins/ucx/ucx_backend_req.h @@ -167,7 +167,10 @@ class nixlUcxBackendReqH : public nixlBackendReqH { return (conn_status != NIXL_SUCCESS) ? conn_status : status; } + // Keeps the connection (which owns the endpoint) alive for the lifetime + // of the request handle. ucx_connection_ptr_t conn_; + // Resolved endpoint over which data and notifications are sent. const nixlUcxEp *ep_ = nullptr; std::vector requests_; nixlUcxWorker *worker_ = nullptr; From d7ea9dca283cb6e21376bbfac89b8530a77fdfc2 Mon Sep 17 00:00:00 2001 From: Ilia Yastrebov Date: Tue, 6 Oct 2026 08:54:30 +0200 Subject: [PATCH 4/4] Removed optional from notif string --- src/plugins/ucx/ucx_backend.cpp | 9 ++++----- src/plugins/ucx/ucx_backend_req.h | 5 +++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/plugins/ucx/ucx_backend.cpp b/src/plugins/ucx/ucx_backend.cpp index 8560d1d284..c9efbc74ed 100644 --- a/src/plugins/ucx/ucx_backend.cpp +++ b/src/plugins/ucx/ucx_backend.cpp @@ -601,19 +601,18 @@ nixl_status_t nixlUcxEngine::checkXfer (nixlBackendReqH* handle) const const auto int_handle = static_cast(handle); const nixl_status_t handle_status = int_handle->status(); - if ((handle_status == NIXL_IN_PROG) || !int_handle->notif) { + if ((handle_status == NIXL_IN_PROG) || int_handle->notif.empty()) { return handle_status; } - std::string msg = std::move(*int_handle->notif); - int_handle->notif.reset(); - if (handle_status != NIXL_SUCCESS) [[unlikely]] { + int_handle->notif.clear(); return handle_status; } nixlUcxReq req; - const nixl_status_t status = sendNotif(std::move(msg), int_handle->getEp(), &req); + const nixl_status_t status = sendNotif(std::move(int_handle->notif), int_handle->getEp(), &req); + int_handle->notif.clear(); if (int_handle->append(status, req) != NIXL_SUCCESS) { return status; diff --git a/src/plugins/ucx/ucx_backend_req.h b/src/plugins/ucx/ucx_backend_req.h index 9258decf31..dec9457dbb 100644 --- a/src/plugins/ucx/ucx_backend_req.h +++ b/src/plugins/ucx/ucx_backend_req.h @@ -31,8 +31,9 @@ class nixlUcxBackendReqH : public nixlBackendReqH { public: - // Notification to be sent over the bound connection after completion of all requests - std::optional notif; + // Notification to be sent over the bound connection after completion of all requests. + // Empty if there is no pending notification. + std::string notif; #ifdef HAVE_UCX_SGL_API std::optional sgl;