Fix vnode/nolocal/vClique topology to preserve real pids (#2153)

Summary:
Pull Request resolved: #2153

Fix IpcRegCache key collision caused by fake pids in virtual topology modes:
- initRankTopologyVnode and initRankTopologyNolocal never set pid (default -1), causing all ranks on the same vnode to share the same gPid(). This made IpcRegCache deduplicate cuMem IPC imports incorrectly, corrupting AllGatherP data with ncclMemAlloc buffers.
- initRankStatesTopology now always does bootstrap allgather when bootstrap is available (prod and dist test paths), falling back to fake topology only when bootstrap is nullptr (unit tests, benchmarks).
- setRankStatesTopologies gains vClique host override alongside existing vnode/nolocal overrides, so the bootstrap path handles all virtual topology modes.
- initRankTopologyVnode/initRankTopologyNolocal now set getpid() for the local rank, so even non-bootstrap callers get a real pid.
- Test makeCtranComm (CtranDistTestUtils) simplified to always use initRankStatesTopology(bootstrap) — virtual topology overrides are handled internally.
- Added VnodeTopologyHasRealPid unit test and fixed gPidTest to match host:pid format.
- Enabled CreateVCliqueSizeFromNcclComm and added host/gPid validation with oobAllGather.

Reviewed By: dsjohns2

Differential Revision: D101393262

fbshipit-source-id: dc7e3e717199ad9a9eb58079bec81e0e7ecd90cd

Author: minsii

comms/ctran/commstate/CommStateX.cc

@@ -38,6 +38,14 @@ namespace {
     CHECK_VALID_RANK(rank, rankStates_.size());            \
   } while (0)
 
+// Return vnode nLocalRanks from global debug CVAR, or 0 if not set.
+int getGlobalVNodeNLocalRanks() {
+  if (NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::vnode) {
+    return static_cast<int>(NCCL_COMM_STATE_DEBUG_TOPO_VNODE_NLOCALRANKS);
+  }
+  return 0;
+}
+
 } // namespace
 
 CommStateX::CommStateX(
@@ -62,7 +70,8 @@ CommStateX::CommStateX(
       noLocal_(
           noLocal ||
           NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::nolocal),
-      vCliqueSize_(vCliqueSize) {
+      vCliqueSize_(
+          vCliqueSize > 0 ? vCliqueSize : getGlobalVNodeNLocalRanks()) {
   setRankStatesTopologies(std::move(rankTopologies));
   setCommRankToWorldRanks(std::move(commRanksToWorldRanks));
 }
@@ -75,24 +84,26 @@ void CommStateX::initRankTopologyNolocal() {
   for (int r = 0; r < nRanks_; ++r) {
     auto& rankState = rankStates_.at(r);
     rankState.rank = r;
+    rankState.pid = (r == rank_) ? getpid() : -1;
     rankState.nodeId = r;
     rankState.localRank = 0;
     rankState.nLocalRanks = 1;
     rankState.localRankToRanks.assign(1, r);
     const std::string nolocalHost("nolocal_node_" + std::to_string(r));
     rankState.host = nolocalHost;
-    hostToRanks_[nolocalHost].emplace_back(r);
     nodeRanks_[rankState.nodeId].emplace_back(rankState.rank);
   }
 }
 
 void CommStateX::initRankTopologyVnode(const int nLocalRanks) {
   rankStates_.resize(nRanks_);
-  nodeRanks_.resize(nRanks_);
+  const int nNodes = (nRanks_ + nLocalRanks - 1) / nLocalRanks;
+  nodeRanks_.resize(nNodes);
   for (int r = 0; r < nRanks_; ++r) {
     auto& rankState = rankStates_.at(r);
     rankState.nLocalRanks = nLocalRanks;
     rankState.rank = r;
+    rankState.pid = (r == rank_) ? getpid() : -1;
     rankState.nodeId = r / rankState.nLocalRanks;
     rankState.localRank = r % rankState.nLocalRanks;
     rankState.localRankToRanks.assign(
@@ -103,20 +114,21 @@ void CommStateX::initRankTopologyVnode(const int nLocalRanks) {
     const std::string vNodeHost(
         "vnode_node_" + std::to_string(rankState.nodeId));
     rankState.host = vNodeHost;
-    hostToRanks_[vNodeHost].emplace_back(r);
     nodeRanks_[rankState.nodeId].emplace_back(r);
   }
 }
 
 void CommStateX::initRankStatesTopology(meta::comms::IBootstrap* bootstrap) {
-  if (noLocal_) {
+  if (bootstrap == nullptr) {
+    FB_CHECKTHROW_EX(
+        nRanks_ == 1,
+        rank_,
+        commHash_,
+        commDesc_,
+        "bootstrap is required for multi-rank topology initialization");
     initRankTopologyNolocal();
     return;
   }
-  if (vCliqueSize_ > 0) {
-    initRankTopologyVnode(vCliqueSize_);
-    return;
-  }
   auto myTopo = ctran::commstate::loadTopology(rank_, NCCL_TOPO_FILE_PATH);
   if (!myTopo) {
     FB_CHECKTHROW_EX(
@@ -349,27 +361,32 @@ void CommStateX::setRankStatesTopologies(
     std::vector<RankTopology> rankTopologies) {
   rankStates_.clear();
   nodeRanks_.clear();
-  hostToRanks_.clear();
-
   rankTopologies_ = rankTopologies;
 
+  // nodeGroupMap uses a virtual grouping key for node assignment.
+  // Virtual topology overrides (nolocal, vnode, vClique) change node grouping
+  // without altering the real hostname stored in rankState.host.
+  std::unordered_map<std::string, std::vector<int>> nodeGroupMap;
+
   for (const auto& rankTopology : rankTopologies_) {
-    std::string host(rankTopology.host);
+    const std::string host(rankTopology.host);
     const std::string rtsw(rankTopology.rtsw);
     const std::string su(rankTopology.su);
     const std::string dc(rankTopology.dc);
     const std::string zone(rankTopology.zone);
 
-    if (NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::nolocal) {
-      host = "nolocal_node_" + std::to_string(rankTopology.rank);
-    } else if (
-        NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::vnode) {
-      host =
-          "vnode_node_" +
-          std::to_string(
-              rankTopology.rank / NCCL_COMM_STATE_DEBUG_TOPO_VNODE_NLOCALRANKS);
+    // Determine node grouping key — may differ from real host for virtual
+    // topos. noLocal_ treats each rank as its own node; vCliqueSize_ (set from
+    // either explicit vClique param or NCCL_COMM_STATE_DEBUG_TOPO=vnode) groups
+    // ranks into virtual nodes of the given size.
+    std::string nodeGroupKey = host;
+    if (noLocal_) {
+      nodeGroupKey = "nolocal_node_" + std::to_string(rankTopology.rank);
+    } else if (vCliqueSize_ > 0) {
+      nodeGroupKey =
+          "vnode_" + std::to_string(rankTopology.rank / vCliqueSize_);
     }
-    hostToRanks_[host].emplace_back(rankTopology.rank);
+    nodeGroupMap[nodeGroupKey].emplace_back(rankTopology.rank);
 
     RankState state;
     state.rank = rankTopology.rank;
@@ -393,21 +410,21 @@ void CommStateX::setRankStatesTopologies(
       }
     }
 
-    state.nodeId = hostToRanks_.size() - 1;
-    state.localRank = hostToRanks_.at(host).size() - 1;
+    state.nodeId = nodeGroupMap.size() - 1;
+    state.localRank = nodeGroupMap.at(nodeGroupKey).size() - 1;
 
     rankStates_.push_back(std::move(state));
   }
 
-  // Populate nodeRanks_ after setup hostToRanks_ so we know how many nodes
+  // Populate nodeRanks_ after the first pass so we know how many nodes
   // there are to resize nodeRanks_
-  nodeRanks_.resize(hostToRanks_.size());
+  nodeRanks_.resize(nodeGroupMap.size());
   for (const auto& state : rankStates_) {
     nodeRanks_[state.nodeId].emplace_back(state.rank);
   }
 
   for (auto& state : rankStates_) {
-    state.localRankToRanks = hostToRanks_.at(state.host);
+    state.localRankToRanks = nodeRanks_.at(state.nodeId);
     state.nLocalRanks = state.localRankToRanks.size();
   }
 
@@ -489,7 +506,7 @@ int CommStateX::nNodes() const {
     }
     return nvlDomainRanks_.size();
   } else {
-    return hostToRanks_.size();
+    return nodeRanks_.size();
   }
 }
 

comms/ctran/commstate/CommStateX.h

@@ -266,12 +266,9 @@ class CommStateX {
   // and an associated communicator.
   std::vector<int> commRanksToWorldRanks_{};
 
-  // e.g map<host-name, localRankToRank>
-  // e.g map<host1: [0, 1, 2, 3], host2: [4, 5, 6, 7]>
-  std::unordered_map<std::string, std::vector<int>> hostToRanks_{};
-
-  // similar to hostToRanks_ but access by nodeId
-  // e.g vector<0: [0, 1, 2, 3], 1: [0, 1, 2, 3]>
+  // Node grouping: nodeRanks_[nodeId] = [rank0, rank1, ...]
+  // For virtual topologies (nolocal, vnode, vClique), node grouping may differ
+  // from physical host grouping.
   std::vector<std::vector<int>> nodeRanks_{};
 
   // similar to nodeRanks, but at nvlDomain level. e.g vector<0: [0, 1, 2, 3],

comms/ctran/commstate/tests/CommStateXTest.cc

@@ -9,6 +9,7 @@
 #include <gtest/gtest.h>
 
 #include "comms/ctran/commstate/CommStateX.h"
+#include "comms/ctran/tests/VerifyCommStateXUtil.h"
 #include "comms/testinfra/TestXPlatUtils.h"
 
 namespace ncclx {
@@ -552,49 +553,114 @@ TEST(CommStateXTest, CommRankToWorldRanks) {
   EXPECT_EQ(commState->gRank(3), 7);
 }
 
-TEST(CommStateXTest, gPidTest) {
-  const int nRanks = 4;
-  const int cudaDev = 0;
-  const int cudaArch = 90;
-  const int64_t busId = 25;
-  const uint64_t commHash = 0;
-  const std::string kSu;
+// Parameterized test: verify host()/gPid() preserve real values and node
+// grouping is correct across all virtual topology modes.
+enum class TopoMode { kSystem, kNolocal, kVnode, kVClique };
 
-  std::vector<RankTopology> rankTopologies{};
-  rankTopologies.emplace_back(
-      createRankTopology(0, kDc, kZone, kSu, kRtsw0, kHost0, -1, 1000));
-  rankTopologies.emplace_back(
-      createRankTopology(1, kDc, kZone, kSu, kRtsw0, kHost0, -1, 1001));
-  rankTopologies.emplace_back(
-      createRankTopology(2, kDc, kZone, kSu, kRtsw0, kHost1, -1, 2000));
-  rankTopologies.emplace_back(
-      createRankTopology(3, kDc, kZone, kSu, kRtsw0, kHost1, -1, 2001));
+struct TopoTestParam {
+  TopoMode mode;
+  int vCliqueSize; // only used for kVClique
+  int expectedNNodes;
+  int expectedNLocalRanks; // for rank 0
+};
 
-  for (int rank = 0; rank < nRanks; ++rank) {
-    auto commState = std::make_unique<CommStateX>(
-        rank,
-        nRanks,
-        cudaDev,
-        cudaArch,
-        busId,
-        commHash,
-        rankTopologies,
-        std::vector<int>{});
-
-    // Test gPid() for default (current) rank
-    std::string expectedGPid = std::string(rankTopologies[rank].host) + ":" +
-        std::to_string(rankTopologies[rank].pid);
-    EXPECT_EQ(commState->gPid(), expectedGPid);
-
-    // Test gPid(rank) for all ranks
-    for (int r = 0; r < nRanks; ++r) {
-      std::string expected = std::string(rankTopologies[r].host) + ":" +
-          std::to_string(rankTopologies[r].pid);
-      EXPECT_EQ(commState->gPid(r), expected);
+std::string topoTestName(const testing::TestParamInfo<TopoTestParam>& info) {
+  switch (info.param.mode) {
+    case TopoMode::kSystem:
+      return "system";
+    case TopoMode::kNolocal:
+      return "nolocal";
+    case TopoMode::kVnode:
+      return "vnode4";
+    case TopoMode::kVClique:
+      return "vclique" + std::to_string(info.param.vCliqueSize);
+  }
+  return "unknown";
+}
+
+class GpidTopoTest : public ::testing::TestWithParam<TopoTestParam> {
+ protected:
+  static constexpr int kRanksPerHost = 4;
+  static constexpr int kNumHosts = 2;
+  static constexpr int kNRanks = kRanksPerHost * kNumHosts;
+  static constexpr int kCudaDev = 0;
+  static constexpr int kCudaArch = 90;
+  static constexpr int64_t kBusId = 25;
+  static constexpr uint64_t kCommHash = 0;
+
+  std::vector<RankTopology> makeRankTopologies() {
+    const std::string kSu;
+    const std::array<std::pair<const char*, const char*>, kNumHosts> hostInfo =
+        {{{kHost0, kRtsw0}, {kHost1, kRtsw1}}};
+    std::vector<RankTopology> topos;
+    topos.reserve(kNRanks);
+    for (int r = 0; r < kNRanks; r++) {
+      const auto& [host, rtsw] = hostInfo[r / kRanksPerHost];
+      const int pid = 1000 * (r / kRanksPerHost + 1) + r % kRanksPerHost;
+      topos.emplace_back(
+          createRankTopology(r, kDc, kZone, kSu, rtsw, host, -1, pid));
     }
+    return topos;
+  }
+};
+
+TEST_P(GpidTopoTest, HostAndGpidPreserved) {
+  const auto& [mode, vCliqueSize, expectedNNodes, expectedNLocalRanks] =
+      GetParam();
+  auto rankTopologies = makeRankTopologies();
+
+  // Set CVAR for vnode mode (needs ncclx EnvRAII, not SysEnvRAII)
+  // nolocal and vClique are passed via constructor params instead.
+  std::unique_ptr<SysEnvRAII> envTopo, envPpn;
+  if (mode == TopoMode::kVnode) {
+    envTopo =
+        std::make_unique<SysEnvRAII>("NCCL_COMM_STATE_DEBUG_TOPO", "vnode");
+    envPpn = std::make_unique<SysEnvRAII>(
+        "NCCL_COMM_STATE_DEBUG_TOPO_VNODE_NLOCALRANKS", "4");
   }
+
+  const bool noLocal = (mode == TopoMode::kNolocal);
+  auto commState = std::make_unique<CommStateX>(
+      0,
+      kNRanks,
+      kCudaDev,
+      kCudaArch,
+      kBusId,
+      kCommHash,
+      rankTopologies,
+      std::vector<int>{},
+      "" /* commDesc */,
+      noLocal,
+      vCliqueSize);
+
+  // Node grouping matches expected virtual topology
+  EXPECT_EQ(commState->nNodes(), expectedNNodes);
+  EXPECT_EQ(commState->nLocalRanks(0), expectedNLocalRanks);
+
+  // Real hostname and gPid preserved for all ranks
+  using Helper = ctran::testing::VerifyCommStateXHelper;
+  for (int r = 0; r < kNRanks; ++r) {
+    Helper::verifyHost(commState.get(), r, rankTopologies[r].host);
+    Helper::verifyGPid(
+        commState.get(), r, rankTopologies[r].host, rankTopologies[r].pid);
+  }
+  Helper::verifyGPidUniqueness(commState.get(), kNRanks);
 }
 
+INSTANTIATE_TEST_SUITE_P(
+    CommStateXTest,
+    GpidTopoTest,
+    ::testing::Values(
+        // system: 2 real hosts × 4 ranks each
+        TopoTestParam{TopoMode::kSystem, 0, 2, 4},
+        // nolocal: each rank is its own node
+        TopoTestParam{TopoMode::kNolocal, 0, 8, 1},
+        // vnode with nLocalRanks=4: 2 virtual nodes
+        TopoTestParam{TopoMode::kVnode, 0, 2, 4},
+        // vClique size=2: 4 virtual nodes of 2 ranks each
+        TopoTestParam{TopoMode::kVClique, 2, 4, 2}),
+    topoTestName);
+
 TEST(CommStateXTest, TopologySetInvalidNvlFabricTopos) {
   const int rank = 0;
   const int nRanks = 4;
@@ -663,9 +729,7 @@ TEST(CommStateXTest, nvlFabricWithNoLocal) {
       "" /* commDesc */,
       true /* noLocal */);
 
-  // noLocal is set at construction; initRankStatesTopology delegates to
-  // initRankTopologyNolocal
-  commState->initRankStatesTopology(nullptr);
+  commState->initRankTopologyNolocal();
 
   // Set up NVL fabric with 2 clusters of 4 ranks each (e.g. GB200 2-GPU trays)
   std::vector<NvlFabricTopology> nvlFabricTopologies{};

comms/ctran/tests/CtranDistTestUtils.cc

@@ -141,15 +141,9 @@ std::unique_ptr<CtranComm> CtranDistTestFixture::makeCtranComm() {
   std::unique_ptr<meta::comms::IBootstrap> commBootstrap(
       meta::comms::createBootstrap("ctrancomm"));
 
-  // Initialize topology
-  if (NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::nolocal) {
-    comm->statex_->initRankTopologyNolocal();
-  } else if (NCCL_COMM_STATE_DEBUG_TOPO == NCCL_COMM_STATE_DEBUG_TOPO::vnode) {
-    comm->statex_->initRankTopologyVnode(
-        NCCL_COMM_STATE_DEBUG_TOPO_VNODE_NLOCALRANKS);
-  } else {
-    comm->statex_->initRankStatesTopology(commBootstrap.get());
-  }
+  // Always use bootstrap to get real pids. Virtual topology overrides
+  // (nolocal, vnode, vClique) are applied inside setRankStatesTopologies.
+  comm->statex_->initRankStatesTopology(commBootstrap.get());
 
   comm->bootstrap_ = std::make_unique<ctran::testing::CtranTestBootstrap>(
       std::move(commBootstrap));