[Libfabric] Add same-agent transfer optimization with memcpy fallback (#809)

- Implement direct memcpy for transfers within the same agent to avoid unnecessary network operations
- Add debug logging for transfer path selection and agent comparison
- Improve debug messaging for same-agent transfers

---------

Signed-off-by: Arun Karthik <akkart@amazon.com>

Author: akkart-aws

src/plugins/libfabric/libfabric_backend.cpp

@@ -1067,7 +1067,40 @@ nixlLibfabricEngine::postXfer(const nixl_xfer_op_t &operation,
         NIXL_DEBUG << "Processing descriptor " << desc_idx << " GPU " << gpu_id
                    << " addr: " << transfer_addr << " size: " << transfer_size;
 
-        // Prepare and submit transfer
+        NIXL_DEBUG << "DEBUG: remote_agent='" << remote_agent << "' localAgent='" << localAgent
+                   << "'";
+
+        // Check for same-agent (local) transfer - handle with direct memcpy
+        if (remote_agent == localAgent) {
+            NIXL_DEBUG << "Same-agent transfer detected from localAgent= " << localAgent
+                       << "to remote_agent " << remote_agent << "for descriptor " << desc_idx
+                       << ", using memcpy fallback for " << transfer_size << " bytes";
+
+            // For same-agent transfers, we need to copy directly between the descriptor addresses
+            // The remote[desc_idx].addr should be the target address for the transfer
+            void *remote_addr = reinterpret_cast<void *>(remote[desc_idx].addr);
+
+            NIXL_DEBUG << "About to perform memcpy: local_addr=" << transfer_addr
+                       << " remote_addr=" << remote_addr << " size=" << transfer_size;
+
+            if (op_type == nixlLibfabricReq::WRITE) {
+                // Write: copy from local_addr to remote_addr
+                std::memcpy(remote_addr, transfer_addr, transfer_size);
+                NIXL_DEBUG << "Same-agent memcpy write completed: " << transfer_addr << " -> "
+                           << remote_addr << " (" << transfer_size << " bytes)";
+            } else {
+                // Read: copy from remote_addr to local_addr
+                std::memcpy(transfer_addr, remote_addr, transfer_size);
+                NIXL_DEBUG << "Same-agent memcpy read completed: " << remote_addr << " -> "
+                           << transfer_addr << " (" << transfer_size << " bytes)";
+            }
+
+            NIXL_DEBUG << "Successfully processed same-agent descriptor " << desc_idx
+                       << " using memcpy fallback";
+            continue; // Skip the rail manager transfer for this descriptor
+        }
+
+        // Prepare and submit transfer for remote agents
         nixl_status_t status = rail_manager.prepareAndSubmitTransfer(
             op_type,
             transfer_addr,
@@ -1098,8 +1131,14 @@ nixlLibfabricEngine::postXfer(const nixl_xfer_op_t &operation,
                << " requests from " << desc_count << " descriptors" << " with "
                << binary_notif->xfer_id_count << " total XFER_IDs";
 
-    // Adjust to actual request count after all submissions complete
-    backend_handle->adjust_total_requests(binary_notif->xfer_id_count);
+    // For same-agent transfers, we need to set the total to 0 since we bypassed all rail operations
+    if (remote_agent == localAgent) {
+        backend_handle->adjust_total_requests(0);
+        NIXL_DEBUG << "Same-agent transfer: adjusted total requests to 0 (all handled via memcpy)";
+    } else {
+        // Adjust to actual request count after all submissions complete
+        backend_handle->adjust_total_requests(binary_notif->xfer_id_count);
+    }
 
     // Send notification immediately after successful request submission
     if (opt_args && opt_args->hasNotif) {

src/utils/libfabric/libfabric_rail.cpp

@@ -307,6 +307,20 @@ nixlLibfabricRail::nixlLibfabricRail(const std::string &device, uint16_t id)
             throw std::runtime_error("fi_ep_bind av failed for rail " + std::to_string(rail_id));
         }
 
+        // Disable shared memory transfers for EFA provider to fix same-agent transfers
+        bool optval = false;
+        ret = fi_setopt(&endpoint->fid,
+                        FI_OPT_ENDPOINT,
+                        FI_OPT_SHARED_MEMORY_PERMITTED,
+                        &optval,
+                        sizeof(optval));
+        if (ret && ret != -FI_ENOSYS) {
+            NIXL_WARN << "fi_setopt FI_OPT_SHARED_MEMORY_PERMITTED failed for rail " << rail_id
+                      << ": " << fi_strerror(-ret) << " - continuing anyway";
+        } else if (ret == 0) {
+            NIXL_DEBUG << "Successfully disabled shared memory transfers for rail " << rail_id;
+        }
+
         // Enable endpoint for this rail
         ret = fi_enable(endpoint);
         if (ret) {