RDMA Executor support

CMakeLists.txt

@@ -56,6 +56,14 @@ set( CMAKE_CXX_FLAGS "${flags_str}${CMAKE_CXX_FLAGS}")
 
 set( CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -O3 -L${ROCM_PATH}/lib")
 include_directories(${ROCM_PATH}/include)
+find_library(IBVERBS_LIBRARY ibverbs)
+if (IBVERBS_LIBRARY)
+    message(STATUS "Found ibverbs: ${IBVERBS_LIBRARY}")
+    link_libraries(ibverbs)    
+else()
+    add_definitions(-DNO_IBV_EXEC)
+    message(WARNING "ibverbs not found")    
+endif()
 link_libraries(numa hsa-runtime64 pthread)
 set (CMAKE_RUNTIME_OUTPUT_DIRECTORY .)
 add_executable(TransferBench src/client/Client.cpp)

Makefile

@@ -21,6 +21,15 @@ NVFLAGS  = -x cu -lnuma -arch=native
 COMMON_FLAGS = -O3 -I./src/header -I./src/client -I./src/client/Presets
 LDFLAGS += -lpthread
 
+# Compile RDMA executor if IBVerbs is found in the Dynamic Linker cache
+ifneq ("$(shell ldconfig -p | grep -c ibverbs)", "0")	
+	LDFLAGS += -libverbs
+	NVFLAGS += -libverbs
+else 
+	CXXFLAGS += -DNO_IBV_EXEC
+	NVFLAGS += -DNO_IBV_EXEC
+endif
+
 all: $(EXE)
 
 TransferBench: ./src/client/Client.cpp $(shell find -regex ".*\.\hpp")

src/client/Client.cpp

@@ -231,13 +231,25 @@ void PrintResults(EnvVars const& ev, int const testNum,
       if (t.exeSubIndex != -1)
         sprintf(exeSubIndexStr, ".%d", t.exeSubIndex);
 
-      printf("     Transfer %02d  %c %7.3f GB/s %c %8.3f ms %c %12lu bytes %c %s -> %s%02d%s:%03d -> %s\n",
-             idx,                    sep,
-             r.avgBandwidthGbPerSec, sep,
-             r.avgDurationMsec,      sep,
-             r.numBytes,             sep,
-             MemDevicesToStr(t.srcs).c_str(), ExeTypeName[exeType], exeIndex,
-             exeSubIndexStr, t.numSubExecs, MemDevicesToStr(t.dsts).c_str());
+      if(exeType == EXE_IBV || exeType == EXE_IBV_NEAREST) {
+        printf("      Transfer %02d %c %7.3f GB/s %c %8.3f ms %c %12lu bytes %c %s -> %s%02d:%s%02d -> %s\n",
+              idx,                    sep,
+              r.avgBandwidthGbPerSec, sep,
+              r.avgDurationMsec,      sep,
+              r.numBytes,             sep,
+              MemDevicesToStr(t.srcs).c_str(),
+              ExeTypeName[exeType], exeIndex,
+              ExeTypeName[exeType], t.exeDstIndex,
+              MemDevicesToStr(t.dsts).c_str());
+      } else {
+        printf("     Transfer %02d  %c %7.3f GB/s %c %8.3f ms %c %12lu bytes %c %s -> %s%02d%s:%03d -> %s\n",
+              idx,                    sep,
+              r.avgBandwidthGbPerSec, sep,
+              r.avgDurationMsec,      sep,
+              r.numBytes,             sep,
+              MemDevicesToStr(t.srcs).c_str(), ExeTypeName[exeType], exeIndex,
+              exeSubIndexStr, t.numSubExecs, MemDevicesToStr(t.dsts).c_str());
+      }
 
       // Show per-iteration timing information
       if (ev.showIterations) {

src/client/Client.hpp

@@ -30,7 +30,7 @@ THE SOFTWARE.
 
 size_t const DEFAULT_BYTES_PER_TRANSFER = (1<<26);
 
-char const ExeTypeName[4][4] = {"CPU", "GPU", "DMA", "IBV"};
+char const ExeTypeName[5][4] = {"CPU", "GPU", "DMA", "IBV", "IBV"};
 
 // Display detected hardware
 void DisplayTopology(bool outputToCsv);

src/client/EnvVars.hpp

@@ -100,6 +100,13 @@ class EnvVars
   int outputToCsv;                   // Output in CSV format
   int samplingFactor;                // Affects how many different values of N are generated (when N set to 0)
 
+  // Rdma options
+  int ibGidIndex;                    // GID Index for RoCE NICs
+  int roceVersion;                   // RoCE version number
+  int ipAddressFamily;               // IP Address Famliy
+  uint8_t ibPort;                    // NIC port number to be used
+  std::string closestNicStr;            // Holds the user-specified list of closest NICs
+
   // Developer features
   int gpuMaxHwQueues;                // Tracks GPU_MAX_HW_QUEUES environment variable
 
@@ -146,8 +153,15 @@ class EnvVars
     useSingleStream   = GetEnvVar("USE_SINGLE_STREAM"   , 1);
     validateDirect    = GetEnvVar("VALIDATE_DIRECT"     , 0);
     validateSource    = GetEnvVar("VALIDATE_SOURCE"     , 0);
+
+    ibGidIndex        = GetEnvVar("IB_GID_INDEX"        ,-1);
+    ibPort            = GetEnvVar("IB_PORT_NUMBER"      , 1);
+    roceVersion       = GetEnvVar("ROCE_VERSION"        , 2);
+    ipAddressFamily   = GetEnvVar("IP_ADDRESS_FAMILY"   , 4);
+    closestNicStr     = GetEnvVar("CLOSEST_NIC"         , "");
 
     gpuMaxHwQueues    = GetEnvVar("GPU_MAX_HW_QUEUES"   , 4);
+
 
     // Check for fill pattern
     char* pattern = getenv("FILL_PATTERN");
@@ -299,6 +313,11 @@ class EnvVars
     printf(" USE_SINGLE_STREAM - Use a single stream per GPU GFX executor instead of stream per Transfer\n");
     printf(" VALIDATE_DIRECT   - Validate GPU destination memory directly instead of staging GPU memory on host\n");
     printf(" VALIDATE_SOURCE   - Validate GPU src memory immediately after preparation\n");
+    printf(" IB_GID_INDEX      - Required for RoCE NICs (default=3)\n");
+    printf(" IB_PORT_NUMBER    - RDMA port count for RDMA NIC (default=1)\n");
+    printf(" IP_ADDRESS_FAMILY - IP address family (4=v4, 6=v6, default=v4)\n");
+    printf(" ROCE_VERSION      - RoCE version (default=2)\n");
+    printf(" CLOSEST_NIC       - Comma seperated list of per-GPU closest NIC (default=auto)\n");
   }
 
   void Print(std::string const& name, int32_t const value, const char* format, ...) const
@@ -381,6 +400,16 @@ class EnvVars
           "Running in %s mode", useInteractive ? "interactive" : "non-interactive");
     Print("USE_SINGLE_STREAM", useSingleStream,
           "Using single stream per GFX %s", useSingleStream ? "device" : "Transfer");
+    Print("IB_GID_INDEX", ibGidIndex,
+          "RoCE GID index is set to %s", (ibGidIndex < 0 ? "auto" : std::to_string(ibGidIndex).c_str()));
+    Print("IB_PORT_NUMBER", ibPort,
+          "IB port number is set to %d", ibPort);
+    Print("ROCE_VERSION", roceVersion,
+          "RoCE version is set to %d", roceVersion);
+    Print("IP_ADDRESS_FAMILY", ipAddressFamily,
+          "IP address family is set to IPv%d", ipAddressFamily);
+    Print("CLOSEST_NIC", (closestNicStr == "" ? "auto" : "user-input"),
+          "Per-GPU closest NIC is set as %s", (closestNicStr == "" ? "auto" : closestNicStr.c_str()));
 
     if (getenv("XCC_PREF_TABLE")) {
       printf("%36s: Preferred XCC Table (XCC_PREF_TABLE)\n", "");
@@ -479,6 +508,29 @@ class EnvVars
     cfg.gfx.useSingleTeam          = gfxSingleTeam;
     cfg.gfx.waveOrder              = gfxWaveOrder;
 
+    cfg.rdma.ibGidIndex            = ibGidIndex;
+    cfg.rdma.ibPort                = ibPort;
+    cfg.rdma.ipAddressFamily       = ipAddressFamily;
+    cfg.rdma.roceVersion           = roceVersion;
+    std::vector<int> closestNics;
+    if(closestNicStr != "") {      
+      std::stringstream ss(closestNicStr);
+      std::string item;
+      while (std::getline(ss, item, ',')) {
+        try {
+          int nic = std::stoi(item);
+          if (nic < 0) {        
+            printf("[ERROR]: NIC index cannot be negative\n");
+            exit(1);
+          }
+          closestNics.push_back(nic);
+        } catch (const std::invalid_argument& e) {
+          printf("[ERROR] Invalid NIC index: %s\n", item.c_str());
+          exit(1);
+        }
+      }
+      TransferBench::SetClosestNics(closestNics);
+    }
     return cfg;
   }
 };

