GPUNETIO: progress independent QPs with CPU-local completion reporting

src/core/nixl_agent.cpp

@@ -1298,15 +1298,16 @@ nixlAgent::releaseXferReq(nixlXferReqH *req_hndl) const {
 
         if(req_hndl->status == NIXL_IN_PROG) {
 
-            req_hndl->status = req_hndl->engine->releaseReqH(
-                                         req_hndl->backendHandle);
+            const nixl_status_t release_status =
+                req_hndl->engine->releaseReqH(req_hndl->backendHandle);
 
-            if (req_hndl->status < 0) {
+            if (release_status < 0) {
                 NIXL_ERROR_FUNC << "backend '" << req_hndl->engine->getType()
                                 << "' could not release transfer request and returned error status "
-                                << req_hndl->status;
+                                << release_status;
                 return NIXL_ERR_REPOST_ACTIVE; // Might need renaming
             }
+            req_hndl->status = release_status;
             // just in case the backend doesn't set to NULL on success
             // this will prevent calling releaseReqH again in destructor
             req_hndl->backendHandle = nullptr;

src/plugins/gpunetio/gpunetio_backend.h

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * 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");
@@ -156,29 +156,29 @@ class nixlDocaEngine : public nixlBackendEngine {
     struct sockaddr oob_saddr;
     struct sockaddr oob_netmask;
     std::thread pthr;
-    uint64_t *last_rsvd_flags;
-    uint64_t *last_posted_flags;
     cudaStream_t post_stream[DOCA_POST_STREAM_NUM];
     cudaStream_t wait_stream;
     mutable std::atomic<uint32_t> xferStream;
-    mutable std::atomic<uint32_t> lastPostedReq;
 
     struct docaXferReqGpu *xferReqRingGpu;
     struct docaXferReqGpu *xferReqRingCpu;
     mutable std::atomic<uint32_t> xferRingPos;
-
-    struct docaXferCompletion *completion_list_gpu;
-    struct docaXferCompletion *completion_list_cpu;
+    mutable std::array<std::atomic_bool, DOCA_XFER_REQ_MAX> xferReqReserved_;
+    mutable std::array<uint32_t, DOCA_XFER_REQ_MAX> xferReqGenerations_{};
+    std::atomic<uint32_t> stopped_{0};
+    docaHostState *host_state_cpu_ = nullptr;
+    docaHostState *host_state_gpu_ = nullptr;
+
+    struct docaProgressState *progress_state_gpu;
+    struct docaProgressState *progress_state_cpu;
+    std::vector<struct docaQpProgress *> qp_progress_gpu_;
     uint32_t *wait_exit_gpu;
     uint32_t *wait_exit_cpu;
     struct docaNotif *notif_fill_gpu;
     struct docaNotif *notif_fill_cpu;
     struct docaNotif *notif_progress_gpu;
     struct docaNotif *notif_progress_cpu;
 
-    struct docaNotif *notif_send_gpu;
-    struct docaNotif *notif_send_cpu;
-
     // Map of agent name to saved nixlDocaConnection info
     std::unordered_map<std::string, nixlDocaConnection> remoteConnMap;
     std::unordered_map<std::string, struct nixlDocaRdmaQp *> qpMap;
@@ -190,17 +190,27 @@ class nixlDocaEngine : public nixlBackendEngine {
     class nixlDocaBckndReq : public nixlBackendReqH {
     private:
     public:
+        enum class completion_state : uint8_t { IN_PROGRESS, COMPLETING, COMPLETE };
+
         cudaStream_t stream;
         uint32_t devId;
-        uint32_t start_pos;
-        uint32_t end_pos;
+        std::vector<uint32_t> positions;
+        std::vector<uint32_t> generations;
+        doca_gpu_dev_verbs_qp *qp_data = nullptr;
         uintptr_t backendHandleGpu;
+        size_t postedCount = 0;
+        nixl_status_t postStatus = NIXL_SUCCESS;
+        std::atomic<completion_state> completionState{completion_state::IN_PROGRESS};
 
         nixlDocaBckndReq() : nixlBackendReqH() {}
 
         ~nixlDocaBckndReq() {}
     };
 
+    void
+    retireRequest(nixlDocaBckndReq *request) const;
+    void
+    markFailed() const;
     nixl_status_t
     progressThreadStart();
     void
@@ -211,8 +221,6 @@ class nixlDocaEngine : public nixlBackendEngine {
     connectClientRdmaQp(int oob_sock_client, const std::string &remote_agent);
     nixl_status_t
     nixlDocaDestroyNotif(doca_gpu *gpu, struct nixlDocaNotif *notif);
-
-    mutable std::mutex notifSendLock;
 };
 
 #endif

src/plugins/gpunetio/gpunetio_backend_aux.h

@@ -18,6 +18,7 @@
 #ifndef GPUNETIO_BACKEND_AUX_H
 #define GPUNETIO_BACKEND_AUX_H
 
+#include <array>
 #include <atomic>
 #include <cstring>
 #include <iostream>
@@ -53,15 +54,12 @@
 // Local includes
 #include "common/nixl_time.h"
 
-constexpr uint32_t DOCA_MAX_COMPLETION_INFLIGHT = 128;
-constexpr uint32_t DOCA_MAX_COMPLETION_INFLIGHT_MASK = (DOCA_MAX_COMPLETION_INFLIGHT - 1);
 constexpr uint32_t RDMA_SEND_QUEUE_SIZE = 2048;
 constexpr uint32_t RDMA_RECV_QUEUE_SIZE = (RDMA_SEND_QUEUE_SIZE * 2);
 constexpr uint32_t DOCA_POST_STREAM_NUM = 4;
 constexpr uint32_t DOCA_XFER_REQ_SIZE = 512;
 constexpr uint32_t DOCA_XFER_REQ_MAX = 32;
 constexpr uint32_t DOCA_XFER_REQ_MASK = (DOCA_XFER_REQ_MAX - 1);
-constexpr uint32_t DOCA_ENG_MAX_CONN = 20;
 constexpr uint32_t DOCA_RDMA_CM_LOCAL_PORT_SERVER = 6544;
 constexpr uint32_t VERBS_TEST_HOP_LIMIT = 255;
 
@@ -78,25 +76,79 @@ constexpr uint32_t DOCA_NOTIF_NULL = 0xFFFFFFFF;
 #endif
 
 struct docaXferReqGpu {
-    uint32_t id;
     uintptr_t lbuf[DOCA_XFER_REQ_SIZE];
     uintptr_t rbuf[DOCA_XFER_REQ_SIZE];
     size_t size[DOCA_XFER_REQ_SIZE];
     uint32_t lkey[DOCA_XFER_REQ_SIZE];
     uint32_t rkey[DOCA_XFER_REQ_SIZE];
     uint16_t num;
-    uint8_t in_use;
     uint32_t conn_idx;
     uint32_t has_notif_msg_idx;
     uint32_t msg_sz;
     uint64_t last_wqe;
+    uint64_t data_ticket;
+    uint64_t notif_wqe;
+    uint64_t notif_ticket;
+    uint32_t generation;
+    uint32_t state;
+    uint32_t data_state;
+    uint32_t notif_state;
     uintptr_t lbuf_notif;
     uint32_t lkey_notif;
-    uint64_t *last_rsvd;
-    uint64_t *last_posted;
     nixl_xfer_op_t backendOp; /* Needed only in case of GPU device transfer */
     doca_gpu_dev_verbs_qp *qp_data;
     doca_gpu_dev_verbs_qp *qp_notif;
+    struct docaQpProgress *qp_progress;
+};
+
+enum docaXferState : uint32_t {
+    DOCA_XFER_STATE_FREE,
+    DOCA_XFER_STATE_PREPARED,
+    DOCA_XFER_STATE_DATA_POSTED,
+    DOCA_XFER_STATE_NOTIF_PENDING,
+    DOCA_XFER_STATE_NOTIF_POSTED,
+    DOCA_XFER_STATE_COMPLETE,
+    DOCA_XFER_STATE_ERROR,
+};
+
+enum docaXferDataState : uint32_t {
+    DOCA_XFER_DATA_NONE,
+    DOCA_XFER_DATA_POSTED,
+    DOCA_XFER_DATA_COMPLETE,
+};
+
+enum docaXferNotifState : uint32_t {
+    DOCA_XFER_NOTIF_NONE,
+    DOCA_XFER_NOTIF_PENDING,
+    DOCA_XFER_NOTIF_POSTED,
+    DOCA_XFER_NOTIF_COMPLETE,
+};
+
+struct docaQpProgress {
+    uint32_t data_producer_lock;
+    uint32_t notif_producer_lock;
+    uint64_t next_data_ticket;
+    uint64_t head_data_ticket;
+    uint64_t next_notif_ticket;
+    uint64_t head_notif_ticket;
+};
+
+struct docaHostCompletion {
+    uint32_t generation;
+    uint32_t state;
+};
+
+struct docaHostState {
+    uint32_t failed;
+    docaHostCompletion completions[DOCA_XFER_REQ_MAX];
+};
+
+struct docaProgressState {
+    docaHostState *host;
+    uint32_t active_bitmap;
+    uint32_t progress_cursor;
+    uint32_t failed;
+    uint32_t active_generation[DOCA_XFER_REQ_MAX];
 };
 
 struct nixlDocaNotif {
@@ -110,11 +162,6 @@ struct nixlDocaNotif {
     std::unique_ptr<nixl::doca::verbs::mr> recv_mr;
 };
 
-struct docaXferCompletion {
-    uint8_t completed;
-    struct docaXferReqGpu *xferReqRingGpu;
-};
-
 struct docaNotif {
     doca_gpu_dev_verbs_qp *qp_gpu;
     uint32_t msg_lkey;
@@ -175,6 +222,7 @@ struct nixlDocaRdmaQp {
     uint32_t qpn_notif;
     uint32_t rqpn_notif;
     uint32_t remote_gid_notif;
+    struct docaQpProgress *progress_gpu;
 };
 
 struct nixlDocaEngine;
@@ -209,18 +257,27 @@ doca_error_t
 doca_kernel_write(cudaStream_t stream,
                   doca_gpu_dev_verbs_qp *qp_gpu,
                   struct docaXferReqGpu *xferReqRing,
+                  struct docaProgressState *progress_state,
+                  uint32_t *exit_flag,
                   uint32_t pos);
 doca_error_t
 doca_kernel_read(cudaStream_t stream,
                  doca_gpu_dev_verbs_qp *qp_gpu,
                  struct docaXferReqGpu *xferReqRing,
+                 struct docaProgressState *progress_state,
+                 uint32_t *exit_flag,
                  uint32_t pos);
 doca_error_t
+doca_kernel_publish_notif(cudaStream_t stream,
+                          struct docaXferReqGpu *xfer_req_ring,
+                          struct docaProgressState *progress_state,
+                          uint32_t pos);
+doca_error_t
 doca_kernel_progress(cudaStream_t stream,
-                     struct docaXferCompletion *completion_list,
+                     struct docaXferReqGpu *xfer_req_ring,
+                     struct docaProgressState *progress_state,
                      struct docaNotif *notif_fill,
                      struct docaNotif *notif_progress,
-                     struct docaNotif *notif_send_gpu,
                      uint32_t *exit_flag);
 
 #endif /* GPUNETIO_BACKEND_AUX_H */