From 91b832b0c4f73b11aeffff4af28b23a73c043c03 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 31 Aug 2026 09:21:23 +0000 Subject: [PATCH 1/2] feat(sdma): key anvil queues on KFD node ids instead of HIP ordinals Wire intra-node SDMA queues by the host-global KFD topology node id (== HSA_AGENT_INFO_NODE) rather than a HIP device ordinal. HIP ordinals are a per-process slice of HIP_VISIBLE_DEVICES, so the old within-node derivations (Context::SameHostPeersBefore, cco's dstDeviceId=lsa, and the symmetric-memory SDMA-peer counter) plus the BDF band-aid only agreed in the full-8-GPU-visibility case and broke under sliced visibility. - Context now resolves each rank's local GPU KFD node id HSA-free from the bound device BDF via sysfs and allgathers it in CollectHostNames; adds KfdNodeId()/LocalKfdNode() accessors. - anvil re-keys sdma_channels_ on (srcNode,dstNode); connect/getSdmaQueue and the engine-mask helpers take node ids; adds agentForNode() to pick the local queue's HSA agent and nodeForHipDevice() for single-process callers. - EnsureSdmaTransport uses node ids and drops hipDeviceEnablePeerAccess: the SDMA copy engine targets already-mapped fabric/IPC VAs, so HIP peer access is unnecessary and was the only hard HIP-visibility dependency. - cco_init and symmetric_memory pass node ids to getSdmaQueue; kernel-facing handle arrays stay indexed by logical rank (lsaRank / global pe). - examples/sdma convert HIP ordinals via nodeForHipDevice(). Cannot be built/tested in this environment (no ROCm/HIP); WIP for hardware validation, including a sliced HIP_VISIBLE_DEVICES run and a full-visibility regression. Co-authored-by: Pavel Emeliyanenko test(cco/sdma): add torch- and jax-style SDMA all-gather tests Two multi-rank CCO SDMA all-gather tests exercising the KFD-node-id queue keying across both launch topologies: - test_sdma_allgather_torch: full HIP visibility, one distinct GPU per rank (hipSetDevice(rank % numDevices)) -- HIP ordinals match physical GPUs. - test_sdma_allgather_jax: each rank slices HIP_VISIBLE_DEVICES= before its first HIP call and binds device 0, so HIP ordinal 0 maps to a different physical GPU per process (ROCR_VISIBLE_DEVICES left unset so HSA still sees all GPUs). This is the sliced-visibility case that HIP-ordinal keying could not handle and that the node-id keying fixes. Shared kernel body + comm setup/verify live in sdma_allgather_common.hpp; each test keeps its own __global__ wrapper so the CMake auto-discovery flags it HIP. Auto-registered as cco_sdma_allgather_torch / cco_sdma_allgather_jax under BUILD_CCO_SDMA; both SKIP when MORI_ENABLE_SDMA is unset. Not built here (no ROCm/HIP); WIP for hardware validation. Co-authored-by: Pavel Emeliyanenko updated anvil refactoring and added tests changed the tests fixed clang some fixes clang fix fix clang refactor(context): dedup rankInNode; derive LocalRankInNode from peerInfos The Context::rankInNode member was a redundant recomputation of peerInfos[LocalRank()].rankInNode (the -1 init just cancelled the self-count in the loop). Drop the member and the counting loop: LocalRankInNode() now returns peerInfos[LocalRank()].rankInNode directly (mirroring LocalKfdNode()), and InitializeTopologyAndTransports takes a local from the same source for NIC selection. No behavior change to rail-only routing or NIC matching. Co-authored-by: Pavel Emeliyanenko refactoring continued restored legacy tests that use hipDevID update fixing precommit updated clang refactor Breaking the loop on the failure Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> interface update --- examples/sdma/sdma_bw.cpp | 18 +- examples/sdma/sdma_bw_allgather.cpp | 16 +- examples/sdma/sdma_latency.cpp | 9 +- examples/sdma/sdma_rate.cpp | 10 +- include/mori/application/context/context.hpp | 15 +- .../mori/application/transport/sdma/anvil.hpp | 54 +- src/application/context/context.cpp | 113 ++--- src/application/memory/symmetric_memory.cpp | 30 +- src/application/transport/sdma/anvil.cpp | 211 ++++---- src/cco/cco_init.cpp | 12 +- tests/cpp/cco/test_sdma_hip_dev_assign.cpp | 473 ++++++++++++++++++ tools/run_cco_tests.sh | 4 + 12 files changed, 728 insertions(+), 237 deletions(-) create mode 100644 tests/cpp/cco/test_sdma_hip_dev_assign.cpp diff --git a/examples/sdma/sdma_bw.cpp b/examples/sdma/sdma_bw.cpp index 3a863f944..53f49564c 100644 --- a/examples/sdma/sdma_bw.cpp +++ b/examples/sdma/sdma_bw.cpp @@ -176,9 +176,20 @@ void runExperiment(int srcDeviceId, const ExperimentParams& params) { size_t totalNumQueues = params.numOfQueues * params.numDestinations; + // Resolve KFD node ids once: anvil queues are keyed on node ids, not HIP + // device ordinals, so both connect() and getSdmaQueue() below must use + // the SAME converted id (calling them separately with raw device ids would + // silently miss the queue lookup whenever a device ordinal != its node id). + int srcNode = anvil::anvil.nodeForHipDevice(srcDeviceId); + std::vector dstNodes; + dstNodes.reserve(dstDeviceIds.size()); for (auto& dstDeviceId : dstDeviceIds) { + dstNodes.push_back(anvil::anvil.nodeForHipDevice(dstDeviceId)); + } + + for (auto& dstNode : dstNodes) { // Better performance if allocating all 8 queues - anvil::anvil.connect(srcDeviceId, dstDeviceId, 8); // params.numOfQueues); + anvil::anvil.connect(srcNode, dstNode, 8); // params.numOfQueues); } anvil::SdmaQueueDeviceHandle** deviceHandles_d = nullptr; @@ -186,10 +197,9 @@ void runExperiment(int srcDeviceId, const ExperimentParams& params) { hipMalloc(&deviceHandles_d, totalNumQueues * sizeof(anvil::SdmaQueueDeviceHandle*))); size_t queueIdx = 0; - for (auto& dstDeviceId : dstDeviceIds) { + for (auto& dstNode : dstNodes) { for (size_t q = 0; q < params.numOfQueues; q++) { - deviceHandles_d[queueIdx] = - anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId, q)->deviceHandle(); + deviceHandles_d[queueIdx] = anvil::anvil.getSdmaQueue(srcNode, dstNode, q)->deviceHandle(); queueIdx++; } } diff --git a/examples/sdma/sdma_bw_allgather.cpp b/examples/sdma/sdma_bw_allgather.cpp index fa6a7fafb..203e141f4 100644 --- a/examples/sdma/sdma_bw_allgather.cpp +++ b/examples/sdma/sdma_bw_allgather.cpp @@ -411,8 +411,17 @@ void runExperimentMPI(int srcDeviceId, int mpiRank, int mpiSize, const Experimen // ============================= std::cout << "Process " << mpiRank << ": Setting up Anvil SDMA queues..." << std::endl; + // Resolve KFD node ids once: anvil queues are keyed on node ids, not HIP + // device ordinals, so both connect() and getSdmaQueue() below must use + // the SAME converted id. Since this example keeps HIP_VISIBLE_DEVICES + // unsliced/identical across all MPI ranks (each rank just binds device == + // its own rank), resolving a peer's node id from THIS process's own device + // enumeration is valid -- no cross-process node-id exchange needed here. + int srcNode = anvil::anvil.nodeForHipDevice(srcDeviceId); + // Establish connection for each destination GPU for (int dstDeviceId : dstDeviceIds) { + int dstNode = anvil::anvil.nodeForHipDevice(dstDeviceId); try { // Ensure correct device context err = hipSetDevice(srcDeviceId); @@ -433,7 +442,7 @@ void runExperimentMPI(int srcDeviceId, int mpiRank, int mpiSize, const Experimen std::cout << "Process " << mpiRank << ": Connecting GPU" << srcDeviceId << " -> GPU" << dstDeviceId << " ..." << std::endl; - anvil::anvil.connect(srcDeviceId, dstDeviceId, params.numOfQueues); + anvil::anvil.connect(srcNode, dstNode, params.numOfQueues); std::cout << "Process " << mpiRank << ": GPU" << srcDeviceId << " -> GPU" << dstDeviceId << " Anvil connection successful" << std::endl; @@ -444,7 +453,7 @@ void runExperimentMPI(int srcDeviceId, int mpiRank, int mpiSize, const Experimen // Try fallback to using 1 queue std::cerr << "Process " << mpiRank << ": Trying to reconnect with 1 queue..." << std::endl; try { - anvil::anvil.connect(srcDeviceId, dstDeviceId, 1); + anvil::anvil.connect(srcNode, dstNode, 1); std::cout << "Process " << mpiRank << ": GPU" << srcDeviceId << " -> GPU" << dstDeviceId << " Fallback connection successful" << std::endl; } catch (const std::exception& e2) { @@ -479,9 +488,10 @@ void runExperimentMPI(int srcDeviceId, int mpiRank, int mpiSize, const Experimen for (size_t dstIdx = 0; dstIdx < dstDeviceIds.size(); dstIdx++) { int dstDeviceId = dstDeviceIds[dstIdx]; + int dstNode = anvil::anvil.nodeForHipDevice(dstDeviceId); for (size_t q = 0; q < params.numOfQueues; q++) { try { - auto queue = anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId, q); + auto queue = anvil::anvil.getSdmaQueue(srcNode, dstNode, q); if (queue && queue->deviceHandle()) { host_device_handles[queueIdx] = queue->deviceHandle(); std::cout << "Process " << mpiRank << ": Destination GPU" << dstDeviceId << " queue[" << q diff --git a/examples/sdma/sdma_latency.cpp b/examples/sdma/sdma_latency.cpp index 12158bc51..d664e02eb 100644 --- a/examples/sdma/sdma_latency.cpp +++ b/examples/sdma/sdma_latency.cpp @@ -148,11 +148,16 @@ void runExperiment(int srcDeviceId, int dstDeviceId, const ExperimentParams& par // 3. Queue Setup // ====================== - anvil::anvil.connect(srcDeviceId, dstDeviceId); + // Resolve KFD node ids once: anvil queues are keyed on node ids, not HIP + // device ordinals, so both connect() and getSdmaQueue() below must use + // the SAME converted id. + int srcNode = anvil::anvil.nodeForHipDevice(srcDeviceId); + int dstNode = anvil::anvil.nodeForHipDevice(dstDeviceId); + anvil::anvil.connect(srcNode, dstNode); anvil::SdmaQueueDeviceHandle* deviceHandle_d = nullptr; - deviceHandle_d = anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId)->deviceHandle(); + deviceHandle_d = anvil::anvil.getSdmaQueue(srcNode, dstNode)->deviceHandle(); // if (params.verbose) // { diff --git a/examples/sdma/sdma_rate.cpp b/examples/sdma/sdma_rate.cpp index 741a89d80..2bc0d4347 100644 --- a/examples/sdma/sdma_rate.cpp +++ b/examples/sdma/sdma_rate.cpp @@ -137,7 +137,12 @@ void runExperiment(int srcDeviceId, int dstDeviceId, const ExperimentParams& par // ====================== size_t totalNumQueues = params.numOfQueues; - anvil::anvil.connect(srcDeviceId, dstDeviceId, params.numOfQueues); + // Resolve KFD node ids once: anvil queues are keyed on node ids, not HIP + // device ordinals, so both connect() and getSdmaQueue() below must use + // the SAME converted id. + int srcNode = anvil::anvil.nodeForHipDevice(srcDeviceId); + int dstNode = anvil::anvil.nodeForHipDevice(dstDeviceId); + anvil::anvil.connect(srcNode, dstNode, params.numOfQueues); anvil::SdmaQueueDeviceHandle** deviceHandles_d = nullptr; CHECK_HIP_ERROR( @@ -145,8 +150,7 @@ void runExperiment(int srcDeviceId, int dstDeviceId, const ExperimentParams& par size_t queueIdx = 0; for (size_t q = 0; q < params.numOfQueues; q++) { - deviceHandles_d[queueIdx] = - anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId, q)->deviceHandle(); + deviceHandles_d[queueIdx] = anvil::anvil.getSdmaQueue(srcNode, dstNode, q)->deviceHandle(); // if (params.verbose) // { diff --git a/include/mori/application/context/context.hpp b/include/mori/application/context/context.hpp index ac63034ca..f67c1d859 100644 --- a/include/mori/application/context/context.hpp +++ b/include/mori/application/context/context.hpp @@ -78,7 +78,6 @@ class Context { int LocalRank() const { return bootNet.GetLocalRank(); } int WorldSize() const { return bootNet.GetWorldSize(); } - int LocalRankInNode() const { return rankInNode; } const std::string& HostName() const { return myHostname; } // Single-value transport selection driven by Context's default policy @@ -95,6 +94,15 @@ class Context { const PeerCapabilities& GetPeerCapabilities(int destRank) const { return peerCaps[destRank]; } const std::vector& GetAllPeerCapabilities() const { return peerCaps; } + // KFD topology node id (== HSA_AGENT_INFO_NODE) of a peer's GPU. This is a + // process-independent, host-global identity: it does not depend on + // HIP_VISIBLE_DEVICES, so it is the correct key for wiring SDMA queues to a + // peer even when the peer GPU is not in this process's HIP device list. + // -1 if the peer's node id could not be resolved. Populated in + // CollectHostNames() via an allgather. + int KfdNodeId(int destRank) const { return peerInfos[destRank].kfdNodeId; } + int LocalKfdNode() const { return peerInfos[LocalRank()].kfdNodeId; } + RdmaContext* GetRdmaContext() const { return rdmaContext.get(); } RdmaDeviceContext* GetRdmaDeviceContext() const { return rdmaDeviceContext.get(); } const std::vector>& GetAllRdmaDeviceContexts() const { @@ -184,16 +192,13 @@ class Context { bool sameHost{false}; bool sameProcess{false}; int rankInNode{-1}; + int kfdNodeId{-1}; // KFD topology node id of this rank's GPU (host-global) }; bool RailOnlyEligible() const; private: - // Number of same-host peers with rank < `rank` (a peer's within-node device id). - int SameHostPeersBefore(int rank) const; - BootstrapNetwork& bootNet; - int rankInNode{-1}; int numQpPerPe{4}; bool sdmaEnabled{false}; bool p2pDisabled{false}; diff --git a/include/mori/application/transport/sdma/anvil.hpp b/include/mori/application/transport/sdma/anvil.hpp index 3e8c713a1..bc95a1ca2 100644 --- a/include/mori/application/transport/sdma/anvil.hpp +++ b/include/mori/application/transport/sdma/anvil.hpp @@ -47,13 +47,12 @@ namespace anvil { class SdmaQueue { public: - SdmaQueue(int localDeviceId, int remoteDeviceId, hsa_agent_t& localAgent, uint32_t engineId); + SdmaQueue(uint32_t localNodeId, uint32_t engineId); ~SdmaQueue(); SdmaQueueDeviceHandle* deviceHandle() const; private: - int remoteDeviceId_; // TODO unused uint64_t* cachedWptr_; uint64_t* committedWptr_; void* queueBuffer_; @@ -76,8 +75,25 @@ class AnvilLib { public: void init(); - bool connect(int srcDeviceId, int dstDeviceId, int numChannels = 1); - SdmaQueue* getSdmaQueue(int srcDeviceId, int dstDeviceId, int channelIdx = 0); + // srcNode/dstNode are KFD topology node ids (== HSA_AGENT_INFO_NODE), a + // host-global GPU identity that does NOT depend on HIP_VISIBLE_DEVICES. This + // is the correct key for a peer even when the peer GPU is not in this + // process's HIP device list. Channels for a pair are shared process-wide. + bool connect(int srcNode, int dstNode, int numChannels = 1); + + // Get the SDMA queue for a given src/dst node pair and channel index. + SdmaQueue* getSdmaQueue(int srcNode, int dstNode, int channelIdx = 0); + + // Map a HIP device ordinal to its KFD node id. For single-process callers + // (e.g. the examples) that only have HIP device ids; multi-process collectives + // should exchange node ids out-of-band instead (see Context::KfdNodeId). + static uint32_t nodeForHipDevice(int hipDev); + + // Resolve the KFD topology node id of the given HIP device WITHOUT initializing + // HSA. The KFD node id (the directory index under /sys/class/kfd/kfd/topology/nodes) + // is identical to what HSA_AGENT_INFO_NODE / hsaKmtGetNodeProperties report, + // and is a host-global identity independent of HIP_VISIBLE_DEVICES. + static int kfdNodeIdForHipDevice(int hipDev); private: /* @@ -101,25 +117,23 @@ class AnvilLib { {5, 3, 2, 4, 6, 1, 0, 7}, {3, 6, 4, 2, 1, 5, 7, 0}}}; - int getOamId(int deviceId); + // xGMI physical (OAM) id for a KFD node, read from the GPU's PCI sysfs. + int getOamId(int node); - int getSdmaEngineId(int srcDeviceId, int dstDeviceId); - - // KFD topology node id for a HIP device id. - uint32_t getNodeId(int deviceId); + int getSdmaEngineId(int srcNode, int dstNode); // Bitmask of SDMA engine ids KFD recommends for the src->dst xGMI link to // reach maximum bandwidth (sysfs recommended_sdma_engine_id_mask). Returns 0 // if the link or property is unavailable, in which case callers fall back to // the static OAM map. - uint32_t getRecommendedEngineMask(int srcDeviceId, int dstDeviceId); + uint32_t getRecommendedEngineMask(int srcNode, int dstNode); // Bitmask of the general (CPU-link, non-xGMI) SDMA engines. Zero if the node // reports no CPU link. Used to spread loopback channels on gfx1250. - uint32_t getHostLinkEngineMask(int srcDeviceId); + uint32_t getHostLinkEngineMask(int srcNode); // True for gfx1250 (gfx12.5), the only arch that spreads loopback channels. - bool isGfx1250(int deviceId); + bool isGfx1250(int node); struct PairHash { std::size_t operator()(const std::pair& p) const { @@ -144,22 +158,6 @@ inline void checkHipError(hipError_t err, const char* msg, const char* file, int } #define CHECK_HIP_ERROR(cmd) anvil::checkHipError((cmd), #cmd, __FILE__, __LINE__) -// Allow access to peerDeviceId from deviceId -inline void EnablePeerAccess(int const deviceId, int const peerDeviceId) { - int canAccess; - CHECK_HIP_ERROR(hipDeviceCanAccessPeer(&canAccess, deviceId, peerDeviceId)); - if (!canAccess) { - std::cerr << "Unable to enable peer access from GPU devices " << deviceId << " to " - << peerDeviceId << "\n"; - } - - CHECK_HIP_ERROR(hipSetDevice(deviceId)); - hipError_t error = hipDeviceEnablePeerAccess(peerDeviceId, 0); - if (error != hipSuccess && error != hipErrorPeerAccessAlreadyEnabled) { - std::cerr << "Unable to enable peer to peer access from " << deviceId << " to " << peerDeviceId - << " (" << hipGetErrorString(error) << ")\n"; - } -} // Hardware cap on SDMA channels per GPU pair on CDNA (4 queues/engine × // 2 recommended engines). Requests above this are clamped, not failed. inline constexpr int kMaxSdmaChannelsPerPair = 8; diff --git a/src/application/context/context.cpp b/src/application/context/context.cpp index 5bd185a3d..dd2c823ff 100644 --- a/src/application/context/context.cpp +++ b/src/application/context/context.cpp @@ -29,7 +29,10 @@ #include #include +#include +#include #include +#include #include #include #include @@ -145,6 +148,8 @@ void Context::BuildInitialEndpoints() { Context::~Context() {} +namespace { + std::string GetLocalIP() { struct ifaddrs *ifaddr, *ifa; char host[NI_MAXHOST]; @@ -178,6 +183,8 @@ std::string GetLocalIP() { return localIP; } +} // namespace + bool Context::CanUseP2P(int destRank) const { if (destRank == LocalRank()) { return false; // Cannot use P2P with self @@ -197,51 +204,50 @@ void Context::CollectHostNames() { gethostname(hostname, HOST_NAME_MAX); myHostname = std::string(hostname); + int hipDev = 0; + HIP_RUNTIME_CHECK(hipGetDevice(&hipDev)); // Key co-location on node id, not hostname: identical hostnames would mark // cross-node ranks as co-located, over-counting rankInNode (trips assert below). std::string nodeId = ResolveNodeId(myHostname); + // Allgather a fixed-layout {pid, kfdNodeId, railFlag, nodeId} record. + struct Pack { + pid_t pid; + int32_t kfdNodeId; + bool railFlag; + char nodeId[256]; + } my = { + .pid = getpid(), + // Local GPU's KFD node id (host-global, HIP_VISIBLE_DEVICES-independent). Used + // as the stable key for wiring SDMA queues to same-host peers. + .kfdNodeId = anvil::anvil.kfdNodeIdForHipDevice(hipDev), + .railFlag = + env::IsEnvVarEnabled("MORI_ENABLE_RAIL_ONLY") || env::IsEnvVarEnabled("MORI_ENABLE_RAIL"), + }; + snprintf(my.nodeId, sizeof(my.nodeId), "%s", nodeId.c_str()); - // Allgather a fixed-layout {pid, railFlag, nodeId} record. - constexpr int kPidSize = sizeof(pid_t); - constexpr int kFlagSize = sizeof(uint8_t); - constexpr int kStrMax = 256; - constexpr int kRecordSize = kPidSize + kFlagSize + kStrMax; - - pid_t myPid = getpid(); - uint8_t myRailFlag = - env::IsEnvVarEnabled("MORI_ENABLE_RAIL_ONLY") || env::IsEnvVarEnabled("MORI_ENABLE_RAIL"); - char localBuffer[kRecordSize] = {}; - memcpy(localBuffer, &myPid, kPidSize); - memcpy(localBuffer + kPidSize, &myRailFlag, kFlagSize); - snprintf(localBuffer + kPidSize + kFlagSize, kStrMax, "%s", nodeId.c_str()); - - std::vector global(kRecordSize * WorldSize()); - bootNet.Allgather(localBuffer, global.data(), kRecordSize); + std::vector global(WorldSize()); + bootNet.Allgather(&my, global.data(), sizeof(Pack)); - std::string myNodeId(localBuffer + kPidSize + kFlagSize); peerInfos.resize(WorldSize()); std::map seenPerNode; bool anyRailMismatch = false; for (int i = 0; i < WorldSize(); i++) { - const char* rec = global.data() + i * kRecordSize; - pid_t peerPid; - memcpy(&peerPid, rec, kPidSize); - uint8_t peerRailFlag; - memcpy(&peerRailFlag, rec + kPidSize, kFlagSize); - std::string peerNodeId(rec + kPidSize + kFlagSize); - peerInfos[i].sameHost = (peerNodeId == myNodeId); - peerInfos[i].sameProcess = peerInfos[i].sameHost && (peerPid == myPid); + const Pack& other = global[i]; + std::string peerNodeId(other.nodeId); + peerInfos[i].sameHost = (peerNodeId == my.nodeId); + peerInfos[i].sameProcess = peerInfos[i].sameHost && (other.pid == my.pid); peerInfos[i].rankInNode = seenPerNode[peerNodeId]++; - if (peerRailFlag != myRailFlag) { + peerInfos[i].kfdNodeId = other.kfdNodeId; + if (other.railFlag != my.railFlag) { MORI_APP_ERROR("MORI_ENABLE_RAIL_ONLY mismatch: rank {} has {}={}, this rank ({}) has {}", i, - peerRailFlag ? "on" : "off", peerRailFlag, LocalRank(), - myRailFlag ? "on" : "off"); + other.railFlag ? "on" : "off", other.railFlag, LocalRank(), + my.railFlag ? "on" : "off"); anyRailMismatch = true; } if (LocalRank() == 0) { - MORI_APP_TRACE("rank {} nodeId={} pid={} rankInNode={} sameHost={} sameProcess={}", i, - peerNodeId, peerPid, peerInfos[i].rankInNode, peerInfos[i].sameHost, - peerInfos[i].sameProcess); + MORI_APP_TRACE("rank {} nodeId={} pid={} rankInNode={} kfdNode={} sameHost={} sameProcess={}", + i, peerNodeId, other.pid, peerInfos[i].rankInNode, other.kfdNodeId, + peerInfos[i].sameHost, peerInfos[i].sameProcess); } } if (anyRailMismatch) { @@ -278,18 +284,10 @@ bool Context::RailOnlyEligible() const { // / Context::IsP2PDisabled() instead of getenv anywhere outside the // constructor. -int Context::SameHostPeersBefore(int rank) const { - int n = 0; - for (int j = 0; j < rank; j++) - if (peerInfos[j].sameHost) n++; - return n; -} - void Context::InitializeTopologyAndTransports() { - // Find my rank in node - for (int i = 0; i <= LocalRank(); i++) { - if (peerInfos[i].sameHost) rankInNode++; - } + // Local rank within this host. Already derived in CollectHostNames from the + // exchanged nodeId ordering (seenPerNode), so reuse it instead of recounting. + int rankInNode = peerInfos[LocalRank()].rankInNode; assert(rankInNode < 8); // Init rdma context — proxy uses vendor-agnostic IBVerbs, IBGDA uses DirectVerbs @@ -321,8 +319,7 @@ void Context::InitializeTopologyAndTransports() { std::cout << "MORI Topology detection is disabled, use static matching" << std::endl; if (!activeDevicePortList.empty()) { devicePortId = (rankInNode % activeDevicePortList.size()); - device = activeDevicePortList[devicePortId].first; - portId = activeDevicePortList[devicePortId].second; + std::tie(device, portId) = activeDevicePortList[devicePortId]; rdmaDeviceContext.reset(device->CreateRdmaDeviceContext()); } } else { @@ -422,11 +419,6 @@ void Context::InitializeTopologyAndTransports() { savedEpConfig.onGpu = true; } } - - // rankInNode bookkeeping (used by NIC selection above and as - // LocalRankInNode() accessor). Kept here so capability discovery still - // computes it as a derived fact. - // (already computed at the top of this function) } /* ------------------------------------------------------------------------ */ @@ -482,15 +474,24 @@ void Context::EnsureSdmaTransport(int requestedChannels) { } MORI_APP_INFO("SDMA num channels per GPU pair: {}", sdmaNumChannels); - // Within-node HIP device id = count of same-host peers before the rank - // (not globalRank % 8, which faults under sliced HIP_VISIBLE_DEVICES). - int localDevId = SameHostPeersBefore(LocalRank()); + // Key SDMA queues on the KFD topology node id (host-global, exchanged in + // CollectHostNames), NOT a HIP device ordinal. This is correct even under + // sliced HIP_VISIBLE_DEVICES, where a peer GPU is not in this process's HIP + // device list. + int localNode = LocalKfdNode(); + if (localNode < 0) { + MORI_APP_ERROR("EnsureSdmaTransport: local KFD node id unresolved for rank {}", LocalRank()); + std::abort(); + } for (int i = 0; i < WorldSize(); i++) { if (!peerCaps[i].canSDMA) continue; - // Peer within-node device id: count of same-host peers before it. - int peerDevId = SameHostPeersBefore(i); - if (i != LocalRank()) anvil::EnablePeerAccess(localDevId, peerDevId); - anvil::anvil.connect(localDevId, peerDevId, sdmaNumChannels); + int peerNode = KfdNodeId(i); + if (peerNode < 0) { + MORI_APP_ERROR("EnsureSdmaTransport: peer {} KFD node id unresolved for rank {}", i, + LocalRank()); + std::abort(); + } + anvil::anvil.connect(localNode, peerNode, sdmaNumChannels); } sdmaChannels_ = sdmaNumChannels; sdmaSetupDone = true; diff --git a/src/application/memory/symmetric_memory.cpp b/src/application/memory/symmetric_memory.cpp index 9258cdeb2..00fba0d53 100644 --- a/src/application/memory/symmetric_memory.cpp +++ b/src/application/memory/symmetric_memory.cpp @@ -260,23 +260,18 @@ SymmMemObjPtr SymmMemManager::RegisterSymmMemObj(void* localPtr, size_t size, bo HIP_RUNTIME_CHECK(hipMemcpy(gpuMemObj->peerRkeys, cpuMemObj->peerRkeys, sizeof(uint32_t) * worldSize, hipMemcpyHostToDevice)); - // SDMA peers (same-host). Each peer needs its within-node HIP device id - // (0-based) for the anvil queue key, but the device-handle array is addressed + // SDMA peers (same-host). The anvil queue key is each GPU's host-global KFD + // node id (exchanged in Context), while the device-handle array is addressed // by GLOBAL pe in the kernels (deviceHandles_d + pe * numQueues). Keep the two - // separate: (pe % 8) is wrong for multi-node / sliced HIP_VISIBLE_DEVICES runs - // where global ranks 4..7 on node 1 map to local HIP devices 0..3. - std::vector> sdmaPeers; // (globalPe, withinNodeDevId) - { - int within = 0; - for (int i = 0; i < worldSize; i++) { - if (context.GetTransportType(i) != TransportType::SDMA) continue; - sdmaPeers.emplace_back(i, within++); - } + // separate: KFD node ids stay correct under multi-node / sliced + // HIP_VISIBLE_DEVICES runs, where HIP ordinals diverge from topology. + std::vector sdmaPeers; // global pe + for (int i = 0; i < worldSize; i++) { + if (context.GetTransportType(i) != TransportType::SDMA) continue; + sdmaPeers.push_back(i); } if (!sdmaPeers.empty()) { - int srcDeviceId = 0; // within-node id of self - for (int j = 0; j < rank; j++) - if (context.GetTransportType(j) == TransportType::SDMA) srcDeviceId++; + int srcNode = context.LocalKfdNode(); // KFD node id of self int numOfQueuesPerDevice = gpuMemObj->sdmaNumQueue; // all sdma queues are inited // Allocate based on worldSize because indexing uses pe * numQ where pe ranges // 0..worldSize-1. Using sdmaPeers.size causes buffer overflow. @@ -288,11 +283,10 @@ SymmMemObjPtr SymmMemManager::RegisterSymmMemObj(void* localPtr, size_t size, bo hipMemset(gpuMemObj->deviceHandles_d, 0, numDevices * numOfQueuesPerDevice * sizeof(anvil::SdmaQueueDeviceHandle*))); - for (auto& peer : sdmaPeers) { - int dstPe = peer.first; // global pe -> array index (kernel-facing) - int dstDeviceId = peer.second; // within-node id -> anvil queue key + for (int dstPe : sdmaPeers) { + int dstNode = context.KfdNodeId(dstPe); // KFD node id -> anvil queue key for (size_t q = 0; q < numOfQueuesPerDevice; q++) { - auto* anvilHandle = anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId, q)->deviceHandle(); + auto* anvilHandle = anvil::anvil.getSdmaQueue(srcNode, dstNode, q)->deviceHandle(); HIP_RUNTIME_CHECK(hipMemcpy(&gpuMemObj->deviceHandles_d[dstPe * numOfQueuesPerDevice + q], &anvilHandle, sizeof(anvilHandle), hipMemcpyHostToDevice)); } diff --git a/src/application/transport/sdma/anvil.cpp b/src/application/transport/sdma/anvil.cpp index 5a35fe5e4..57b743564 100644 --- a/src/application/transport/sdma/anvil.cpp +++ b/src/application/transport/sdma/anvil.cpp @@ -37,18 +37,19 @@ #include #include #include + +#include "mori/utils/mori_log.hpp" namespace anvil { -auto checkHsaError = [](hsa_status_t s, const char* msg, const char* file, int line) { - if (s != HSA_STATUS_SUCCESS) { - const char* hsa_err_msg; - hsa_status_string(s, &hsa_err_msg); - throw(std::runtime_error{std::string("HSA error at ") + file + std::string(":") + - std::to_string(line) + std::string(" - ") + hsa_err_msg}); - } -}; +namespace { -#define CHECK_HSA_ERROR(cmd) checkHsaError((cmd), #cmd, __FILE__, __LINE__) +#define CHECK_HSA_ERROR(cmd) \ + if (auto s = (cmd); s != HSA_STATUS_SUCCESS) { \ + const char* hsa_err_msg; \ + hsa_status_string(s, &hsa_err_msg); \ + throw std::runtime_error{std::string("HSA error at " __FILE__ ":") + \ + std::to_string(__LINE__) + std::string(" - ") + hsa_err_msg}; \ + } #define CHECK_HSAKMT_SUCCESS(call, msg) \ do { \ @@ -59,39 +60,6 @@ auto checkHsaError = [](hsa_status_t s, const char* msg, const char* file, int l } \ } while (0) -#if 0 -inline void checkHipError(hipError_t err, const char* msg, const char* file, int line) -{ - if (err != hipSuccess) - { - std::cerr << "HIP error at " << file << ":" << line << " — " << msg << "\n" - << " Code: " << err << " (" << hipGetErrorString(err) << ")" << std::endl; - std::exit(EXIT_FAILURE); - } -} - -#define CHECK_HIP_ERROR(cmd) checkHipError((cmd), #cmd, __FILE__, __LINE__) - -// Allow access to peerDeviceId from deviceId -inline void EnablePeerAccess(int const deviceId, int const peerDeviceId) -{ - int canAccess; - CHECK_HIP_ERROR(hipDeviceCanAccessPeer(&canAccess, deviceId, peerDeviceId)); - if (!canAccess) - { - std::cerr << "Unable to enable peer access from GPU devices " << deviceId << " to " << peerDeviceId << "\n"; - } - - CHECK_HIP_ERROR(hipSetDevice(deviceId)); - hipError_t error = hipDeviceEnablePeerAccess(peerDeviceId, 0); - if (error != hipSuccess && error != hipErrorPeerAccessAlreadyEnabled) - { - std::cerr << "Unable to enable peer to peer access from " << deviceId << " to " << peerDeviceId << " (" - << hipGetErrorString(error) << ")\n"; - } -} -#endif - // HSA agents std::vector cpuAgents_; std::vector gpuAgents_; @@ -135,18 +103,38 @@ void SetUpKFD() { void CloseKFD() { (void)hsaKmtCloseKFD(); } -// Convert a logical deviceId index to the NVML device minor number -static const std::string getBusId(int deviceId) { +// PCI bus id ("domain:bus:dev.func") of a HIP device ordinal. +// Optionally returns the domain. +uint32_t getBusId(int deviceId, uint32_t* pdomain = nullptr) { // On most systems, the PCI bus ID comes back as in the 0000:00:00.0 // format. Still need to allocate proper space in case PCI domain goes // higher. - char busIdChar[] = "00000000:00:00.0"; - CHECK_HIP_ERROR(hipDeviceGetPCIBusId(busIdChar, sizeof(busIdChar), deviceId)); - // we need the hex in lower case format - for (size_t i = 0; i < sizeof(busIdChar); i++) { - busIdChar[i] = std::tolower(busIdChar[i]); + char busId[] = "00000000:00:00.0"; + CHECK_HIP_ERROR(hipDeviceGetPCIBusId(busId, sizeof(busId), deviceId)); + uint32_t domain = 0, bus = 0, dev = 0, func = 0; + if (std::sscanf(busId, "%x:%x:%x.%x", &domain, &bus, &dev, &func) != 4) { + MORI_APP_ERROR("Failed to parse PCI bus ID for device {}", deviceId); + return ~0u; + } + if (pdomain) *pdomain = domain; + return ((bus & 0xFF) << 8) | ((dev & 0x1F) << 3) | (func & 0x7); +} + +std::pair locIdAndDomainForNode(int node) { + uint32_t locId = ~0u, domain = ~0u; + std::string path = "/sys/class/kfd/kfd/topology/nodes/" + std::to_string(node) + "/properties"; + std::ifstream f(path); + if (!f.is_open()) return std::pair{locId, domain}; + std::string key, valStr; + // Read tokens as strings: some KFD properties (e.g. hive_id) are 64-bit values + // that would fail a numeric extraction and abort the scan early. + while (f >> key >> valStr) { + if (key == "location_id") + locId = std::strtol(valStr.c_str(), nullptr, 0); + else if (key == "domain") + domain = std::strtol(valStr.c_str(), nullptr, 0); } - return std::string(busIdChar); + return std::pair{locId, domain}; } // hsa_iterate_agents (SetUp) enumerates ALL physical GPU agents in HSA order, @@ -159,25 +147,20 @@ static const std::string getBusId(int deviceId) { // so the selection is correct in all cases. In the common HIP_VISIBLE_DEVICES= // 0..N-1 case this resolves to the identity map (BDF matches at the same index) // so the default path is behavior-identical. -static int gpuAgentIndexForHipDevice(int hipDeviceId) { +hsa_agent_t gpuAgentForHipDevice(int hipDeviceId) { static std::mutex mapMutex; static std::unordered_map hipToAgent; std::lock_guard lock(mapMutex); auto it = hipToAgent.find(hipDeviceId); - if (it != hipToAgent.end()) return it->second; + if (it != hipToAgent.end()) return gpuAgents_[it->second]; // BDF of the HIP device, parsed from its "domain:bus:device.function" string. - std::string busId = getBusId(hipDeviceId); - unsigned domain = 0, bus = 0, dev = 0, func = 0; - std::sscanf(busId.c_str(), "%x:%x:%x.%x", &domain, &bus, &dev, &func); - uint32_t hipBdf = ((bus & 0xFF) << 8) | ((dev & 0x1F) << 3) | (func & 0x7); - + uint32_t hipBdf = getBusId(hipDeviceId); // HSA_AMD_AGENT_INFO_BDFID exposes only the 16-bit bus/device/function, not the // PCI domain, so on a multi-segment machine two GPUs can share the same 16-bit // BDF. Only trust the match when it is UNIQUE; otherwise keep the identity // fallback rather than risk selecting the wrong agent. domain is parsed but // cannot be matched against the HSA side. - (void)domain; int match = hipDeviceId; // identity fallback (also correct when HIP and HSA order align) int nMatch = 0, firstMatch = -1; for (size_t a = 0; a < gpuAgents_.size(); ++a) { @@ -192,25 +175,12 @@ static int gpuAgentIndexForHipDevice(int hipDeviceId) { } if (nMatch == 1) match = firstMatch; hipToAgent[hipDeviceId] = match; - return match; + return gpuAgents_[match]; } -SdmaQueue::SdmaQueue(int localDeviceId, int remoteDeviceId, hsa_agent_t& localAgent, - uint32_t engineId) - : remoteDeviceId_(remoteDeviceId) { - // cachedWptr_(detail::gpuCallocUncachedShared()), - // committedWptr_(detail::gpuCallocUncachedShared()) { - int originalDeviceId; - - CHECK_HIP_ERROR(hipGetDevice(&originalDeviceId)); // Save the current device - - uint32_t localNodeId; - hsa_status_t status = hsa_agent_get_info(localAgent, HSA_AGENT_INFO_NODE, &localNodeId); - if (status != HSA_STATUS_SUCCESS) { - printf("Failure to get device info: 0x%x", status); - // return status; - } +} // namespace +SdmaQueue::SdmaQueue(uint32_t localNodeId, uint32_t engineId) { // Allocate SDMA queue buffer on device side, requires ExecuteAccess HsaMemFlags memFlags = {}; memFlags.ui32.NonPaged = 1; @@ -307,13 +277,36 @@ void AnvilLib::init() { }); } -bool AnvilLib::connect(int srcDeviceId, int dstDeviceId, int numChannels) { +// Map a HIP device ordinal to its KFD node id. +/*static*/ uint32_t AnvilLib::nodeForHipDevice(int hipDev) { + uint32_t nodeId = 0; + CHECK_HSA_ERROR(hsa_agent_get_info(gpuAgentForHipDevice(hipDev), HSA_AGENT_INFO_NODE, &nodeId)); + return nodeId; +} + +// Resolve the KFD topology node id of the given HIP device WITHOUT initializing HSA. +/*static*/ int AnvilLib::kfdNodeIdForHipDevice(int hipDev) { + uint32_t wantDomain = 0, wantLocId = getBusId(hipDev, &wantDomain); + // KFD node ids are contiguous from 0; stop at the first gap. + for (int node = 0;; node++) { + auto [locId, domain] = locIdAndDomainForNode(node); + if (locId == ~0u && domain == ~0u) break; + if (locId == wantLocId && domain == wantDomain) { + return node; + } + } + MORI_APP_ERROR("Failed to find KFD node for device {}", hipDev); + return -1; +} + +bool AnvilLib::connect(int srcNode, int dstNode, int numChannels) { std::lock_guard lock(channels_mutex_); // Spread the channels across the engines recommended for this peer link. On // MI350 the mask typically reports 2 engines per peer; on platforms with a // single recommended engine all channels share it. std::vector engines; - if (srcDeviceId == dstDeviceId) { + engines.reserve(2); + if (srcNode == dstNode) { // Loopback has no self io_link, so KFD recommends no engine. On gfx1250 each // engine holds only 6 queues and ROCr's blit queues already sit on the low // ones, so pinning every loopback channel to engine 0 hits NO_MEMORY at the @@ -321,51 +314,41 @@ bool AnvilLib::connect(int srcDeviceId, int dstDeviceId, int numChannels) { // Other archs are not engine-0-bound for loopback (gfx950: 8 queues/engine, // only engines 0-1 general) and regress if a channel lands on a busy engine, // so keep them pinned to engine 0. - if (isGfx1250(srcDeviceId)) { - uint32_t mask = getHostLinkEngineMask(srcDeviceId); + if (isGfx1250(srcNode)) { + uint32_t mask = getHostLinkEngineMask(srcNode); for (uint32_t b = 0; b < 32; ++b) { if (mask & (1u << b)) engines.push_back(b); } } if (engines.empty()) engines.push_back(0); } else { - uint32_t mask = getRecommendedEngineMask(srcDeviceId, dstDeviceId); + uint32_t mask = getRecommendedEngineMask(srcNode, dstNode); for (uint32_t b = 0; b < 32; ++b) { if (mask & (1u << b)) engines.push_back(b); } // Fall back to the static OAM table if KFD did not report a mask. if (engines.empty()) { - int e = getSdmaEngineId(srcDeviceId, dstDeviceId); + int e = getSdmaEngineId(srcNode, dstNode); engines.push_back(e); } } int numEngines = static_cast(engines.size()); // Queues live in this process-global singleton and are shared across every - // Context/comm for this device pair (getSdmaQueue keys on device ids, not on + // Context/comm for this node pair (getSdmaQueue keys on KFD node ids, not on // the comm), and are only reclaimed when the process exits. So create just the // shortfall: appending on every connect() would pile up unused duplicate // hardware queues (getSdmaQueue only ever indexes the first numChannels) and // eventually exhaust the per-engine queue slots. - auto& channels = sdma_channels_[std::make_pair(srcDeviceId, dstDeviceId)]; + auto& channels = sdma_channels_[std::make_pair(srcNode, dstNode)]; for (int c = static_cast(channels.size()); c < numChannels; ++c) { uint32_t engineId = engines[c % numEngines]; - channels.emplace_back(std::make_unique( - srcDeviceId, dstDeviceId, gpuAgents_[gpuAgentIndexForHipDevice(srcDeviceId)], engineId)); + channels.emplace_back(std::make_unique(srcNode, engineId)); } return true; } -uint32_t AnvilLib::getNodeId(int deviceId) { - uint32_t nodeId = 0; - CHECK_HSA_ERROR(hsa_agent_get_info(gpuAgents_[gpuAgentIndexForHipDevice(deviceId)], - HSA_AGENT_INFO_NODE, &nodeId)); - return nodeId; -} - -uint32_t AnvilLib::getRecommendedEngineMask(int srcDeviceId, int dstDeviceId) { - uint32_t srcNode = getNodeId(srcDeviceId), dstNode = getNodeId(dstDeviceId); - +uint32_t AnvilLib::getRecommendedEngineMask(int srcNode, int dstNode) { HsaNodeProperties props{}; if (hsaKmtGetNodeProperties(srcNode, &props) != HSAKMT_STATUS_SUCCESS || props.NumIOLinks == 0) { return 0; @@ -386,8 +369,7 @@ uint32_t AnvilLib::getRecommendedEngineMask(int srcDeviceId, int dstDeviceId) { // Engines KFD recommends for this GPU's link to a CPU node, i.e. the general // (non-xGMI) ones. Zero if the node reports no such link. -uint32_t AnvilLib::getHostLinkEngineMask(int srcDeviceId) { - const uint32_t srcNode = getNodeId(srcDeviceId); +uint32_t AnvilLib::getHostLinkEngineMask(int srcNode) { HsaNodeProperties props{}; if (hsaKmtGetNodeProperties(srcNode, &props) != HSAKMT_STATUS_SUCCESS || props.NumIOLinks == 0) { return 0; @@ -408,15 +390,15 @@ uint32_t AnvilLib::getHostLinkEngineMask(int srcDeviceId) { // gfx12.5+ (gfx1250): the only arch whose loopback channels are spread over // engines. See connect() for why. -bool AnvilLib::isGfx1250(int deviceId) { +bool AnvilLib::isGfx1250(int node) { HsaNodeProperties props{}; - if (hsaKmtGetNodeProperties(getNodeId(deviceId), &props) != HSAKMT_STATUS_SUCCESS) return false; + if (hsaKmtGetNodeProperties(node, &props) != HSAKMT_STATUS_SUCCESS) return false; return props.EngineId.ui32.Major == 12 && props.EngineId.ui32.Minor == 5; } -SdmaQueue* AnvilLib::getSdmaQueue(int srcDeviceId, int dstDeviceId, int channel_idx) { +SdmaQueue* AnvilLib::getSdmaQueue(int srcNode, int dstNode, int channel_idx) { std::lock_guard lock(channels_mutex_); - auto key = std::make_pair(srcDeviceId, dstDeviceId); + auto key = std::make_pair(srcNode, dstNode); auto it = sdma_channels_.find(key); if (it == sdma_channels_.end()) { return nullptr; @@ -437,24 +419,27 @@ AnvilLib& AnvilLib::getInstance() { return *instance; } -int AnvilLib::getOamId(int deviceId) { - std::string busId = getBusId(deviceId); - std::string file_str = "/sys/bus/pci/devices/" + busId + "/xgmi_physical_id"; - std::ifstream file(file_str); +int AnvilLib::getOamId(int node) { + auto [locId, domain] = locIdAndDomainForNode(node); + uint32_t bus = (locId >> 8) & 0xFF, dev = (locId >> 3) & 0x1F, func = locId & 0x7; + + char fpath[128]; + std::snprintf(fpath, sizeof(fpath), "/sys/bus/pci/devices/%04x:%02x:%02x.%01x/xgmi_physical_id", + domain, bus, dev, func); + std::ifstream file(fpath); int xgmi_physical_id; - if (file.is_open()) { - if (!(file >> xgmi_physical_id)) { - throw std::runtime_error("Failed to read xGMI physical id from file: " + file_str); - } - } else { - throw std::runtime_error("Failed to open file: " + file_str); + if (!file.is_open()) { + throw std::runtime_error("Failed to open file: " + std::string(fpath)); + } + if (!(file >> xgmi_physical_id)) { + throw std::runtime_error("Failed to read xGMI physical id from file: " + std::string(fpath)); } return xgmi_physical_id; } -int AnvilLib::getSdmaEngineId(int srcDeviceId, int dstDeviceId) { - int srcOamId = getOamId(srcDeviceId); - int dstOamId = getOamId(dstDeviceId); +int AnvilLib::getSdmaEngineId(int srcNode, int dstNode) { + int srcOamId = getOamId(srcNode); + int dstOamId = getOamId(dstNode); // Use even engines only return mi300xOamMap[srcOamId][dstOamId] * 2; diff --git a/src/cco/cco_init.cpp b/src/cco/cco_init.cpp index 72bd11a4f..1da0972a8 100644 --- a/src/cco/cco_init.cpp +++ b/src/cco/cco_init.cpp @@ -93,9 +93,11 @@ void ccoSdmaSetupCommQueues(ccoComm* comm, int requestedChannels) { comm->ctx->EnsureSdmaTransport(requestedChannels); comm->sdmaNumQueue = comm->ctx->SdmaChannels(); - // sdmaDevHandles is lsaSize × sdmaNumQueue, indexed by lsaRank. Assumes ranks - // bind 1:1 to GPUs within a node (rank lsa ⇒ GPU lsa). - int srcDeviceId = comm->hipDev; + // sdmaDevHandles is lsaSize × sdmaNumQueue, indexed by lsaRank (kernel-facing + // logical slot). The anvil queue lookup, however, is keyed on the host-global + // KFD node id of each GPU (exchanged in Context), so it stays correct even + // under sliced HIP_VISIBLE_DEVICES where HIP ordinals diverge from topology. + int srcNode = comm->ctx->LocalKfdNode(); size_t numSlots = static_cast(comm->lsaSize) * comm->sdmaNumQueue; HIP_RUNTIME_CHECK(hipMalloc(&comm->sdmaDevHandles, numSlots * sizeof(ccoSdmaQueueDeviceHandle*))); HIP_RUNTIME_CHECK( @@ -104,11 +106,11 @@ void ccoSdmaSetupCommQueues(ccoComm* comm, int requestedChannels) { for (int lsa = 0; lsa < comm->lsaSize; lsa++) { int pe = comm->myNodeStart + lsa; if (!comm->ctx->GetPeerCapabilities(pe).canSDMA) continue; - int dstDeviceId = lsa; + int dstNode = comm->ctx->KfdNodeId(pe); for (int q = 0; q < comm->sdmaNumQueue; q++) { // anvil returns its own SdmaQueueDeviceHandle*; cco stores it as an opaque // ccoSdmaQueueDeviceHandle* (layout-compatible, byte-copied by sizeof). - auto* handle = anvil::anvil.getSdmaQueue(srcDeviceId, dstDeviceId, q)->deviceHandle(); + auto* handle = anvil::anvil.getSdmaQueue(srcNode, dstNode, q)->deviceHandle(); HIP_RUNTIME_CHECK(hipMemcpy(&comm->sdmaDevHandles[lsa * comm->sdmaNumQueue + q], &handle, sizeof(handle), hipMemcpyHostToDevice)); } diff --git a/tests/cpp/cco/test_sdma_hip_dev_assign.cpp b/tests/cpp/cco/test_sdma_hip_dev_assign.cpp new file mode 100644 index 000000000..c2f035d87 --- /dev/null +++ b/tests/cpp/cco/test_sdma_hip_dev_assign.cpp @@ -0,0 +1,473 @@ +// Copyright © Advanced Micro Devices, Inc. All rights reserved. +// +// MIT License +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in all +// copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +// +// CLI usage: +// test_sdma_hip_dev_assign [--n_processes N] [--n_local_devices N] +// [--visible_devices D0,D1,D2,...] [--torch_style] +// --n_processes number of OS processes to fork (process_count). Default: +// the auto-detected total GPU count (or the length of +// --visible_devices, if given) divided by +// --n_local_devices (i.e. "use every visible GPU") — +// under --torch_style, NOT divided by --n_local_devices +// (which is ignored), since each process owns 1 GPU. +// --n_local_devices number of GPUs owned by each process (devices_per_process). +// Default: 1. Ignored when --torch_style is set. +// --visible_devices comma-separated physical device ordinals to carve up +// instead of the default contiguous 0,1,2,... +// assignment. Process i gets the n_local_devices-sized +// slice at offset i*n_local_devices, in list order — e.g. +// --visible_devices 1,2,4,5,7,3,0,6 --n_processes 4 +// --n_local_devices 2 gives processes HIP_VISIBLE_DEVICES +// "1,2", "4,5", "7,3", "0,6" respectively. The list length +// must equal n_processes * n_local_devices exactly. +// Under --torch_style, the FULL list (unsliced) is given +// to every process as HIP_VISIBLE_DEVICES, and only needs +// length >= n_processes (extra entries are simply +// visible-but-unused, matching real torch behavior). +// --torch_style launch torch-style instead of jax-style: n_processes +// processes, each owning exactly 1 GPU (n_local_devices +// is ignored). Every process sees the SAME (unsliced) +// visible-device set — the full --visible_devices list if +// given, else every native GPU (HIP_VISIBLE_DEVICES left +// unset) — and process i binds device ordinal i within +// that shared set via hipSetDevice(i), mirroring a real +// torch multi-process launch. globalRank = processIdx, +// worldSize = n_processes (no SPMT threading, since +// there's exactly one GPU per process). +// +// Examples: +// test_sdma_hip_dev_assign +// jax-style, all defaults: 1 GPU/process, one process per detected GPU. +// test_sdma_hip_dev_assign --n_processes 2 --n_local_devices 4 +// jax-style: 2 processes x 4 GPUs each (8-way world), contiguous +// HIP_VISIBLE_DEVICES "0,1,2,3" / "4,5,6,7". +// test_sdma_hip_dev_assign --visible_devices 1,2,4,5,7,3,0,6 \ +// --n_processes 4 --n_local_devices 2 +// jax-style with a custom device order: HIP_VISIBLE_DEVICES +// "1,2" / "4,5" / "7,3" / "0,6" for processes 0..3. +// test_sdma_hip_dev_assign --torch_style +// torch-style, all defaults: one process per detected GPU, every +// process sees every GPU (HIP_VISIBLE_DEVICES unset), rank i binds +// device ordinal i. +// test_sdma_hip_dev_assign --torch_style --n_processes 4 +// torch-style: 4 processes, each seeing all native GPUs, binding +// ordinals 0..3. +// test_sdma_hip_dev_assign --torch_style \ +// --visible_devices 1,2,4,5,7,3,0,6 --n_processes 4 +// torch-style with a restricted, shared device set: every process gets +// HIP_VISIBLE_DEVICES "1,2,4,5,7,3,0,6"; ranks 0..3 bind ordinals 0..3 +// within that list (physical devices 1, 2, 4, 5). +// +// NOTE: run in fork mode (default), so no HIP call happens before the slice +// is applied. Requires MORI_ENABLE_SDMA=1; otherwise the comm has no SDMA +// queues and the test SKIPs. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "cco_test_harness.hpp" + +// ── inlined from sdma_allgather_common.hpp ─────────────────────────────── +// +// All-gather layout (COUNT floats per rank): +// input[i] = rank * 1000 + i (this rank's chunk) +// gather[s*COUNT + i] = s * 1000 + i for every source rank s +// Each rank SDMA-puts its own input chunk into slot `lsaRank` of every peer's +// gather window (self included, via the loopback queue), then quiet()s. When +// all ranks have done so, every rank's gather window holds all N chunks. + +static const size_t SDMA_ALLGATHER_COUNT = 16384; +static const size_t SDMA_ALLGATHER_VMM_SIZE = 16ULL * 1024 * 1024; + +__global__ void SdmaAllGatherJaxKernel(mori::cco::ccoWindow_t gather, mori::cco::ccoWindow_t input, + size_t chunkBytes, mori::cco::ccoDevComm devComm) { + int myRank = devComm.lsaRank; + int nRanks = devComm.lsaSize; + + int p = threadIdx.x; + if (p >= nRanks) return; + mori::cco::ccoSdma sdma{devComm}; + sdma.put(p, gather, myRank * chunkBytes, input, 0, chunkBytes); + sdma.quiet(p); +} + +// Per-rank state threaded between setup, launch, and verify. +struct SdmaAllGatherCtx { + mori::cco::ccoComm* comm{nullptr}; + mori::cco::ccoDevComm devComm{}; + mori::cco::ccoWindow_t inputWin{nullptr}; + mori::cco::ccoWindow_t gatherWin{nullptr}; + void* inputBuf{nullptr}; + void* gatherBuf{nullptr}; + hipStream_t stream{nullptr}; + size_t chunkBytes{0}; + int rank{0}; + int nranks{0}; + bool hasSdma{false}; // false => no SDMA queues; test SKIPs the data phase +}; + +// Create comm + windows + devComm and (when SDMA is available) a stream. +// Assumes the caller has already bound this rank's GPU. Returns 0 on +// success; on failure returns nonzero and the caller should bail out +// (nothing to tear down that a leaked process/thread exit won't reclaim). +static int SdmaAllGatherSetup(int rank, int nranks, const mori::cco::ccoUniqueId& uid, + SdmaAllGatherCtx* ctx) { + using namespace mori::cco; + ctx->rank = rank; + ctx->nranks = nranks; + + int hipDev = -1, visibleDevices = 0; + HIP_CHECK(hipGetDevice(&hipDev)); + HIP_CHECK(hipGetDeviceCount(&visibleDevices)); + printf("[rank %d/%d] pid=%d hipDev=%d visibleDevices=%d\n", rank, nranks, getpid(), hipDev, + visibleDevices); + fflush(stdout); + + if (ccoCommCreate(uid, nranks, rank, SDMA_ALLGATHER_VMM_SIZE, &ctx->comm) != 0) { + fprintf(stderr, "[rank %d] CommCreate failed\n", rank); + return 1; + } + + ctx->chunkBytes = SDMA_ALLGATHER_COUNT * sizeof(float); + const size_t gatherBytes = ctx->chunkBytes * nranks; + if (ccoMemAlloc(ctx->comm, ctx->chunkBytes, &ctx->inputBuf) != 0 || + ccoMemAlloc(ctx->comm, gatherBytes, &ctx->gatherBuf) != 0) { + fprintf(stderr, "[rank %d] MemAlloc failed\n", rank); + return 1; + } + + std::vector hostInput(SDMA_ALLGATHER_COUNT); + for (size_t i = 0; i < SDMA_ALLGATHER_COUNT; i++) + hostInput[i] = static_cast(rank * 1000 + i); + HIP_CHECK(hipMemcpy(ctx->inputBuf, hostInput.data(), ctx->chunkBytes, hipMemcpyHostToDevice)); + HIP_CHECK(hipMemset(ctx->gatherBuf, 0xff, gatherBytes)); + + if (ccoWindowRegister(ctx->comm, ctx->inputBuf, ctx->chunkBytes, &ctx->inputWin) != 0 || + ccoWindowRegister(ctx->comm, ctx->gatherBuf, gatherBytes, &ctx->gatherWin) != 0) { + fprintf(stderr, "[rank %d] WindowRegister failed\n", rank); + return 1; + } + + // SDMA needs no GDA connectivity; the signal pool is materialized whenever the + // comm has SDMA queues (set up in ccoCommCreate for canSDMA peers). + ccoDevCommRequirements reqs = CCO_DEV_COMM_REQUIREMENTS_INITIALIZER; + reqs.gdaConnectionType = CCO_GDA_CONNECTION_NONE; + reqs.gdaContextCount = 0; + reqs.gdaSignalCount = 0; + reqs.gdaCounterCount = 0; + if (ccoDevCommCreate(ctx->comm, &reqs, &ctx->devComm) != 0) { + fprintf(stderr, "[rank %d] DevCommCreate failed\n", rank); + return 1; + } + + ctx->hasSdma = ctx->devComm.sdma.sdmaNumQueue != 0; + if (ctx->hasSdma) { + HIP_CHECK(hipStreamCreate(&ctx->stream)); + } else { + printf("[rank %d] SKIP — no SDMA queues (set MORI_ENABLE_SDMA=1)\n", rank); + } + return 0; +} + +// Verify the gathered result (when SDMA ran) and tear the comm down. Returns 0 +// on pass/skip, 1 on mismatch. +static int VerifyAndTeardown(SdmaAllGatherCtx* ctx) { + using namespace mori::cco; + bool ok = true; + + if (ctx->hasSdma) { + const size_t gatherBytes = ctx->chunkBytes * ctx->nranks; + std::vector host(SDMA_ALLGATHER_COUNT * ctx->nranks); + HIP_CHECK(hipMemcpy(host.data(), ctx->gatherBuf, gatherBytes, hipMemcpyDeviceToHost)); + for (int s = 0; s < ctx->nranks && ok; s++) { + for (size_t i = 0; i < SDMA_ALLGATHER_COUNT; i++) { + float expected = static_cast(s * 1000 + i); + if (host[s * SDMA_ALLGATHER_COUNT + i] != expected) { + fprintf(stderr, "[rank %d] ALLGATHER mismatch [src=%d][%zu]: got %.0f expected %.0f\n", + ctx->rank, s, i, host[s * SDMA_ALLGATHER_COUNT + i], expected); + ok = false; + break; + } + } + } + printf("[rank %d] allgather %s\n", ctx->rank, ok ? "PASSED" : "FAILED"); + HIP_CHECK(hipStreamDestroy(ctx->stream)); + } + + ccoDevCommDestroy(ctx->comm, &ctx->devComm); + ccoWindowDeregister(ctx->comm, ctx->gatherWin); + ccoWindowDeregister(ctx->comm, ctx->inputWin); + ccoMemFree(ctx->comm, ctx->gatherBuf); + ccoMemFree(ctx->comm, ctx->inputBuf); + ccoCommDestroy(ctx->comm); + return ok ? 0 : 1; +} + +// ── per-thread (per local GPU) rank driver ─────────────────────────────── + +struct ThreadResult { + int rank{-1}; + bool passed{false}; + char detail[256]{}; +}; + +// Runs entirely on its own thread. `rank`/`nranks` here ARE the CCO GPU +// rank/world size (globalRank / process_count*devices_per_process) — unlike +// run_test below, no reinterpretation here. `localDevice` is this thread's +// ordinal within the process's (already-sliced) HIP_VISIBLE_DEVICES list +// (0..devices_per_process-1); hipSetDevice is thread-local, so each thread +// must bind its own device before making any other HIP call. +static void RunLocalRank(int localDevice, int rank, int nranks, const mori::cco::ccoUniqueId& uid, + ThreadResult* result) { + // cco_test_harness.hpp's HIP_CHECK macro reports failures via the + // process-wide g_rank global (not thread-local). Setting it here is + // best-effort for diagnostics only: under concurrent threads the printed + // rank in a HIP_CHECK failure message can race, but HIP_CHECK always + // _exit(1)s the whole process immediately regardless, so this never + // affects correctness — only which rank number an error message blames. + g_rank = rank; + result->rank = rank; + result->passed = false; + + HIP_CHECK(hipSetDevice(localDevice)); + + SdmaAllGatherCtx ctx; + if (SdmaAllGatherSetup(rank, nranks, uid, &ctx) != 0) { + snprintf(result->detail, sizeof(result->detail), "setup failed"); + return; + } + + if (ctx.hasSdma) { + mori::cco::ccoBarrierAll(ctx.comm); + SdmaAllGatherJaxKernel<<<1, 64, 0, ctx.stream>>>(ctx.gatherWin, ctx.inputWin, ctx.chunkBytes, + ctx.devComm); + HIP_CHECK(hipStreamSynchronize(ctx.stream)); + mori::cco::ccoBarrierAll(ctx.comm); + } + + bool ok = VerifyAndTeardown(&ctx) == 0; + result->passed = ok; + snprintf(result->detail, sizeof(result->detail), ok ? "OK" : "allgather mismatch"); +} + +// devices_per_process: not known to cco_test_harness.hpp's run_test(rank, +// nranks, uid) signature, so main() parses it from --n_local_devices into +// this global before handing off to ccoTestForkMode. +static int g_devicesPerProcess = 1; + +// Optional custom physical-device ordinals from --visible_devices, in the +// order they should be carved up across processes. Empty means "use the +// default contiguous 0,1,2,... assignment" (see run_test below). +static std::vector g_visibleDevices; + +// --torch_style: every process sees the same (unsliced) device set and picks +// device ordinal == its rank, instead of jax-style per-process slicing. See +// run_test below. +static bool g_torchStyle = false; + +// Invoked ONCE PER FORKED PROCESS by cco_test_harness.hpp's fork mode. Here +// `processIdx`/`numProcesses` are this PROCESS's index / the total process +// count (process_count) — NOT a GPU rank/world size like other tests that +// use this harness. See the file header comment for why. +int run_test(int processIdx, int numProcesses, const mori::cco::ccoUniqueId& uid) { + g_rank = processIdx; + + if (g_torchStyle) { + // torch-style: every process shares the SAME (unsliced) visible-device + // set and picks device ordinal == its own rank, mirroring a real torch + // multi-process launch where all GPUs are visible to every process. + // --n_local_devices is ignored (each process owns exactly 1 GPU). + if (!g_visibleDevices.empty()) { + std::string vis; + for (size_t i = 0; i < g_visibleDevices.size(); i++) { + if (i) vis += ","; + vis += std::to_string(g_visibleDevices[i]); + } + setenv("HIP_VISIBLE_DEVICES", vis.c_str(), /*overwrite=*/1); + } + // Else: leave HIP_VISIBLE_DEVICES unset, so native full visibility applies. + + ThreadResult result; + RunLocalRank(processIdx, processIdx, numProcesses, uid, &result); + printf("[proc %d][rank %d] %s: %s\n", processIdx, result.rank, + result.passed ? "PASSED" : "FAILED", result.detail); + return result.passed ? 0 : 1; + } + + int devicesPerProcess = g_devicesPerProcess; + // jax-style: slice HIP visibility to this process's GPU range BEFORE any + // further HIP call, so local ordinals 0..devicesPerProcess-1 map to the + // DISTINCT physical GPUs [gpuOffset, gpuOffset+devicesPerProcess) — or, if + // --visible_devices was given, to the devicesPerProcess-sized slice of that + // list at the same offset (main() already validated the list length). + // Leave ROCR_VISIBLE_DEVICES unset so HSA still sees all GPUs (required by + // the KFD node-id SDMA path). + const int gpuOffset = processIdx * devicesPerProcess; + std::string vis; + for (int i = 0; i < devicesPerProcess; i++) { + if (i) vis += ","; + int physicalDev = g_visibleDevices.empty() ? gpuOffset + i : g_visibleDevices[gpuOffset + i]; + vis += std::to_string(physicalDev); + } + setenv("HIP_VISIBLE_DEVICES", vis.c_str(), /*overwrite=*/1); + + const int worldSize = numProcesses * devicesPerProcess; + std::vector results(devicesPerProcess); + std::vector threads; + threads.reserve(devicesPerProcess); + for (int local = 0; local < devicesPerProcess; local++) { + int globalRank = gpuOffset + local; + threads.emplace_back(RunLocalRank, local, globalRank, worldSize, std::cref(uid), + &results[local]); + } + for (auto& t : threads) t.join(); + + int fail = 0; + for (auto& r : results) { + printf("[proc %d][rank %d] %s: %s\n", processIdx, r.rank, r.passed ? "PASSED" : "FAILED", + r.detail); + if (!r.passed) fail++; + } + return fail > 0 ? 1 : 0; +} + +// Total KFD GPU nodes visible on the host, via sysfs (no HIP calls, so this +// is safe to call before HIP_VISIBLE_DEVICES is set for any process/thread). +static int DetectTotalGpuCount() { + int count = 0; + for (int i = 0; i < 64; i++) { + char path[128]; + snprintf(path, sizeof(path), "/sys/class/kfd/kfd/topology/nodes/%d/gpu_id", i); + FILE* f = fopen(path, "r"); + if (!f) break; + unsigned long gpuId = 0; + if (fscanf(f, "%lu", &gpuId) == 1 && gpuId != 0) count++; + fclose(f); + } + return count; +} + +static void PrintUsageAndExit(const char* argv0) { + fprintf(stderr, + "usage: %s [--n_processes N] [--n_local_devices N] [--visible_devices D0,D1,...] " + "[--torch_style]\n" + " --n_processes N number of OS processes to fork (default: " + "auto-detected GPU count, or --visible_devices length, / --n_local_devices; " + "not divided under --torch_style)\n" + " --n_local_devices N GPUs owned by each process (default: 1; ignored under " + "--torch_style)\n" + " --visible_devices LIST comma-separated physical device ordinals to carve up " + "instead of the default contiguous 0,1,2,... assignment; length must equal " + "n_processes * n_local_devices (jax-style) or be >= n_processes (--torch_style, " + "unsliced)\n" + " --torch_style torch-style launch: n_processes processes, each owning " + "1 GPU from a SHARED (unsliced) visible-device set, binding ordinal == rank\n", + argv0); + exit(1); +} + +// Parses a comma-separated list of non-negative integers, e.g. "1,2,4,5,7,3,0,6". +// Returns false (leaving *out unspecified) on any malformed token or empty list. +static bool ParseVisibleDevicesList(const char* s, std::vector* out) { + out->clear(); + std::string str(s); + size_t pos = 0; + while (pos <= str.size()) { + size_t comma = str.find(',', pos); + std::string tok = str.substr(pos, comma == std::string::npos ? std::string::npos : comma - pos); + if (tok.empty() || + !std::all_of(tok.begin(), tok.end(), [](unsigned char c) { return isdigit(c); })) + return false; + out->push_back(atoi(tok.c_str())); + if (comma == std::string::npos) break; + pos = comma + 1; + } + return !out->empty(); +} + +int main(int argc, char** argv) { + int nProcesses = -1; + + for (int i = 1; i < argc; i++) { + if (!strcmp(argv[i], "--n_processes") && i + 1 < argc) { + nProcesses = atoi(argv[++i]); + } else if (!strcmp(argv[i], "--n_local_devices") && i + 1 < argc) { + g_devicesPerProcess = std::max(1, atoi(argv[++i])); + } else if (!strcmp(argv[i], "--visible_devices") && i + 1 < argc) { + if (!ParseVisibleDevicesList(argv[++i], &g_visibleDevices)) { + fprintf(stderr, + "--visible_devices: invalid device list '%s' (expected comma-separated " + "non-negative integers)\n", + argv[i]); + return 1; + } + } else if (!strcmp(argv[i], "--torch_style")) { + g_torchStyle = true; + } else if (!strcmp(argv[i], "-h") || !strcmp(argv[i], "--help")) { + PrintUsageAndExit(argv[0]); + } else { + fprintf(stderr, "unknown argument: %s\n", argv[i]); + PrintUsageAndExit(argv[0]); + } + } + + const int totalDevices = + g_visibleDevices.empty() ? DetectTotalGpuCount() : static_cast(g_visibleDevices.size()); + + if (nProcesses < 0) { + // --torch_style ignores --n_local_devices (always 1 GPU/process), so the + // "use every visible GPU" default is just totalDevices, not divided. + nProcesses = + g_torchStyle ? std::max(1, totalDevices) : std::max(1, totalDevices / g_devicesPerProcess); + } else if (nProcesses < 1) { + fprintf(stderr, "--n_processes must be >= 1\n"); + return 1; + } + + if (g_torchStyle) { + // Every process shares the same visible set and picks ordinal == rank, so + // it only needs >= n_processes entries (unlike jax-style's exact-match + // slicing) -- extra entries are simply visible-but-unused. + if (nProcesses > totalDevices) { + fprintf(stderr, "--torch_style: --n_processes(%d) exceeds available devices (%d)\n", + nProcesses, totalDevices); + return 1; + } + } else if (!g_visibleDevices.empty() && + static_cast(g_visibleDevices.size()) != nProcesses * g_devicesPerProcess) { + fprintf(stderr, + "--visible_devices has %zu entries but --n_processes(%d) * --n_local_devices(%d) " + "= %d\n", + g_visibleDevices.size(), nProcesses, g_devicesPerProcess, + nProcesses * g_devicesPerProcess); + return 1; + } + + return ccoTestForkMode(nProcesses, "CCO SDMA allgather (various dev assignments)", + "/tmp/cco_sdma_allgather_uid", 0); +} diff --git a/tools/run_cco_tests.sh b/tools/run_cco_tests.sh index a57a5087b..02b6c9f9c 100755 --- a/tools/run_cco_tests.sh +++ b/tools/run_cco_tests.sh @@ -15,6 +15,10 @@ for bin in tests/cpp/cco/test_*; do # SDMA tests need MORI_ENABLE_SDMA=1 to build queues (else they self-SKIP); # run separately in CI with that env set, so skip them in this default sweep. test_sdma_put|test_sdma_get|test_sdma_put_mt|test_sdma_block|test_sdma_edge) continue ;; + # Takes --n_processes/--n_local_devices/--visible_devices instead of a + # bare positional nranks arg, and (like its SDMA siblings above) needs + # MORI_ENABLE_SDMA=1 to build queues + test_sdma_hip_dev_assign) continue ;; esac # GDA-FULL tests need intranode cross-rail RDMA (FULL connections). On runners # where cross-rail is unavailable they can't pass; skip when From 81c3fc472279ed6c9ba5ab15dd7a242b637c87e7 Mon Sep 17 00:00:00 2001 From: Pavel Emeliyanenko Date: Tue, 1 Sep 2026 15:30:05 +0000 Subject: [PATCH 2/2] restored LocalRankInNode func --- include/mori/application/context/context.hpp | 4 ++++ src/application/context/context.cpp | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/include/mori/application/context/context.hpp b/include/mori/application/context/context.hpp index f67c1d859..be2132577 100644 --- a/include/mori/application/context/context.hpp +++ b/include/mori/application/context/context.hpp @@ -103,6 +103,10 @@ class Context { int KfdNodeId(int destRank) const { return peerInfos[destRank].kfdNodeId; } int LocalKfdNode() const { return peerInfos[LocalRank()].kfdNodeId; } + // Local rank within this host (0-based index among same-host peers). + // Derived from peerInfos rather than a separately-tracked member. + int LocalRankInNode() const { return peerInfos[LocalRank()].rankInNode; } + RdmaContext* GetRdmaContext() const { return rdmaContext.get(); } RdmaDeviceContext* GetRdmaDeviceContext() const { return rdmaDeviceContext.get(); } const std::vector>& GetAllRdmaDeviceContexts() const { diff --git a/src/application/context/context.cpp b/src/application/context/context.cpp index dd2c823ff..f742c4da3 100644 --- a/src/application/context/context.cpp +++ b/src/application/context/context.cpp @@ -88,7 +88,7 @@ void Context::BuildInitialEndpoints() { // (C) Build & connect the initial QP set (worldSize × numQpPerPe). // Under MORI_ENABLE_RAIL_ONLY, cross-rail cross-node peers get stubs. - int myRail = peerInfos[LocalRank()].rankInNode; + int myRail = LocalRankInNode(); auto shouldConnect = [&](int i) { if (transportTypes[i] != TransportType::RDMA) return false; if (railOnly && !peerInfos[i].sameHost && peerInfos[i].rankInNode != myRail) return false;