src/client/Topology.hpp

@@ -24,6 +24,11 @@ THE SOFTWARE.
 
 #include "TransferBench.hpp"
 
+#ifndef NO_IBV_EXEC
+#include <infiniband/verbs.h>
+#include <filesystem>
+#endif
+
 static int RemappedCpuIndex(int origIdx)
 {
   static std::vector<int> remappingCpu;
@@ -38,21 +43,107 @@ static int RemappedCpuIndex(int origIdx)
   return remappingCpu[origIdx];
 }
 
+static void PrintNicToGPUTopo(bool outputToCsv) 
+{
+#ifndef NO_IBV_EXEC    
+    if (outputToCsv) {
+      printf("Device Index,Device Name,Port Active,Closest GPU(s),PCIe Bus ID\n");
+    }
+    else {
+      printf("Device Index | Device Name | Port Active | Closest GPU(s) | PCIe Bus ID\n");
+      printf("-------------+-------------+-------------+----------------+------------\n");
+    }
+    std::vector<std::string> devBusIds;
+    std::vector<std::string> devNames;
+    std::vector<bool> devPortsActive;
+    int devCount; 
+    struct ibv_device **deviceList = ibv_get_device_list(&devCount);
+    if (deviceList && devCount > 0) {
+      devBusIds.resize(devCount, "");
+      devNames.resize(devCount, "");
+      devPortsActive.resize(devCount, false);
+      for (int i = 0; i < devCount; ++i) {
+        struct ibv_context *ctx = ibv_open_device(deviceList[i]);
+        if (ctx) {
+          struct ibv_device_attr deviceAttr;
+          if (ibv_query_device(ctx, &deviceAttr) == 0) {
+            devNames[i] = deviceList[i]->name;
+            for (int port = 1; port <= deviceAttr.phys_port_cnt; ++port) {
+              struct ibv_port_attr portAttr;
+              if (ibv_query_port(ctx, port, &portAttr) == 0 && portAttr.state == IBV_PORT_ACTIVE) {
+                devPortsActive[i] = true;
+                break;              
+              }
+            }
+          }
+
+          std::string devicePath(deviceList[i]->dev_path);
+          if (std::filesystem::exists(devicePath))
+          {
+            std::string pciPath = std::filesystem::canonical(devicePath + "/device").string();
+            std::size_t pos = pciPath.find_last_of('/');
+            if (pos != std::string::npos) {
+              std::string nicBusId = pciPath.substr(pos + 1);
+              devBusIds[i] = nicBusId;              
+            }
+          }
+          ibv_close_device(ctx);
+        }
+      }
+      ibv_free_device_list(deviceList);
+    }
+    for (int i = 0; i < devBusIds.size(); ++i)
+    {
+      std::string nicDevice = devNames[i];
+      bool portActive = devPortsActive[i];
+
+      auto closestGpus = GetClosestGpusToNic(i);
+      std::string closestGpusStr;
+      for (size_t j = 0; j < closestGpus.size(); ++j) {
+        closestGpusStr += std::to_string(closestGpus[j]);
+        if (j < closestGpus.size() - 1) {
+          closestGpusStr += ",";
+        }
+      }
+
+      if (outputToCsv) {
+        printf("%d,%s,%s,%s,%s\n", 
+               i, 
+               nicDevice.c_str(), 
+               portActive ? "Yes" : "No", 
+               closestGpusStr.c_str(), 
+               devBusIds[i].c_str());
+      }
+      else {
+        printf("%-12d | %-11s | %-11s | %-13s | %-11s\n", 
+               i, 
+               nicDevice.c_str(), 
+               portActive ? "Yes" : "No", 
+               closestGpusStr.c_str(), 
+               devBusIds[i].c_str());
+      }
+    }
+    printf("\n");
+#endif
+}
+
 void DisplayTopology(bool outputToCsv)
 {
   int numCpus = TransferBench::GetNumExecutors(EXE_CPU);
   int numGpus = TransferBench::GetNumExecutors(EXE_GPU_GFX);
-
+  int numNics = TransferBench::GetNumExecutors(EXE_IBV);
   char sep = (outputToCsv ? ',' : '|');
 
   if (outputToCsv) {
     printf("NumCpus,%d\n", numCpus);
     printf("NumGpus,%d\n", numGpus);
+    printf("NumNics,%d\n", numNics);
   } else {
     printf("\nDetected Topology:\n");
     printf("==================\n");
     printf("  %d configured CPU NUMA node(s) [%d total]\n", numCpus, numa_max_node() + 1);
     printf("  %d GPU device(s)\n", numGpus);
+    printf("  %d Supported NIC device(s)\n", numNics);
   }
 
   // Print out detected CPU topology
@@ -91,8 +182,10 @@ void DisplayTopology(bool outputToCsv)
   }
   printf("\n");
 
-  // Print out detected GPU topology
+  // Print out detected NIC topology
+  PrintNicToGPUTopo(outputToCsv);  
 
+  // Print out detected GPU topology
 #if defined(__NVCC__)
   for (int i = 0; i < numGpus; i++) {
     hipDeviceProp_t prop;
@@ -157,4 +250,4 @@ void DisplayTopology(bool outputToCsv)
            TransferBench::GetNumExecutorSubIndices({EXE_GPU_GFX, i}));
   }
 #endif
-}
+}