fix(gpunetio): preserve request state across ring reuse

benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp

@@ -20,6 +20,7 @@
 #include <cassert>
 #include <cctype>
 #include <chrono>
+#include <cstdlib>
 #include <cstring>
 #include <fcntl.h>
 #include <filesystem>
@@ -1777,7 +1778,10 @@ postSlot(nixlAgent *agent, xferBenchStats &thread_stats, slotState &slot) {
 static nixl_status_t
 recycleSlot(nixlAgent *agent, nixlBackendH *backend_engine, slotState &slot) {
     if (slot.req) {
-        agent->releaseXferReq(slot.req);
+        const nixl_status_t rc = agent->releaseXferReq(slot.req);
+        if (rc != NIXL_SUCCESS) {
+            return rc;
+        }
         slot.req = nullptr;
     }
     if (xferBenchConfig::prepared_xfer) {
@@ -1794,33 +1798,30 @@ recycleSlot(nixlAgent *agent, nixlBackendH *backend_engine, slotState &slot) {
     if (xferBenchConfig::reregister_mem && slot.registered) {
         nixl_status_t rc =
             deregisterIterationMem(agent, slot.local_iov, slot.remote_iov, backend_engine);
-        slot.registered = false;
         if (rc != NIXL_SUCCESS) {
             return rc;
         }
+        slot.registered = false;
     }
     return NIXL_SUCCESS;
 }
 
-// Best-effort teardown for early-exit / error paths.
+// Stop before stack unwinding can free memory that a backend still owns.
 static void
-cleanupSlots(nixlAgent *agent, nixlBackendH *backend_engine, std::vector<slotState> &slots) {
+cleanupSlots(nixlAgent *agent,
+             nixlBackendH *backend_engine,
+             std::vector<slotState> &slots,
+             xferBenchRT *runtime) {
     for (auto &slot : slots) {
-        if (slot.req) {
-            agent->releaseXferReq(slot.req);
-            slot.req = nullptr;
-        }
-        if (slot.prep_local_dlist) {
-            agent->releasedDlistH(slot.prep_local_dlist);
-            slot.prep_local_dlist = nullptr;
-        }
-        if (slot.prep_remote_dlist) {
-            agent->releasedDlistH(slot.prep_remote_dlist);
-            slot.prep_remote_dlist = nullptr;
-        }
-        if (xferBenchConfig::reregister_mem && slot.registered) {
-            deregisterIterationMem(agent, slot.local_iov, slot.remote_iov, backend_engine);
-            slot.registered = false;
+        const nixl_status_t rc = recycleSlot(agent, backend_engine, slot);
+        if (rc != NIXL_SUCCESS) {
+            std::cerr << "Failed to release transfer resources: " << nixlEnumStrings::statusStr(rc)
+                      << "; terminating before memory cleanup" << std::endl;
+#pragma omp critical(nixlbench_fatal_cleanup)
+            {
+                runtime->cleanupForExit();
+                std::_Exit(EXIT_FAILURE);
+            }
         }
     }
 }
@@ -1840,6 +1841,7 @@ execTransferLoop(nixlAgent *agent,
                  xferBenchStats &thread_stats,
                  const std::vector<xferBenchIOV> &local_iov,
                  const std::vector<xferBenchIOV> &remote_iov,
+                 xferBenchRT *runtime,
                  const std::atomic<int> *terminate_ptr = nullptr) {
     const int depth = std::min(xferBenchConfig::pipeline_depth, num_iter);
     if (depth < xferBenchConfig::pipeline_depth) {
@@ -1868,30 +1870,30 @@ execTransferLoop(nixlAgent *agent,
 
     for (int s = 0; s < depth; s++) {
         if (terminate_ptr && terminate_ptr->load()) [[unlikely]] {
-            cleanupSlots(agent, backend_engine, slots);
+            cleanupSlots(agent, backend_engine, slots, runtime);
             return -1;
         }
         nixl_status_t rc =
             prepareSlot(agent, backend_engine, op, target, params, thread_stats, slots[s]);
         if (rc != NIXL_SUCCESS) [[unlikely]] {
             std::cerr << "prepareSlot failed for slot " << s << ": "
                       << nixlEnumStrings::statusStr(rc) << std::endl;
-            cleanupSlots(agent, backend_engine, slots);
+            cleanupSlots(agent, backend_engine, slots, runtime);
             return -1;
         }
         rc = postSlot(agent, thread_stats, slots[s]);
         if (rc != NIXL_SUCCESS) [[unlikely]] {
             std::cerr << "postSlot failed for slot " << s << ": " << nixlEnumStrings::statusStr(rc)
                       << std::endl;
-            cleanupSlots(agent, backend_engine, slots);
+            cleanupSlots(agent, backend_engine, slots, runtime);
             return -1;
         }
         issued++;
     }
 
     while (completed < num_iter) {
         if (terminate_ptr && terminate_ptr->load()) [[unlikely]] {
-            cleanupSlots(agent, backend_engine, slots);
+            cleanupSlots(agent, backend_engine, slots, runtime);
             return -1;
         }
         for (int s = 0; s < depth; s++) {
@@ -1907,7 +1909,7 @@ execTransferLoop(nixlAgent *agent,
             if (rc != NIXL_SUCCESS) [[unlikely]] {
                 std::cerr << "Transfer failed on slot " << s << ": "
                           << nixlEnumStrings::statusStr(rc) << std::endl;
-                cleanupSlots(agent, backend_engine, slots);
+                cleanupSlots(agent, backend_engine, slots, runtime);
                 return -1;
             }
 
@@ -1920,7 +1922,7 @@ execTransferLoop(nixlAgent *agent,
             }
 
             if (terminate_ptr && terminate_ptr->load()) [[unlikely]] {
-                cleanupSlots(agent, backend_engine, slots);
+                cleanupSlots(agent, backend_engine, slots, runtime);
                 return -1;
             }
 
@@ -1929,14 +1931,14 @@ execTransferLoop(nixlAgent *agent,
                 if (rc != NIXL_SUCCESS) [[unlikely]] {
                     std::cerr << "recycleSlot failed for slot " << s << ": "
                               << nixlEnumStrings::statusStr(rc) << std::endl;
-                    cleanupSlots(agent, backend_engine, slots);
+                    cleanupSlots(agent, backend_engine, slots, runtime);
                     return -1;
                 }
                 rc = prepareSlot(agent, backend_engine, op, target, params, thread_stats, slots[s]);
                 if (rc != NIXL_SUCCESS) [[unlikely]] {
                     std::cerr << "prepareSlot failed on resubmit for slot " << s << ": "
                               << nixlEnumStrings::statusStr(rc) << std::endl;
-                    cleanupSlots(agent, backend_engine, slots);
+                    cleanupSlots(agent, backend_engine, slots, runtime);
                     return -1;
                 }
             }
@@ -1945,14 +1947,14 @@ execTransferLoop(nixlAgent *agent,
             if (rc != NIXL_SUCCESS) [[unlikely]] {
                 std::cerr << "postSlot failed on resubmit for slot " << s << ": "
                           << nixlEnumStrings::statusStr(rc) << std::endl;
-                cleanupSlots(agent, backend_engine, slots);
+                cleanupSlots(agent, backend_engine, slots, runtime);
                 return -1;
             }
             issued++;
         }
     }
 
-    cleanupSlots(agent, backend_engine, slots);
+    cleanupSlots(agent, backend_engine, slots, runtime);
     return 0;
 }
 
@@ -1965,6 +1967,7 @@ execTransfer(nixlAgent *agent,
              const int num_iter,
              const int num_threads,
              xferBenchStats &stats,
+             xferBenchRT *runtime,
              const std::atomic<int> *terminate_ptr = nullptr) {
     int ret = 0;
     stats.clear();
@@ -1994,6 +1997,7 @@ execTransfer(nixlAgent *agent,
                                       thread_stats,
                                       local_iov,
                                       remote_iov,
+                                      runtime,
                                       terminate_ptr);
 
         if (result != 0) [[unlikely]] {
@@ -2269,6 +2273,7 @@ xferBenchNixlWorker::transfer(size_t block_size,
                                skip,
                                xferBenchConfig::num_threads,
                                stats,
+                               rt,
                                &terminate);
         }
         if (ret < 0) {
@@ -2298,6 +2303,7 @@ xferBenchNixlWorker::transfer(size_t block_size,
                            num_iter,
                            xferBenchConfig::num_threads,
                            stats,
+                           rt,
                            &terminate);
     }
     if (ret < 0) {

src/core/nixl_agent.cpp

@@ -1382,12 +1382,9 @@ nixlAgent::queryXferBackend(const nixlXferReqH* req_hndl,
 
 nixl_status_t
 nixlAgent::releaseXferReq(nixlXferReqH *req_hndl) const {
-
     NIXL_SHARED_LOCK_GUARD(data->lock);
-    //attempt to cancel request
-    if(req_hndl->status == NIXL_IN_PROG) {
-        req_hndl->status = req_hndl->engine->checkXfer(
-                                     req_hndl->backendHandle);
+    if (req_hndl->status == NIXL_IN_PROG) {
+        req_hndl->status = req_hndl->engine->checkXfer(req_hndl->backendHandle);
         if (req_hndl->status == NIXL_SUCCESS) {
             traceXferComplete(data->tracer_.get(), *req_hndl);
             if (data->telemetry_) {
@@ -1396,24 +1393,18 @@ nixlAgent::releaseXferReq(nixlXferReqH *req_hndl) const {
         } else if (req_hndl->status < 0) {
             data->addErrorTelemetry(req_hndl->status);
         }
+    }
 
-        if(req_hndl->status == NIXL_IN_PROG) {
-
-            const nixl_status_t release_status =
-                req_hndl->engine->releaseReqH(req_hndl->backendHandle);
-
-            if (release_status < 0) {
-                NIXL_ERROR_FUNC << "backend '" << req_hndl->engine->getType()
-                                << "' could not release transfer request and returned error status "
-                                << release_status;
-                data->addErrorTelemetry(NIXL_ERR_REPOST_ACTIVE);
-                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;
+    if (req_hndl->backendHandle != nullptr && req_hndl->engine != nullptr) {
+        const nixl_status_t release_status = req_hndl->engine->releaseReqH(req_hndl->backendHandle);
+        if (release_status < 0) {
+            NIXL_ERROR_FUNC << "backend '" << req_hndl->engine->getType()
+                            << "' could not release transfer request and returned error status "
+                            << release_status;
+            data->addErrorTelemetry(NIXL_ERR_REPOST_ACTIVE);
+            return NIXL_ERR_REPOST_ACTIVE;
         }
+        req_hndl->backendHandle = nullptr;
     }
     delete req_hndl;
     return NIXL_SUCCESS;