diff --git a/CHANGELOG.md b/CHANGELOG.md index 84847a37..3d86af2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,9 @@ Documentation for TransferBench is available at ## v1.67.00 ### Added -- Added NIC_TRAFFIC_CLASS to set the DSCP/traffic class byte in the RoCE GRH for QPs (RoCE only) -- Added NIC_SERVICE_LEVEL to set the IB service level (sl) for QPs (IB and RoCE) +- Added NIC_TRAFFIC_CLASS to set the DSCP/traffic class byte in the RoCE GRH for QPs (equivalent to NCCL_IB_TC) +- Added NIC_TRAFFIC_CLASS_FIFO to set a DSCP/traffic class to steer the control traffic into another priority queue (equivalent to NCCL_IB_FIFO_TC) +- Added NIC_SERVICE_LEVEL to set the IB service level (sl) for QPs (equivalent to NCCL_IB_SL) - Initial support for pod communication. Requires compatible hardware / ROCm version and subject to further testing - This potentially enables GFX/DMA executors to access SRC/DST memory locations on GPUs within the same pod - Pod membership requires amd-smi however can be skipped by setting TB_FORCE_SINGLE_POD=1 diff --git a/src/client/EnvVars.hpp b/src/client/EnvVars.hpp index 34940b69..4319e3f1 100644 --- a/src/client/EnvVars.hpp +++ b/src/client/EnvVars.hpp @@ -120,6 +120,7 @@ class EnvVars int nicChunkBytes; // Number of bytes to send per chunk for RDMA operations int nicCqPollBatch; // Number of CQ entries to poll per ibv_poll_cq call int nicRelaxedOrder; // Use relaxed ordering for RDMA + int nicFifoTrafficClass; // DSCP/traffic class byte for control (FIFO) QPs int nicServiceLevel; // IB service level (sl) for InfiniBand QPs int nicTrafficClass; // DSCP/traffic class byte for RoCE GRH int roceVersion; // RoCE version number @@ -186,11 +187,20 @@ class EnvVars ipAddressFamily = GetEnvVar("IP_ADDRESS_FAMILY" , 4); nicChunkBytes = GetEnvVar("NIC_CHUNK_BYTES" , 1073741824); nicCqPollBatch = GetEnvVar("NIC_CQ_POLL_BATCH" , 4); - nicRelaxedOrder = GetEnvVar("NIC_RELAX_ORDER" , 1); - nicServiceLevel = GetEnvVar("NIC_SERVICE_LEVEL" , 0); - nicTrafficClass = GetEnvVar("NIC_TRAFFIC_CLASS" , 0); + nicFifoTrafficClass = GetEnvVar("NIC_TRAFFIC_CLASS_FIFO", 0); + nicRelaxedOrder = GetEnvVar("NIC_RELAX_ORDER" , 1); + nicServiceLevel = GetEnvVar("NIC_SERVICE_LEVEL" , 0); + nicTrafficClass = GetEnvVar("NIC_TRAFFIC_CLASS" , 0); // Check that NIC service level and traffic class are in valid ranges + if (nicFifoTrafficClass < 0 || nicFifoTrafficClass > 255) { + printf("[ERROR] NIC_TRAFFIC_CLASS_FIFO must be in range 0..255 (got %d)\n", nicFifoTrafficClass); + exit(1); + } + if (nicFifoTrafficClass != 0 && numIterations <= 0) { + printf("[ERROR] NIC_TRAFFIC_CLASS_FIFO requires NUM_ITERATIONS > 0 (timed/infinite mode is not supported)\n"); + exit(1); + } if (nicServiceLevel < 0 || nicServiceLevel > 15) { printf("[ERROR] NIC_SERVICE_LEVEL must be in range 0..15 (got %d)\n", nicServiceLevel); exit(1); @@ -379,6 +389,7 @@ class EnvVars #if NIC_EXEC_ENABLED printf(" NIC_CHUNK_BYTES - Number of bytes to send at a time using NIC (default = 1GB)\n"); printf(" NIC_CQ_POLL_BATCH - Number of CQ entries to poll per ibv_poll_cq call (default = 4)\n"); + printf(" NIC_TRAFFIC_CLASS_FIFO - DSCP/traffic class byte for control (FIFO) QP GRH (default=0)\n"); printf(" NIC_RELAX_ORDER - Set to non-zero to use relaxed ordering\n"); printf(" NIC_SERVICE_LEVEL - IB service level (sl) for InfiniBand QPs (default=0)\n"); printf(" NIC_TRAFFIC_CLASS - DSCP/traffic class byte for RoCE GRH (default=0)\n"); @@ -520,6 +531,8 @@ class EnvVars "Polling %d CQ entries per ibv_poll_cq call", nicCqPollBatch); Print("NIC_RELAX_ORDER", nicRelaxedOrder, "Using %s ordering for NIC RDMA", nicRelaxedOrder ? "relaxed" : "strict"); + Print("NIC_TRAFFIC_CLASS_FIFO", nicFifoTrafficClass, + "RoCE FIFO/ctrl traffic class (DSCP) set to %d", nicFifoTrafficClass); Print("NIC_SERVICE_LEVEL", nicServiceLevel, "IB service level (sl) set to %d", nicServiceLevel); Print("NIC_TRAFFIC_CLASS", nicTrafficClass, @@ -744,6 +757,7 @@ class EnvVars cfg.nic.ibGidIndex = ibGidIndex; cfg.nic.ibPort = ibPort; cfg.nic.ipAddressFamily = ipAddressFamily; + cfg.nic.fifoTrafficClass = nicFifoTrafficClass; cfg.nic.useRelaxedOrder = nicRelaxedOrder; cfg.nic.serviceLevel = nicServiceLevel; cfg.nic.trafficClass = nicTrafficClass; diff --git a/src/header/TransferBench.hpp b/src/header/TransferBench.hpp index d94eb725..3cb5e92c 100644 --- a/src/header/TransferBench.hpp +++ b/src/header/TransferBench.hpp @@ -270,6 +270,7 @@ namespace TransferBench int queueSize = 100; ///< Completion queue size int roceVersion = 2; ///< RoCE version (used for auto GID detection) int useRelaxedOrder = 1; ///< Use relaxed ordering + uint8_t fifoTrafficClass = 0; ///< DSCP/traffic class byte for control (FIFO) QPs uint8_t serviceLevel = 0; ///< IB service level (sl) for InfiniBand QPs uint8_t trafficClass = 0; ///< DSCP/traffic class byte for RoCE GRH int useNuma = 0; ///< Switch to closest numa thread for execution @@ -2001,9 +2002,10 @@ namespace { if (nic.maxRecvWorkReq != cfg.nic.maxRecvWorkReq) ADD_ERROR("cfg.nic.maxRecvWorkReq"); if (nic.maxSendWorkReq != cfg.nic.maxSendWorkReq) ADD_ERROR("cfg.nic.maxSendWorkReq"); // nic.queueSize is permitted to be different across ranks - if (nic.roceVersion != cfg.nic.roceVersion) ADD_ERROR("cfg.nic.roceVersion"); - if (nic.serviceLevel != cfg.nic.serviceLevel) ADD_ERROR("cfg.nic.serviceLevel"); - if (nic.trafficClass != cfg.nic.trafficClass) ADD_ERROR("cfg.nic.trafficClass"); + if (nic.roceVersion != cfg.nic.roceVersion) ADD_ERROR("cfg.nic.roceVersion"); + if (nic.fifoTrafficClass != cfg.nic.fifoTrafficClass) ADD_ERROR("cfg.nic.fifoTrafficClass"); + if (nic.serviceLevel != cfg.nic.serviceLevel) ADD_ERROR("cfg.nic.serviceLevel"); + if (nic.trafficClass != cfg.nic.trafficClass) ADD_ERROR("cfg.nic.trafficClass"); if (nic.useRelaxedOrder != cfg.nic.useRelaxedOrder) ADD_ERROR("cfg.nic.useRelaxedOrder"); if (nic.useNuma != cfg.nic.useNuma) ADD_ERROR("cfg.nic.useNuma"); } @@ -2749,6 +2751,11 @@ namespace { ibv_gid dstGid; ///< GID handle for DST NIC vector srcQueuePairs; ///< Queue pairs for SRC NIC vector dstQueuePairs; ///< Queue pairs for DST NIC + ibv_cq* srcCtrlCompQueue; ///< Completion queue for SRC ctrl QPs (FIFO TC) + ibv_cq* dstCtrlCompQueue; ///< Completion queue for DST ctrl QPs (FIFO TC) + vector srcCtrlQueuePairs; ///< Control QPs on SRC NIC (FIFO TC) + vector dstCtrlQueuePairs; ///< Control QPs on DST NIC (FIFO TC) + ibv_send_wr ctrlSendWr; ///< Send WR for ctrl signal (zero-byte inline, reused per iteration) ibv_mr* srcMemRegion; ///< Memory region for SRC ibv_mr* dstMemRegion; ///< Memory region for DST int srcDmabufFd; ///< DMA-BUF file descriptor for SRC (if using dmabuf) @@ -3230,17 +3237,18 @@ static bool IsConfiguredGid(union ibv_gid const& gid) static ErrResult CreateQueuePair(ConfigOptions const& cfg, struct ibv_pd* pd, struct ibv_cq* cq, - struct ibv_qp*& qp) + struct ibv_qp*& qp, + int maxRecvWr = -1) { // Set queue pair attributes struct ibv_qp_init_attr attr = {}; - attr.qp_type = IBV_QPT_RC; // Set type to reliable connection - attr.send_cq = cq; // Send completion queue - attr.recv_cq = cq; // Recv completion queue - attr.cap.max_send_wr = cfg.nic.maxSendWorkReq; // Max send work requests - attr.cap.max_recv_wr = cfg.nic.maxRecvWorkReq; // Max recv work requests - attr.cap.max_send_sge = 1; // Max send scatter-gather entries - attr.cap.max_recv_sge = 1; // Max recv scatter-gather entries + attr.qp_type = IBV_QPT_RC; // Set type to reliable connection + attr.send_cq = cq; // Send completion queue + attr.recv_cq = cq; // Recv completion queue + attr.cap.max_send_wr = cfg.nic.maxSendWorkReq; // Max send work requests + attr.cap.max_recv_wr = (maxRecvWr >= 0) ? maxRecvWr : cfg.nic.maxRecvWorkReq; // Max recv work requests + attr.cap.max_send_sge = 1; // Max send scatter-gather entries + attr.cap.max_recv_sge = 1; // Max recv scatter-gather entries qp = ibv_create_qp(pd, &attr); if (qp == NULL) @@ -3554,6 +3562,37 @@ static bool IsConfiguredGid(union ibv_gid const& gid) rss.sendWorkRequests.resize(rss.qpCount); } + // Create control (FIFO) QPs when fifoTrafficClass is non-zero and this is an RDMA_WRITE + // transfer (srcIsExeNic == true). RDMA_READ transfers never use ctrl QPs so skip them. + // Compute ctrlNumPrepost in 64-bit to avoid signed overflow for large iteration counts, + // then clamp to INT_MAX before casting (NIC_TRAFFIC_CLASS_FIFO already rejects numIterations<=0). + int64_t ctrlNumPrepost64 = (int64_t)(cfg.general.numWarmups + cfg.general.numIterations) * + std::max(1, cfg.general.numSubIterations); + int ctrlNumPrepost = (int)std::min(ctrlNumPrepost64, (int64_t)INT_MAX); + if (cfg.nic.fifoTrafficClass != 0 && rss.srcIsExeNic) { + if (GetRank() == srcMemRank) { + IBV_PTR_CALL(rss.srcCtrlCompQueue, ibv_create_cq, + rss.srcContext, rss.qpCount, NULL, NULL, 0); + rss.srcCtrlQueuePairs.resize(rss.qpCount); + for (int i = 0; i < rss.qpCount; i++) { + // SRC ctrl QP only posts sends, so use default max_recv_wr + ERR_CHECK(CreateQueuePair(cfg, rss.srcProtect, rss.srcCtrlCompQueue, rss.srcCtrlQueuePairs[i])); + ERR_CHECK(InitQueuePair(rss.srcCtrlQueuePairs[i], port, rdmaAccessFlags)); + } + } + if (GetRank() == dstMemRank) { + IBV_PTR_CALL(rss.dstCtrlCompQueue, ibv_create_cq, + rss.dstContext, ctrlNumPrepost * rss.qpCount, NULL, NULL, 0); + rss.dstCtrlQueuePairs.resize(rss.qpCount); + for (int i = 0; i < rss.qpCount; i++) { + // DST ctrl QP pre-posts ctrlNumPrepost recv WRs; ensure max_recv_wr is large enough + ERR_CHECK(CreateQueuePair(cfg, rss.dstProtect, rss.dstCtrlCompQueue, rss.dstCtrlQueuePairs[i], + ctrlNumPrepost)); + ERR_CHECK(InitQueuePair(rss.dstCtrlQueuePairs[i], port, rdmaAccessFlags)); + } + } + } + // Broadcast SRC/DST port link_layer so that all ranks know it so that they can be compared System::Get().Broadcast(srcMemRank, sizeof(rss.srcPortAttr.link_layer), &rss.srcPortAttr.link_layer); System::Get().Broadcast(dstMemRank, sizeof(rss.dstPortAttr.link_layer), &rss.dstPortAttr.link_layer); @@ -3562,6 +3601,11 @@ static bool IsConfiguredGid(union ibv_gid const& gid) rss.srcNicIndex, srcMemRank, rss.dstNicIndex, dstMemRank, rss.srcPortAttr.link_layer, rss.dstPortAttr.link_layer}; } + // Shared result type for broadcasting QP transition success/failure across MPI ranks + struct QpTransitionResult { ErrType errType; bool rtrFailed; }; + static_assert(std::is_trivially_copyable::value, + "QpTransitionResult must be trivially copyable for MPI broadcast"); + ConnInfo dstConnInfo = {}; ConnInfo srcConnInfo = {}; for (int i = 0; i < rss.qpCount; i++) { @@ -3591,8 +3635,6 @@ static bool IsConfiguredGid(union ibv_gid const& gid) // Then move them to read-to-send (RTS) // Broadcast each rank's result so all ranks fail together rather than // hanging on the next iteration's Broadcast when qpCount > 1. - struct QpTransitionResult { ErrType errType; bool rtrFailed; }; - static_assert(std::is_trivially_copyable::value, "QpTransitionResult must be trivially copyable for MPI broadcast"); QpTransitionResult srcQpResult = {ERR_NONE, false}; if (GetRank() == srcMemRank) { ErrResult err = TransitionQpToRtr(rss.srcQueuePairs[i], dstConnInfo, port, srcIsRoCE, rss.srcPortAttr.active_mtu, cfg.nic.trafficClass, cfg.nic.serviceLevel); @@ -3675,6 +3717,86 @@ static bool IsConfiguredGid(union ibv_gid const& gid) } } } + + // Exchange ctrl QP numbers and connect them with fifoTrafficClass. + // Guard on srcIsExeNic to match the ctrl QP creation block above. + if (cfg.nic.fifoTrafficClass != 0 && rss.srcIsExeNic) { + ConnInfo srcCtrlInfo = {}, dstCtrlInfo = {}; + for (int i = 0; i < rss.qpCount; i++) { + if (GetRank() == srcMemRank) { + srcCtrlInfo.lid = rss.srcPortAttr.lid; + srcCtrlInfo.gid = rss.srcGid; + srcCtrlInfo.gidIdx = srcGidIndex; + srcCtrlInfo.qpn = rss.srcCtrlQueuePairs[i]->qp_num; + srcCtrlInfo.rkey = 0; + srcCtrlInfo.vaddr = 0; + } + System::Get().Broadcast(srcMemRank, sizeof(srcCtrlInfo), &srcCtrlInfo); + + if (GetRank() == dstMemRank) { + dstCtrlInfo.lid = rss.dstPortAttr.lid; + dstCtrlInfo.gid = rss.dstGid; + dstCtrlInfo.gidIdx = dstGidIndex; + dstCtrlInfo.qpn = rss.dstCtrlQueuePairs[i]->qp_num; + dstCtrlInfo.rkey = 0; + dstCtrlInfo.vaddr = 0; + } + System::Get().Broadcast(dstMemRank, sizeof(dstCtrlInfo), &dstCtrlInfo); + + QpTransitionResult srcCtrlResult = {ERR_NONE, false}; + if (GetRank() == srcMemRank) { + ErrResult err = TransitionQpToRtr(rss.srcCtrlQueuePairs[i], dstCtrlInfo, port, srcIsRoCE, + rss.srcPortAttr.active_mtu, cfg.nic.fifoTrafficClass, cfg.nic.serviceLevel); + srcCtrlResult.rtrFailed = (err.errType != ERR_NONE); + if (err.errType == ERR_NONE) err = TransitionQpToRts(rss.srcCtrlQueuePairs[i]); + srcCtrlResult.errType = err.errType; + } + System::Get().Broadcast(srcMemRank, sizeof(srcCtrlResult), &srcCtrlResult); + if (srcCtrlResult.errType != ERR_NONE) + return {ERR_FATAL, "SRC rank %d failed to transition ctrl QP %d to %s", + srcMemRank, i, srcCtrlResult.rtrFailed ? "RTR" : "RTS"}; + + QpTransitionResult dstCtrlResult = {ERR_NONE, false}; + if (GetRank() == dstMemRank) { + ErrResult err = TransitionQpToRtr(rss.dstCtrlQueuePairs[i], srcCtrlInfo, port, dstIsRoCE, + rss.dstPortAttr.active_mtu, cfg.nic.fifoTrafficClass, cfg.nic.serviceLevel); + dstCtrlResult.rtrFailed = (err.errType != ERR_NONE); + if (err.errType == ERR_NONE) err = TransitionQpToRts(rss.dstCtrlQueuePairs[i]); + dstCtrlResult.errType = err.errType; + } + System::Get().Broadcast(dstMemRank, sizeof(dstCtrlResult), &dstCtrlResult); + if (dstCtrlResult.errType != ERR_NONE) + return {ERR_FATAL, "DST rank %d failed to transition ctrl QP %d to %s", + dstMemRank, i, dstCtrlResult.rtrFailed ? "RTR" : "RTS"}; + } + + // Pre-build reusable ctrl send WR (zero-byte inline IBV_WR_SEND, posted once per iteration) + rss.ctrlSendWr = {}; + rss.ctrlSendWr.sg_list = nullptr; + rss.ctrlSendWr.num_sge = 0; + rss.ctrlSendWr.opcode = IBV_WR_SEND; + rss.ctrlSendWr.send_flags = IBV_SEND_SIGNALED | IBV_SEND_INLINE; + + // DST rank pre-posts recv WRs for all expected iterations so no per-iteration + // coordination (barrier) is needed between executor and non-executor ranks. + // ctrlNumPrepost was computed above and matches the max_recv_wr used when creating ctrl QPs. + if (GetRank() == dstMemRank) { + ibv_recv_wr ctrlRecvWr = {}; + ctrlRecvWr.sg_list = nullptr; + ctrlRecvWr.num_sge = 0; + ibv_recv_wr* badRecvWr; + for (int i = 0; i < rss.qpCount; i++) { + for (int n = 0; n < ctrlNumPrepost; n++) { + ibv_recv_wr wr = ctrlRecvWr; + int err = ibv_post_recv(rss.dstCtrlQueuePairs[i], &wr, &badRecvWr); + if (err) + return {ERR_FATAL, "Transfer %d: ibv_post_recv pre-post on ctrl QP %d failed (%s)", + rss.transferIdx, i, strerror(err)}; + } + } + } + } + return ERR_NONE; } @@ -3711,6 +3833,18 @@ static bool IsConfiguredGid(union ibv_gid const& gid) rss.dstQueuePairs.clear(); } + // Destroy ctrl queue pairs and completion queues (only exist when fifoTrafficClass != 0) + if (isSrcRank && !rss.srcCtrlQueuePairs.empty()) { + for (auto qp : rss.srcCtrlQueuePairs) IBV_CALL(ibv_destroy_qp, qp); + rss.srcCtrlQueuePairs.clear(); + IBV_CALL(ibv_destroy_cq, rss.srcCtrlCompQueue); + } + if (isDstRank && !rss.dstCtrlQueuePairs.empty()) { + for (auto qp : rss.dstCtrlQueuePairs) IBV_CALL(ibv_destroy_qp, qp); + rss.dstCtrlQueuePairs.clear(); + IBV_CALL(ibv_destroy_cq, rss.dstCtrlCompQueue); + } + // Destroy completion queues if (isSrcRank) IBV_CALL(ibv_destroy_cq, rss.srcCompQueue); if (isDstRank) IBV_CALL(ibv_destroy_cq, rss.dstCompQueue); @@ -4653,7 +4787,30 @@ static bool IsConfiguredGid(union ibv_gid const& gid) int const exeIndex, TransferResources& rss) { - // Loop over each of the queue pairs and post work request + // Ctrl path: executor (src) fires one zero-byte inline IBV_WR_SEND per ctrl QP. + // Recv WRs are pre-posted on the dst side during setup, so no barrier is needed. + if (cfg.nic.fifoTrafficClass != 0 && rss.srcIsExeNic) { + ibv_send_wr* badSendWr; + for (int qpIdx = 0; qpIdx < rss.qpCount; qpIdx++) { + rss.ctrlSendWr.wr_id = qpIdx; + int err = ibv_post_send(rss.srcCtrlQueuePairs[qpIdx], &rss.ctrlSendWr, &badSendWr); + if (err) + return {ERR_FATAL, "Transfer %d: ibv_post_send on ctrl QP %d failed (%s)", + rss.transferIdx, qpIdx, strerror(err)}; + } + for (int qpIdx = 0; qpIdx < rss.qpCount; qpIdx++) { + ibv_wc wc; + int nc; + while ((nc = ibv_poll_cq(rss.srcCtrlCompQueue, 1, &wc)) == 0) {} + if (nc < 0) + return {ERR_FATAL, "Transfer %d: ctrl CQ poll error", rss.transferIdx}; + if (wc.status != IBV_WC_SUCCESS) + return {ERR_FATAL, "Transfer %d: ctrl send CQ error on QP %llu [status %d]", + rss.transferIdx, wc.wr_id, wc.status}; + } + } + + // Data path — unchanged: post all RDMA send WRs (stamped with trafficClass) ibv_send_wr* badWorkReq; for (int qpIndex = 0; qpIndex < rss.qpCount; qpIndex++) { size_t numChunks = rss.sendWorkRequests[qpIndex].size();