fix(rccl): gate and bound the destroy-path intranode barrier

projects/rccl/src/bootstrap.cc

@@ -440,13 +440,13 @@ struct extInfo {
 // drive the retry from here: close the rejected socket, yield 1ms, retry.
 // abortFlag is honored between iterations.
 static ncclResult_t bootstrapAccept(struct ncclSocket* sock, struct ncclSocket* listenSock,
-                                    volatile uint32_t* abortFlag) {
+                                    volatile uint32_t* abortFlag, int64_t deadlineNs = 0) {
   while (1) {
     if (abortFlag != NULL && __atomic_load_n(abortFlag, __ATOMIC_ACQUIRE) != 0) {
       return ncclInternalError;
     }
     NCCLCHECK(ncclSocketInit(sock));
-    NCCLCHECK(ncclSocketAccept(sock, listenSock, /*retry=*/false));
+    NCCLCHECK(ncclSocketAccept(sock, listenSock, /*retry=*/false, deadlineNs));
     if (sock->state != ncclSocketStateBadHandshake) return ncclSuccess;
     INFO(NCCL_INIT | NCCL_NET, "bootstrap: rejected bad-magic peer connection on listen sock, retrying accept");
     (void)ncclSocketClose(sock);
@@ -1745,7 +1745,8 @@ static void unexpectedFree(struct bootstrapState* state) {
 }
 
 // We can't know who we'll receive from, so we need to receive everything at once
-static ncclResult_t socketAccept(void* commState, int peer, int tag, struct ncclSocket* sock) {
+static ncclResult_t socketAccept(void* commState, int peer, int tag, struct ncclSocket* sock,
+                                 int64_t deadlineNs = 0) {
   ncclResult_t ret = ncclSuccess;
   struct bootstrapState* state = (struct bootstrapState*)commState;
 
@@ -1757,7 +1758,7 @@ static ncclResult_t socketAccept(void* commState, int peer, int tag, struct nccl
   // Then look for new connections
   while (1) {
     struct socketAckInfo ack = {0};
-    NCCLCHECKGOTO(bootstrapAccept(sock, &STATE_LISTEN(state, peerSocket), state->abortFlag), ret, fail);
+    NCCLCHECKGOTO(bootstrapAccept(sock, &STATE_LISTEN(state, peerSocket), state->abortFlag, deadlineNs), ret, fail);
     NCCLCHECKGOTO(socketRecv(sock, &ack, sizeof(struct socketAckInfo)), ret, fail);
     // Match: tag must match, and peer must match (peer < 0 means wildcard)
     if (ack.tag == tag && (peer < 0 || ack.rank == peer)) return ncclSuccess;
@@ -1771,7 +1772,8 @@ static ncclResult_t socketAccept(void* commState, int peer, int tag, struct nccl
 }
 
 // We can't know who we'll receive from, so we need to receive everything at once
-ncclResult_t bootstrapRecv(void* commState, int peer, int tag, void* data, int size) {
+// [RCCL] deadlineNs bounds only the wait for the peer to connect; see bootstrapIntraNodeBarrierTimeout().
+ncclResult_t bootstrapRecvDeadline(void* commState, int peer, int tag, void* data, int size, int64_t deadlineNs) {
   ncclResult_t ret;
   struct ncclSocket sock;
   struct bootstrapState* state = (struct bootstrapState*)commState;
@@ -1784,7 +1786,7 @@ ncclResult_t bootstrapRecv(void* commState, int peer, int tag, void* data, int s
     WARN("bootstrapRecv: an earlier asynchronous bootstrap send failed");
     return ret;
   }
-  NCCLCHECK(socketAccept(commState, peer, tag, &sock));
+  NCCLCHECK(socketAccept(commState, peer, tag, &sock, deadlineNs));
   TRACE(NCCL_BOOTSTRAP, "Receiving tag=%d peer=%d size=%d", tag, peer, size);
   NCCLCHECKGOTO(socketRecv(&sock, ((char*)data), size), ret, fail);
   NCCLCHECKGOTO(ncclSocketClose(&sock, /*wait*/ true), ret, fail);
@@ -1794,6 +1796,10 @@ ncclResult_t bootstrapRecv(void* commState, int peer, int tag, void* data, int s
   return ret;
 }
 
+ncclResult_t bootstrapRecv(void* commState, int peer, int tag, void* data, int size) {
+  return bootstrapRecvDeadline(commState, peer, tag, data, size, /*deadlineNs=*/0);
+}
+
 static ncclResult_t netRingAllGather(ncclNet_t* net, void* sendComm, void* recvComm, int rank, int nranks, char* data,
                                      int size, volatile uint32_t* abortFlag) {
   ncclResult_t res;
@@ -2034,7 +2040,8 @@ ncclResult_t bootstrapAllGather(void* commState, void* allData, int size) {
   return res;
 }
 
-static ncclResult_t bootstrapP2PBarrier(void* commState, int* ranks, int rank, int nranks, int tag) {
+static ncclResult_t bootstrapP2PBarrier(void* commState, int* ranks, int rank, int nranks, int tag,
+                                        int64_t deadlineNs = 0) {
   if (nranks == 1) return ncclSuccess;
   /* Simple [intra] process barrier
    *
@@ -2046,7 +2053,11 @@ static ncclResult_t bootstrapP2PBarrier(void* commState, int* ranks, int rank, i
     int src = (rank - mask + nranks) % nranks;
     int dst = (rank + mask) % nranks;
     NCCLCHECK(bootstrapSend(commState, ranks ? ranks[dst] : dst, tag, data, sizeof(data)));
-    NCCLCHECK(bootstrapRecv(commState, ranks ? ranks[src] : src, tag, data, sizeof(data)));
+    // [RCCL] One absolute deadline shared by every round rather than a fresh timeout per round, so rounds that
+    // waited cannot buy later rounds extra time. This bounds the wait for each peer to connect, not the whole
+    // call: bootstrapSend above keeps its own connect retry, so the wall-clock worst case is the deadline plus
+    // that retry, not the deadline alone.
+    NCCLCHECK(bootstrapRecvDeadline(commState, ranks ? ranks[src] : src, tag, data, sizeof(data), deadlineNs));
   }
   return ncclSuccess;
 }
@@ -2060,6 +2071,45 @@ ncclResult_t bootstrapIntraNodeBarrier(void* commState, int* ranks, int rank, in
   return ncclSuccess;
 }
 
+// [RCCL] bootstrapIntraNodeBarrier with a bound on how long a missing peer can hold everyone up.
+//
+// The dissemination barrier has no cancellation protocol, so without this a single rank that never arrives blocks
+// every peer that did arrive, forever (AICOMRCCL-2467 / AICOMRCCL-2468).
+//
+// The case this exists for is specifically a peer that is *alive but not participating* -- it took the
+// abortFlag != 0 branch in commDestroySync() and skipped the barrier, so it will never connect. A peer that
+// actually dies needs no timeout: the kernel closes its sockets on process exit, so the recv below sees a closed
+// socket and already returns an error. That asymmetry is also why the ack read in socketAccept() is left
+// unbounded -- reaching it means a peer completed the TCP connect, which a dead process cannot do.
+//
+// The cost is that an alive-but-not-participating peer is indistinguishable from a merely slow one. Expiry here
+// is a guess that the peer is gone, and if the guess is wrong the peer loses whatever it was establishing. So the
+// timeout wants to be comfortably longer than any legitimate skew between ranks reaching ncclCommDestroy.
+//
+// timeoutMsec <= 0 restores the unbounded behaviour. On expiry the caller gets ncclRemoteError and must decide
+// what to do; callers on a teardown path generally want to log and continue, since by then the alternative is a
+// hang. Note that a rank which gives up stops feeding its own dst, so its peers hit their own deadlines in turn:
+// the whole host drains within roughly one timeout rather than cascading.
+ncclResult_t bootstrapIntraNodeBarrierTimeout(void* commState, int* ranks, int rank, int nranks, int tag,
+                                              int64_t timeoutMsec) {
+  if (timeoutMsec <= 0) return bootstrapIntraNodeBarrier(commState, ranks, rank, nranks, tag);
+
+  uint64_t time = 0;
+  int64_t const deadlineNs = clockNano() + timeoutMsec * 1000000LL;
+  BOOTSTRAP_PROF_OPEN(time);
+  ncclResult_t ret = bootstrapP2PBarrier(commState, ranks, rank, nranks, tag, deadlineNs);
+  BOOTSTRAP_PROF_CLOSE(time);
+  if (ret != ncclSuccess) {
+    WARN("bootstrapIntraNodeBarrierTimeout: rank %d of %d gave up after %" PRId64
+         " ms; a host-local peer never arrived. If that peer was healthy rather than gone, raise "
+         "RCCL_DESTROY_BARRIER_TIMEOUT_MSEC (0 disables the deadline).",
+         rank, nranks, timeoutMsec);
+    return ret;
+  }
+  TRACE(NCCL_BOOTSTRAP | NCCL_PROFILE, "bootstrapIntraNodeBarrier done in %f sec", time / 1e9);
+  return ncclSuccess;
+}
+
 ncclResult_t bootstrapBarrier(void* commState, int rank, int nranks, int tag) {
   uint64_t time = 0;
   BOOTSTRAP_PROF_OPEN(time);

projects/rccl/src/graph/paths.cc

@@ -807,6 +807,9 @@ int ncclPxnDisable(struct ncclComm* comm) {
   // pxnDisable); without one, honour the env default and skip PXN.
   if (comm == NULL) return ncclParamPxnDisable();
   if (comm->pxnDisable > RCCL_VALUE_INVALID) return comm->pxnDisable;
+  // commDestroySync() consults this on the teardown path, where a comm whose init failed before the topology was
+  // built can still reach us with a live user handle. rcclSetPxn() reads comm->topo unconditionally.
+  if (comm->topo == NULL) return ncclParamPxnDisable();
   if (comm->ncclNetVer == 4) {
     INFO(NCCL_INIT, "PXN Disabled as plugin is v4");
     comm->pxnDisable = 1;

projects/rccl/src/include/bootstrap.h

@@ -29,9 +29,16 @@ ncclResult_t bootstrapSplit(uint64_t magic, struct ncclComm* comm, struct ncclCo
 ncclResult_t bootstrapAllGather(void* commState, void* allData, int size);
 ncclResult_t bootstrapSend(void* commState, int peer, int tag, void* data, int size);
 ncclResult_t bootstrapRecv(void* commState, int peer, int tag, void* data, int size);
+// [RCCL] As bootstrapRecv, but gives up with ncclRemoteError if no peer connects by deadlineNs (clockNano clock).
+// 0 waits indefinitely.
+ncclResult_t bootstrapRecvDeadline(void* commState, int peer, int tag, void* data, int size, int64_t deadlineNs);
 ncclResult_t bootstrapBarrier(void* commState, int rank, int nranks, int tag);
 ncclResult_t bootstrapBroadcast(void* commState, int rank, int nranks, int root, void* bcastData, int size);
 ncclResult_t bootstrapIntraNodeBarrier(void* commState, int* ranks, int rank, int nranks, int tag);
+// [RCCL] As bootstrapIntraNodeBarrier, but returns ncclRemoteError instead of blocking forever when a host-local
+// peer never arrives. timeoutMsec <= 0 restores the unbounded wait.
+ncclResult_t bootstrapIntraNodeBarrierTimeout(void* commState, int* ranks, int rank, int nranks, int tag,
+                                              int64_t timeoutMsec);
 ncclResult_t bootstrapIntraNodeAllGather(void* commState, int* ranks, int rank, int nranks, void* allData, int size);
 ncclResult_t bootstrapIntraNodeBroadcast(void* commState, int* ranks, int rank, int nranks, int root, void* bcastData,
                                          int size);

projects/rccl/src/include/socket.h

@@ -150,7 +150,11 @@ ncclResult_t ncclSocketConnect(struct ncclSocket* sock);
 ncclResult_t ncclSocketReady(struct ncclSocket* sock, int* running);
 // Accept an incoming connection from listenSock->socketDescriptor and keep the file descriptor in
 // sock->socketDescriptor, with the remote side IP/port in sock->addr.
-ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* ulistenSock, bool retry = true);
+// [RCCL] deadlineNs bounds the wait for a peer to show up, on the clockNano() clock. 0 waits indefinitely,
+// which is upstream's behaviour and what every caller but the destroy barrier wants. On expiry the socket is
+// left untouched and ncclRemoteError is returned, so the caller can retry or give up.
+ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* ulistenSock, bool retry = true,
+                              int64_t deadlineNs = 0);
 ncclResult_t ncclSocketGetFd(struct ncclSocket* sock, ncclSocketDescriptor* socketDescriptor);
 ncclResult_t ncclSocketSetFd(ncclSocketDescriptor socketDescriptor, struct ncclSocket* sock);
 

projects/rccl/src/init.cc

@@ -125,6 +125,14 @@ NCCL_PARAM(GroupCudaStream, "GROUP_CUDA_STREAM", NCCL_GROUP_CUDA_STREAM);
 NCCL_PARAM(CheckPointers, "CHECK_POINTERS", 0);
 NCCL_PARAM(CommBlocking, "COMM_BLOCKING", NCCL_CONFIG_UNDEF_INT);
 NCCL_PARAM(RuntimeConnect, "RUNTIME_CONNECT", 0);
+// [RCCL] Deadline for the host-local rendezvous in commDestroySync(). 0 restores upstream's unbounded wait.
+//
+// What this bounds is the skew between host-local ranks of one comm reaching ncclCommDestroy, which in a correct
+// program is short -- they have just finished the same collective. 120s is therefore orders of magnitude above the
+// expected value, which is what we want: expiry is a guess that a peer is gone, and the cost of guessing wrong is
+// borne by a healthy peer. It is also kept below PyTorch's 600s ProcessGroupNCCL watchdog default, since a
+// deadline longer than the framework's own would never be the thing that fires.
+RCCL_PARAM(DestroyBarrierTimeoutMsec, "DESTROY_BARRIER_TIMEOUT_MSEC", 120000);
 // When enabled (default), defer PAT QP creation until PAT is first selected by an AG/RS; NCCL_PAT_LAZY_INIT=0 restores eager connect at init.
 NCCL_PARAM(PatLazyInit, "PAT_LAZY_INIT", 1);
 // When enabled (default), PAT ReduceScatter and AllGather share one connection set.
@@ -4260,7 +4268,37 @@ static ncclResult_t commDestroySync(struct ncclAsyncJob* job_) {
              comm->commHash, comm->rank);
       }
     }
-    if (COMPILER_ATOMIC_LOAD(comm->abortFlag, std::memory_order_acquire) == 0) {
+    // RCCL: two changes to the upstream barrier below, both because it is a cross-process rendezvous that upstream
+    // gives no way out of -- a rank that does not arrive strands every peer that did, forever
+    // (AICOMRCCL-2467 / AICOMRCCL-2468).
+    //
+    // 1. Only run it when PXN is actually enabled for this comm.  The barrier exists solely to protect PXN
+    //    connection establishment, and RCCL (unlike upstream) defaults NCCL_PXN_DISABLE=1 and only auto-enables PXN
+    //    at nRanks >= 64 (gfx942) / >= 32 (gfx950), so on the overwhelming majority of comms it was deadlock risk
+    //    protecting nothing.  ncclPxnDisable() is derived from arch, nRanks and env and is therefore agreed by every
+    //    rank by construction, which is what matters here: a barrier some ranks enter and others skip is itself a
+    //    hang.  For the same reason we do not try to vary the membership -- e.g. dropping ranks an
+    //    NCCL_SHRINK_ABORT excluded.  Only the shrink survivors know about that exclusion; the excluded rank still
+    //    holds its own parent handle and would enter the full barrier, so filtering just moves the hang onto it.
+    //
+    // 2. When it does run, bound it.  A peer can fail to arrive for reasons no membership calculation can predict
+    //    -- most of all an ncclCommAbort elsewhere in the process, which makes that rank take the abortFlag branch
+    //    above and skip the barrier entirely -- so it gets a deadline and reports an error instead of blocking
+    //    forever.  Expiry is non-fatal: the result is logged and teardown continues.
+    //
+    //    What expiry costs is bounded.  Proceeding here means running ncclProxyStop() below while a peer might
+    //    still be establishing a PXN connection through us, which is the race the barrier exists to prevent, so it
+    //    is worth being precise about the consequence.  The proxy service thread force-closes its peer sockets and
+    //    its listen socket on stop (proxy.cc:2090-2095, 2108), so the peer gets a closed socket or ECONNREFUSED
+    //    and fails cleanly -- it does not hang, and it does not touch freed state.  In particular nothing shared is
+    //    unmapped out from under it: the ncclShmClose() in ncclProxyStop() releases only this rank's client-side
+    //    view, because a client attach leaves handle->shmPath NULL and ncclOsShmClose() unlinks only when that is
+    //    non-NULL (os/linux.cc:777, 892).  Teardown is also refCount-gated (proxy.cc:2284), so comms that share
+    //    this proxy state are untouched.
+    //
+    //    The residual risk is a false positive: a peer that was merely slow to reach destroy, rather than gone,
+    //    loses whatever it was establishing.  That is why the default is generous rather than tight.
+    if (COMPILER_ATOMIC_LOAD(comm->abortFlag, std::memory_order_acquire) == 0 && ncclPxnDisable(comm) != 1) {
       int* hostRanks;
       int hostRank = 0;
       int nHostRanks = 0;
@@ -4276,7 +4314,8 @@ static ncclResult_t commDestroySync(struct ncclAsyncJob* job_) {
         }
       }
       ncclResult_t barrierRet = ncclSuccess;
-      NCCLCHECKIGNORE(bootstrapIntraNodeBarrier(comm->bootstrap, hostRanks, hostRank, nHostRanks, hostRanks[0]),
+      NCCLCHECKIGNORE(bootstrapIntraNodeBarrierTimeout(comm->bootstrap, hostRanks, hostRank, nHostRanks, hostRanks[0],
+                                                       rcclParamDestroyBarrierTimeoutMsec()),
                       barrierRet);
       if (barrierRet != ncclSuccess) {
         INFO(NCCL_DESTROY, "comm 0x%" PRIx64 " rank %d intranode barrier error %d", comm->commHash, comm->rank,

projects/rccl/src/misc/socket.cc

@@ -484,8 +484,9 @@ ncclResult_t ncclSocketConnect(struct ncclSocket* sock) {
   }
 }
 
-ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* listenSock, bool retry) {
+ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* listenSock, bool retry, int64_t deadlineNs) {
   ncclResult_t ret = ncclSuccess;
+  bool timedOut = false;
 
   if (listenSock == NULL || sock == NULL) {
     WARN("ncclSocketAccept: pass NULL socket");
@@ -515,12 +516,24 @@ ncclResult_t ncclSocketAccept(struct ncclSocket* sock, struct ncclSocket* listen
       sock->state = ncclSocketStateAccepting;
       if (!retry) break;
     }
+    // [RCCL] Only a caller that passed a deadline can time out; everyone else keeps upstream's unbounded spin.
+    if (deadlineNs != 0 && clockNano() > deadlineNs) {
+      timedOut = true;
+      break;
+    }
   } while (sock->asyncFlag == 0 &&
            (sock->abortFlag == NULL || COMPILER_ATOMIC_LOAD(sock->abortFlag, std::memory_order_acquire) == 0) &&
            (sock->state == ncclSocketStateAccepting || sock->state == ncclSocketStateAccepted));
 
   if (sock->abortFlag && COMPILER_ATOMIC_LOAD(sock->abortFlag, std::memory_order_acquire)) return ncclInternalError;
 
+  // [RCCL] Must precede the switch: it maps a still-Accepting socket to ncclSuccess for the asyncFlag callers,
+  // so falling through would report a peer that never arrived as a completed accept.
+  if (timedOut) {
+    ret = ncclRemoteError;
+    goto exit;
+  }
+
   switch (sock->state) {
   case ncclSocketStateAccepting:
   case ncclSocketStateAccepted: