NCCL 2.26.2-1

Profiler improvements
 * Add events for CUDA kernel start and end.
 * Allow network plugins to generate profiling events
 * Enable profiling on a per-operation basis, rather than per-communicator.
 * Add support for graph capturing.

Add implicit launch order
 * Allow to prevent deadlocks when using multiple NCCL communicators per
   device by implicitly ordering NCCL operations using the host program
   order. Disabled by default, set NCCL_LAUNCH_ORDER_IMPLICIT=1 to enable.
 * Add a complementary mechanism to detect host threads racing to launch
   to the same device. Enabled by default, set NCCL_LAUNCH_RACE_FATAL=0 to
   disable.

Optimize the PAT algorithm
 * Separate the computation and execution of PAT steps on different warps,
   allowing to run up to 16 PAT steps in parallel to significantly
   accelerate PAT and reduce its linear part.

Add support for setting QoS per communicator
 * Add a new trafficClass field to the communicator configuration, to
   allow the application to select a particular traffic class for a
   given communicator. The meaning of the traffic class is
   network-specific and should be set in accordance with the network
   configuration.
 * For the IB/RoCE plugin, existing config variables such as NCCL_IB_SL
   and NCCL_IB_TC take precedence.

Allow to enable GPU Direct RDMA specifically on C2C platforms
 * Disabled by default, set NCCL_NET_GDR_C2C=1 to enable.

Do not disable user buffer registration unless PXN is really used
 * Only disable UB when a communicator has more than one rank per
   node on any node.

RAS subsystem improvements
 * Report operation counts separately for each collective operation type.
 * Provide details about missing communicator ranks and reliably
   distinguish ranks that are no longer a given communicator's members
   (now reported as NOCOMM) from those that failed to respond.

Add support for timestamps to NCCL diagnostic messages
 * On by default for WARN messages; NCCL_DEBUG_TIMESTAMP_LEVELS can be
   used to enable them for other debug levels as well.
 * The format can be changed using the NCCL_DEBUG_TIMESTAMP_FORMAT config
   variable.

Reduce the memory usage with NVLink SHARP (NVLS)
 * Potentially save hundreds of MBs of device memory, considering the
   multicast buffer size granularity separately from the address alignment.

Update performance tuning for recent Intel CPUs
 * Improve algorithm/protocol selection on recent CPUs such as Emerald
   Rapids and Sapphire Rapids.

Improve channel scheduling when mixing LL and Simple operations.
 * Make LL operations account for 4x more traffic to ensure LL and simple
   operations complete at the same time.

Refactor the plugin code
 * Clean up and harmonize the support code across the network, tuner,
   and profiler plugins.

Add support for comment lines (starting with #) in the nccl.conf file
* Issue #1540.

Make user buffer registration problems print an INFO instead of a WARN.

Drop support for network plugin interface version 5.

Fix a race condition with split-shared communicators
 * NCCL could hang during connection setup if multiple communicators
   were grouped together that share resources.

Fix a performance regression when using NCCL_CROSS_NIC=1
 * NCCL would unnecessarily alternate rings, breaking the GPU-NIC
   associations.

Make GID index detection code more resilient
 * Dynamic GID detection code was giving up too soon if the
   detected index was not available (e.g., wasn't mapped to the
   container's sysfs).
 * Issues #1538, #1573.

Fix a race condition with non-blocking operation
 * Fix issue when creating a non-blocking communicator after a non-
   blocking collective operation on another communicator.

Fix shared memory usage on recent Blackwell GPUs.
 * Issues NVIDIA/nccl-tests#287, NVIDIA/nccl-tests#291, #1637.

Fix an error with NIC fusion and IB SHARP when recreating communicators
 * Disable the unloading of network plugins

Make the auto-merge failures in the NIC fusion non-fatal
 * This could happen when trying to merge IB and RoCE devices.

Fixes to ncclCommAbort
 * Fix hangs due to the progress thread spinning indefinitely on the
   network progress.
 * Reduce the abort time by up to two orders of magnitude.

Fix a crash when libnccl.so was dynamically unloaded
 * The RAS subsystem was missing a clean-up handler.

Fix a hang if the network plugin's test() call returns an error.

Fix a hang on heterogeneous architectures
 * Ensure we harmonize the tuning to avoid different tuning choices,
   causing a hang.

Fix double-free on failed ncclCommInitRank and ncclCommFinalize.

Fix a potential list traversal bug during a group launch of multiple
communicators
 * Issue #1599.

Unify the handling of NCCL configuration variables
 * Under rare circumstances, some variables specified in the config file
   could be ignored.

src/ras/collectives.cc

@@ -4,7 +4,7 @@
  * See LICENSE.txt for license information
  ************************************************************************/
 
-#define NDEBUG // Comment out duriyng development only!
+#define NDEBUG // Comment out during development only!
 #include <cassert>
 #include <mutex>
 
@@ -12,6 +12,7 @@
 #include "checks.h"
 #include "comm.h"
 #include "nccl.h"
+#include "transport.h"
 #include "utils.h"
 #include "ras_internal.h"
 
@@ -32,14 +33,14 @@ static int nRasCollHistory, rasCollHistNextIdx;
 // Monotonically increased to ensure that each collective originating locally has a unique Id.
 static uint64_t rasCollLastId;
 
-// Array keeping track of ongoing collective operations (apart from broadcasts, which have no response so require
+// Keeping track of ongoing collective operations (apart from broadcasts, which have no response so require
 // no such tracking).
-struct rasCollective* rasCollectives;
-static int nRasCollectives;
+struct rasCollective* rasCollectivesHead;
+struct rasCollective* rasCollectivesTail;
 
 static ncclResult_t getNewCollEntry(struct rasCollective** pColl);
 static ncclResult_t rasLinkSendCollReq(struct rasLink* link, struct rasCollective* coll,
-                                       const struct rasCollRequest* req, size_t reqLen, int fromConnIdx);
+                                       const struct rasCollRequest* req, size_t reqLen, struct rasConnection* fromConn);
 static ncclResult_t rasConnSendCollReq(struct rasConnection* conn, const struct rasCollRequest* req, size_t reqLen);
 static ncclResult_t rasCollReadyResp(struct rasCollective* coll);
 static ncclResult_t rasConnSendCollResp(struct rasConnection* conn,
@@ -47,12 +48,17 @@ static ncclResult_t rasConnSendCollResp(struct rasConnection* conn,
                                         const union ncclSocketAddress* peers, int nPeers,
                                         const char* data, int nData, int nLegTimeouts);
 
-static ncclResult_t rasCollConnsInit(char** pData, int* pNData);
+static ncclResult_t rasCollConnsInit(struct rasCollRequest** pReq, size_t* pReqLen, char** pData, int* pNData);
 static ncclResult_t rasCollConnsMerge(struct rasCollective* coll, struct rasMsg* msg);
 
-static ncclResult_t rasCollCommsInit(char** pData, int* pNData);
+static ncclResult_t rasCollCommsInit(struct rasCollRequest** pReq, size_t* pReqLen, char** pData, int* pNData);
 static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg* msg);
+static bool rasCollCommsSkipMissing(const struct rasCollRequest* req, struct ncclComm* comm);
 static int ncclCommsCompare(const void* p1, const void* p2);
+static int peersHashesCompare(const void* p1, const void* p2);
+static int peersHashesSearch(const void* k, const void* e);
+static int rasCommIdCompare(const void* p1, const void* p2);
+static int rasCollCommsMissingRankSearch(const void* k, const void* e);
 
 
 ///////////////////////////////////////////////////////////////////////////////////////
@@ -62,22 +68,26 @@ static int ncclCommsCompare(const void* p1, const void* p2);
 // Returns the index of the first available entry in the rasCollectives array, enlarging the array if necessary.
 static ncclResult_t getNewCollEntry(struct rasCollective** pColl) {
   struct rasCollective* coll;
-  int i;
-  for (i = 0; i < nRasCollectives; i++)
-    if (rasCollectives[i].type == RAS_MSG_NONE)
-      break;
-  if (i == nRasCollectives) {
-    NCCLCHECK(ncclRealloc(&rasCollectives, nRasCollectives, nRasCollectives+RAS_INCREMENT));
-    nRasCollectives += RAS_INCREMENT;
-  }
+  int nRasConns;
+
+  NCCLCHECK(ncclCalloc(&coll, 1));
 
-  coll = rasCollectives+i;
-  memset(coll, '\0', sizeof(*coll));
   coll->startTime = clockNano();
-  coll->fromConnIdx = -1;
+  coll->fromConn = nullptr;
   // We are unlikely to use the whole array, but at least we won't need to realloc.
+  nRasConns = 0;
+  for (struct rasConnection* conn = rasConnsHead; conn; conn = conn->next)
+    nRasConns++;
   NCCLCHECK(ncclCalloc(&coll->fwdConns, nRasConns));
 
+  if (rasCollectivesHead) {
+    rasCollectivesTail->next = coll;
+    coll->prev = rasCollectivesTail;
+    rasCollectivesTail = coll;
+  } else {
+    rasCollectivesHead = rasCollectivesTail = coll;
+  }
+
   *pColl = coll;
   return ncclSuccess;
 }
@@ -95,21 +105,23 @@ void rasCollReqInit(struct rasCollRequest* req) {
 // in preparation for collective response messages.
 // pAllDone indicates on return if the collective operation is already finished, which is unusual, but possible
 // in scenarios such as a total of two peers.
-// pCollIdx provides on return an index of the allocated rasCollective structure to track this collective (unless
+// pColl provides on return a pointer to the allocated rasCollective structure to track this collective (unless
 // it's a broadcast, which require no such tracking).
-ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, size_t reqLen, bool* pAllDone, int* pCollIdx,
-                               int fromConnIdx) {
+ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, bool* pAllDone,
+                               struct rasCollective** pColl, struct rasConnection* fromConn) {
   struct rasCollective* coll = nullptr;
+  struct rasCollRequest* reqMod = (struct rasCollRequest*)req;
+  size_t reqLen = 0;
   if (req->type >= RAS_COLL_CONNS) {
     // Keep track of this collective operation so that we can handle the responses appropriately.
     NCCLCHECK(getNewCollEntry(&coll));
-    if (pCollIdx)
-      *pCollIdx = coll-rasCollectives;
+    if (pColl)
+      *pColl = coll;
     memcpy(&coll->rootAddr, &req->rootAddr, sizeof(coll->rootAddr));
     coll->rootId = req->rootId;
     coll->type = req->type;
     coll->timeout = req->timeout;
-    coll->fromConnIdx = fromConnIdx;
+    coll->fromConn = fromConn;
     if (ncclCalloc(&coll->peers, 1) == ncclSuccess) {
       memcpy(coll->peers, &rasNetListeningSocket.addr, sizeof(*coll->peers));
       coll->nPeers = 1;
@@ -117,9 +129,9 @@ ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, size_t reqLen,
 
     // Collective-specific initialization of accumulated data (using local data for now).
     if (req->type == RAS_COLL_CONNS)
-      (void)rasCollConnsInit(&coll->data, &coll->nData);
+      (void)rasCollConnsInit(&reqMod, &reqLen, &coll->data, &coll->nData);
     else if (req->type == RAS_COLL_COMMS)
-      (void)rasCollCommsInit(&coll->data, &coll->nData);
+      (void)rasCollCommsInit(&reqMod, &reqLen, &coll->data, &coll->nData);
   } else { // req->type < RAS_COLL_CONNS
     // Add the info to the collective message history.
     nRasCollHistory = std::min(nRasCollHistory+1, COLL_HISTORY_SIZE);
@@ -131,42 +143,42 @@ ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, size_t reqLen,
     // Collective-specific message handling.
     if (req->type == RAS_BC_DEADPEER) {
       bool done = false;
-      rasMsgHandleBCDeadPeer(req, &done);
+      rasMsgHandleBCDeadPeer(&reqMod, &reqLen, &done);
       if (done)
         goto exit;
     }
   } // req->type < RAS_COLL_CONNS
 
-  for (int connIdx = 0; connIdx < nRasConns; connIdx++)
-    rasConns[connIdx].linkFlag = false;
+  for (struct rasConnection* conn = rasConnsHead; conn; conn = conn->next)
+    conn->linkFlag = false;
 
-  (void)rasLinkSendCollReq(&rasNextLink, coll, req, reqLen, fromConnIdx);
-  (void)rasLinkSendCollReq(&rasPrevLink, coll, req, reqLen, fromConnIdx);
+  (void)rasLinkSendCollReq(&rasNextLink, coll, reqMod, reqLen, fromConn);
+  (void)rasLinkSendCollReq(&rasPrevLink, coll, reqMod, reqLen, fromConn);
 
   if (coll && pAllDone)
     *pAllDone = (coll->nFwdSent == coll->nFwdRecv);
 exit:
+  if (reqMod != req)
+    free(reqMod);
   return ncclSuccess;
 }
 
 // Sends the collective message through all connections associated with this link (with the exception of the one
 // the message came from, if any).
 static ncclResult_t rasLinkSendCollReq(struct rasLink* link, struct rasCollective* coll,
-                                       const struct rasCollRequest* req, size_t reqLen, int fromConnIdx) {
-  for (int i = 0; i < link->nConns; i++) {
-    struct rasLinkConn* linkConn = link->conns+i;
-    if (linkConn->connIdx != -1 && linkConn->connIdx != fromConnIdx) {
-      struct rasConnection* conn = rasConns+linkConn->connIdx;
-      if (!conn->linkFlag) {
-        // We send collective messages through fully established and operational connections only.
-        if (conn->sockIdx != -1 && rasSockets[conn->sockIdx].status == RAS_SOCK_READY && !conn->experiencingDelays) {
-          if (rasConnSendCollReq(conn, req, reqLen) == ncclSuccess && coll != nullptr)
-            coll->fwdConns[coll->nFwdSent++] = linkConn->connIdx;
-        } // if (conn->sockIdx != -1 && RAS_SOCK_READY)
-        conn->linkFlag = true;
-      } // if (!conn->linkFlag)
-    } // if (linkConn->connIdx != -1 && linkConn->connIdx != fromConnIdx)
-  } // for (i)
+                                       const struct rasCollRequest* req, size_t reqLen,
+                                       struct rasConnection* fromConn) {
+  for (struct rasLinkConn* linkConn = link->conns; linkConn; linkConn = linkConn->next) {
+    if (linkConn->conn && linkConn->conn != fromConn && !linkConn->conn->linkFlag) {
+      // We send collective messages through fully established and operational connections only.
+      if (linkConn->conn->sock && linkConn->conn->sock->status == RAS_SOCK_READY &&
+          !linkConn->conn->experiencingDelays) {
+        if (rasConnSendCollReq(linkConn->conn, req, reqLen) == ncclSuccess && coll != nullptr)
+          coll->fwdConns[coll->nFwdSent++] = linkConn->conn;
+      } // linkConn->conn is fully established and operational.
+      linkConn->conn->linkFlag = true;
+    } // if (linkConn->conn && linkConn->conn != fromConn && !linkConn->con->linkFlag)
+  } // for (linkConn)
 
   return ncclSuccess;
 }
@@ -190,8 +202,8 @@ static ncclResult_t rasConnSendCollReq(struct rasConnection* conn, const struct
 // in which case it can immediately send the response.
 ncclResult_t rasMsgHandleCollReq(struct rasMsg* msg, struct rasSocket* sock) {
   bool allDone = false;
-  int collIdx = -1;
-  assert(sock->connIdx != -1);
+  struct rasCollective* coll = nullptr;
+  assert(sock->conn);
 
   // First check if we've already handled this request (through another connection).
   for (int i = 0; i < nRasCollHistory; i++) {
@@ -202,7 +214,7 @@ ncclResult_t rasMsgHandleCollReq(struct rasMsg* msg, struct rasSocket* sock) {
       if (msg->collReq.type >= RAS_COLL_CONNS) {
         // Send an empty response so that the sender can account for it.  The non-empty response has already been
         // sent through the connection that we received the request through first.
-        NCCLCHECK(rasConnSendCollResp(rasConns+sock->connIdx, &msg->collReq.rootAddr, msg->collReq.rootId,
+        NCCLCHECK(rasConnSendCollResp(sock->conn, &msg->collReq.rootAddr, msg->collReq.rootId,
                                       /*peers*/nullptr, /*nPeers*/0, /*data*/nullptr, /*nData*/0, /*nLegTimeouts*/0));
       }
       goto exit;
@@ -211,31 +223,29 @@ ncclResult_t rasMsgHandleCollReq(struct rasMsg* msg, struct rasSocket* sock) {
 
   if (msg->collReq.type >= RAS_COLL_CONNS) {
     // Check if we're currently handling this collective request.
-    for (int i = 0; i < nRasCollectives; i++) {
-      struct rasCollective* coll = rasCollectives+i;
-      if (coll->type != RAS_MSG_NONE &&
-          memcmp(&msg->collReq.rootAddr, &coll->rootAddr, sizeof(msg->collReq.rootAddr)) == 0 &&
+    for (coll = rasCollectivesHead; coll; coll = coll->next) {
+      if (memcmp(&msg->collReq.rootAddr, &coll->rootAddr, sizeof(msg->collReq.rootAddr)) == 0 &&
           msg->collReq.rootId == coll->rootId) {
         assert(msg->collReq.type == coll->type);
 
         // Send an empty response so that the sender can account for it.  The non-empty response will be
         // sent through the connection that we received the request through first.
-        NCCLCHECK(rasConnSendCollResp(rasConns+sock->connIdx, &msg->collReq.rootAddr, msg->collReq.rootId,
+        NCCLCHECK(rasConnSendCollResp(sock->conn, &msg->collReq.rootAddr, msg->collReq.rootId,
                                       /*peers*/nullptr, /*nPeers*/0, /*data*/nullptr, /*nData*/0, /*nLegTimeouts*/0));
         goto exit;
       } // if match
-    } // for (i)
+    } // for (coll)
   } // if (msg->collReq.type >= RAS_COLL_CONNS)
 
   // Re-broadcast the message to my peers (minus the one it came from) and handle it locally.
-  NCCLCHECK(rasNetSendCollReq(&msg->collReq, rasCollDataLength(msg->collReq.type), &allDone, &collIdx, sock->connIdx));
+  NCCLCHECK(rasNetSendCollReq(&msg->collReq, &allDone, &coll, sock->conn));
 
   if (msg->collReq.type >= RAS_COLL_CONNS && allDone) {
-    assert(collIdx != -1);
+    assert(coll);
     // We are a leaf process -- send the response right away.  This can probably trigger only for the case of a total
     // of two peers, and hence just one RAS connection, or during communication issues, because normally every peer
     // has more than one connection so there should always be _some_ other peer to forward the request to.
-    NCCLCHECK(rasCollReadyResp(rasCollectives+collIdx));
+    NCCLCHECK(rasCollReadyResp(coll));
   }
 exit:
   return ncclSuccess;
@@ -245,9 +255,9 @@ exit:
 // Invoked when we are finished waiting for the collective responses from other peers (i.e., either there weren't
 // any peers (unlikely), the peers sent their responses (likely), or we timed out.
 static ncclResult_t rasCollReadyResp(struct rasCollective* coll) {
-  if (coll->fromConnIdx != -1) {
+  if (coll->fromConn) {
     // For remotely-initiated collectives, send the response back.
-    NCCLCHECK(rasConnSendCollResp(rasConns+coll->fromConnIdx, &coll->rootAddr, coll->rootId,
+    NCCLCHECK(rasConnSendCollResp(coll->fromConn, &coll->rootAddr, coll->rootId,
                                   coll->peers, coll->nPeers, coll->data, coll->nData, coll->nLegTimeouts));
 
     // Add the identifying info to the collective message history.
@@ -302,18 +312,15 @@ static ncclResult_t rasConnSendCollResp(struct rasConnection* conn,
 // the data from the response into the accumulated data.  If all the responses have been accounted for, sends the
 // accumulated response back.
 ncclResult_t rasMsgHandleCollResp(struct rasMsg* msg, struct rasSocket* sock) {
-  int collIdx;
-  struct rasCollective* coll = nullptr;
+  struct rasCollective* coll;
   char line[SOCKET_NAME_MAXLEN+1];
 
-  for (collIdx = 0; collIdx < nRasCollectives; collIdx++) {
-    coll = rasCollectives+collIdx;
-    if (coll->type != RAS_MSG_NONE &&
-        memcmp(&msg->collResp.rootAddr, &coll->rootAddr, sizeof(msg->collResp.rootAddr)) == 0 &&
+  for (coll = rasCollectivesHead; coll; coll = coll->next) {
+    if (memcmp(&msg->collResp.rootAddr, &coll->rootAddr, sizeof(msg->collResp.rootAddr)) == 0 &&
         msg->collResp.rootId == coll->rootId)
       break;
   }
-  if (collIdx == nRasCollectives) {
+  if (coll == nullptr) {
     INFO(NCCL_RAS, "RAS failed to find a matching ongoing collective for response %s:%ld from %s!",
          ncclSocketToString(&msg->collResp.rootAddr, line), msg->collResp.rootId,
          ncclSocketToString(&sock->sock.addr, rasLine));
@@ -321,11 +328,11 @@ ncclResult_t rasMsgHandleCollResp(struct rasMsg* msg, struct rasSocket* sock) {
   }
 
   coll->nLegTimeouts += msg->collResp.nLegTimeouts;
-  assert(sock->connIdx != -1);
-  // Account for the received response in our collective operation tracking.
+  assert(sock->conn);
+  // Account for the received response in our collective operations tracking.
   for (int i = 0; i < coll->nFwdSent; i++) {
-    if (coll->fwdConns[i] == sock->connIdx) {
-      coll->fwdConns[i] = -1;
+    if (coll->fwdConns[i] == sock->conn) {
+      coll->fwdConns[i] = nullptr;
       break;
     }
   }
@@ -353,46 +360,53 @@ exit:
 
 // Removes a connection from all ongoing collectives.  Called when a connection is experiencing a delay or is being
 // terminated.
-void rasCollsPurgeConn(int connIdx) {
-  for (int i = 0; i < nRasCollectives; i++) {
-    struct rasCollective* coll = rasCollectives+i;
-    if (coll->type != RAS_MSG_NONE) {
-      char line[SOCKET_NAME_MAXLEN+1];
-      if (coll->fromConnIdx == connIdx) {
-        INFO(NCCL_RAS, "RAS purging collective %s:%ld because it comes from %s",
-             ncclSocketToString(&coll->rootAddr, line), coll->rootId,
-             ncclSocketToString(&rasConns[connIdx].addr, rasLine));
-        rasCollFree(coll);
-      } else {
-        for (int j = 0; j < coll->nFwdSent; j++) {
-          if (coll->fwdConns[j] == connIdx) {
-            coll->fwdConns[j] = -1;
-            coll->nFwdRecv++;
-            coll->nLegTimeouts++;
-            INFO(NCCL_RAS, "RAS not waiting for response from %s to collective %s:%ld "
-                 "(nFwdSent %d, nFwdRecv %d, nLegTimeouts %d)",
-                 ncclSocketToString(&rasConns[connIdx].addr, rasLine), ncclSocketToString(&coll->rootAddr, line),
-                 coll->rootId, coll->nFwdSent, coll->nFwdRecv, coll->nLegTimeouts);
-            if (coll->nFwdSent == coll->nFwdRecv)
-              (void)rasCollReadyResp(coll);
-            break;
-          }
-        } // for (j)
-      } // coll->fromConnIdx != connIdx
-    } // !RAS_MSG_NONE
-  } // for (i)
+void rasCollsPurgeConn(struct rasConnection* conn) {
+  for (struct rasCollective* coll = rasCollectivesHead; coll;) {
+    struct rasCollective* collNext = coll->next;
+    char line[SOCKET_NAME_MAXLEN+1];
+    if (coll->fromConn == conn) {
+      INFO(NCCL_RAS, "RAS purging collective %s:%ld because it comes from %s",
+           ncclSocketToString(&coll->rootAddr, line), coll->rootId,
+           ncclSocketToString(&conn->addr, rasLine));
+      rasCollFree(coll);
+    } else {
+      for (int i = 0; i < coll->nFwdSent; i++) {
+        if (coll->fwdConns[i] == conn) {
+          coll->fwdConns[i] = nullptr;
+          coll->nFwdRecv++;
+          coll->nLegTimeouts++;
+          INFO(NCCL_RAS, "RAS not waiting for response from %s to collective %s:%ld "
+               "(nFwdSent %d, nFwdRecv %d, nLegTimeouts %d)",
+               ncclSocketToString(&conn->addr, rasLine), ncclSocketToString(&coll->rootAddr, line), coll->rootId,
+               coll->nFwdSent, coll->nFwdRecv, coll->nLegTimeouts);
+          if (coll->nFwdSent == coll->nFwdRecv)
+            (void)rasCollReadyResp(coll);
+          break;
+        }
+      } // for (i)
+    } // coll->fromConn != conn
+    coll = collNext;
+  } // for (coll)
 }
 
 // Frees a rasCollective entry and any memory associated with it.
 void rasCollFree(struct rasCollective* coll) {
+  if (coll == nullptr)
+    return;
+
   free(coll->fwdConns);
-  coll->fwdConns = nullptr;
   free(coll->peers);
-  coll->peers = nullptr;
   free(coll->data);
-  coll->data = nullptr;
-  coll->fromConnIdx = -1;
-  coll->type = RAS_MSG_NONE;
+
+  if (coll == rasCollectivesHead)
+    rasCollectivesHead = rasCollectivesHead->next;
+  if (coll == rasCollectivesTail)
+    rasCollectivesTail = rasCollectivesTail->prev;
+  if (coll->prev)
+    coll->prev->next = coll->next;
+  if (coll->next)
+    coll->next->prev = coll->prev;
+  free(coll);
 }
 
 // Invoked from the main RAS thread loop to handle timeouts of the collectives.
@@ -407,64 +421,64 @@ void rasCollFree(struct rasCollective* coll) {
 // and send back whatever we have.  Unfortunately, the peer that the RAS client is connected to will in all likelihood
 // time out first, so at that point any delayed responses that eventually arrive are likely to be too late...
 void rasCollsHandleTimeouts(int64_t now, int64_t* nextWakeup) {
-  for (int collIdx = 0; collIdx < nRasCollectives; collIdx++) {
-    struct rasCollective* coll = rasCollectives+collIdx;
-    if (coll->type == RAS_MSG_NONE || coll->timeout == 0)
-      continue;
-
-    if (now - coll->startTime > coll->timeout) {
-      // We've exceeded the leg timeout.  For all outstanding responses, check their connections.
-      if (!coll->timeoutWarned) {
-        INFO(NCCL_RAS, "RAS collective %s:%ld timeout warning (%lds) -- %d responses missing",
-             ncclSocketToString(&coll->rootAddr, rasLine), coll->rootId,
-             (now - coll->startTime) / CLOCK_UNITS_PER_SEC, coll->nFwdSent - coll->nFwdRecv);
-        coll->timeoutWarned = true;
-      }
-      for (int i = 0; i < coll->nFwdSent; i++) {
-        if (coll->fwdConns[i] != -1) {
-          struct rasConnection* conn = rasConns+coll->fwdConns[i];
-          char line[SOCKET_NAME_MAXLEN+1];
-          if (!conn->experiencingDelays && conn->sockIdx != -1) {
-            struct rasSocket* sock = rasSockets+conn->sockIdx;
-            // Ensure that the connection is fully established and operational, and that the socket hasn't been
-            // re-created during the handling of the collective (which would suggest that the request may have been
-            // lost).
-            if (sock->status == RAS_SOCK_READY && sock->createTime < coll->startTime)
-              continue;
-          }
-          // In all other cases we declare a timeout so that we can (hopefully) recover.
-          INFO(NCCL_RAS, "RAS not waiting for response from %s to collective %s:%ld "
-               "(nFwdSent %d, nFwdRecv %d, nLegTimeouts %d)",
-               ncclSocketToString(&conn->addr, rasLine), ncclSocketToString(&coll->rootAddr, line),
-               coll->rootId, coll->nFwdSent, coll->nFwdRecv, coll->nLegTimeouts);
-          coll->fwdConns[i] = -1;
-          coll->nFwdRecv++;
-          coll->nLegTimeouts++;
-        } // if (coll->fwdConns[i] != -1)
-      } // for (i)
-      if (coll->nFwdSent == coll->nFwdRecv) {
-        (void)rasCollReadyResp(coll);
-      } else {
-        // At least some of the delays are *not* due to this process' connections experiencing delays, i.e., they
-        // must be due to delays at other processes.  Presumably those processes will give up waiting soon and the
-        // (incomplete) responses will arrive shortly, so we should wait a little longer.
-        if (now - coll->startTime > coll->timeout + RAS_COLLECTIVE_EXTRA_TIMEOUT) {
-          // We've exceeded even the longer timeout, which is unexpected.  Try to return whatever we have (though
-          // the originator of the collective, if it's not us, may have timed out already anyway).
-          INFO(NCCL_RAS, "RAS collective %s:%ld timeout error (%lds) -- giving up on %d missing responses",
+  for (struct rasCollective* coll = rasCollectivesHead; coll;) {
+    struct rasCollective* collNext = coll->next;
+    if (coll->timeout > 0) {
+      if (now - coll->startTime > coll->timeout) {
+        // We've exceeded the leg timeout.  For all outstanding responses, check their connections.
+        if (!coll->timeoutWarned) {
+          INFO(NCCL_RAS, "RAS collective %s:%ld timeout warning (%lds) -- %d responses missing",
                ncclSocketToString(&coll->rootAddr, rasLine), coll->rootId,
                (now - coll->startTime) / CLOCK_UNITS_PER_SEC, coll->nFwdSent - coll->nFwdRecv);
-          coll->nLegTimeouts += coll->nFwdSent - coll->nFwdRecv;
-          coll->nFwdRecv = coll->nFwdSent;
+          coll->timeoutWarned = true;
+        }
+        for (int i = 0; i < coll->nFwdSent; i++) {
+          if (coll->fwdConns[i]) {
+            struct rasConnection* conn = coll->fwdConns[i];
+            char line[SOCKET_NAME_MAXLEN+1];
+            if (!conn->experiencingDelays && conn->sock) {
+              // Ensure that the connection is fully established and operational, and that the socket hasn't been
+              // re-created during the handling of the collective (which would suggest that the request may have been
+              // lost).
+              if (conn->sock->status == RAS_SOCK_READY && conn->sock->createTime < coll->startTime)
+                continue;
+            }
+            // In all other cases we declare a timeout so that we can (hopefully) recover.
+            INFO(NCCL_RAS, "RAS not waiting for response from %s to collective %s:%ld "
+                 "(nFwdSent %d, nFwdRecv %d, nLegTimeouts %d)",
+                 ncclSocketToString(&conn->addr, rasLine), ncclSocketToString(&coll->rootAddr, line),
+                 coll->rootId, coll->nFwdSent, coll->nFwdRecv, coll->nLegTimeouts);
+            coll->fwdConns[i] = nullptr;
+            coll->nFwdRecv++;
+            coll->nLegTimeouts++;
+          } // if (coll->fwdConns[i])
+        } // for (i)
+        if (coll->nFwdSent == coll->nFwdRecv) {
           (void)rasCollReadyResp(coll);
         } else {
-          *nextWakeup = std::min(*nextWakeup, coll->startTime+coll->timeout+RAS_COLLECTIVE_EXTRA_TIMEOUT);
-        }
-      } // conn->nFwdRecv < conn->nFwdSent
-    } else {
-      *nextWakeup = std::min(*nextWakeup, coll->startTime+coll->timeout);
-    }
-  } // for (collIdx)
+          // At least some of the delays are *not* due to this process' connections experiencing delays, i.e., they
+          // must be due to delays at other processes.  Presumably those processes will give up waiting soon and the
+          // (incomplete) responses will arrive shortly, so we should wait a little longer.
+          if (now - coll->startTime > coll->timeout + RAS_COLLECTIVE_EXTRA_TIMEOUT) {
+            // We've exceeded even the longer timeout, which is unexpected.  Try to return whatever we have (though
+            // the originator of the collective, if it's not us, may have timed out already anyway).
+            INFO(NCCL_RAS, "RAS collective %s:%ld timeout error (%lds) -- giving up on %d missing responses",
+                 ncclSocketToString(&coll->rootAddr, rasLine), coll->rootId,
+                 (now - coll->startTime) / CLOCK_UNITS_PER_SEC, coll->nFwdSent - coll->nFwdRecv);
+            coll->nLegTimeouts += coll->nFwdSent - coll->nFwdRecv;
+            coll->nFwdRecv = coll->nFwdSent;
+            (void)rasCollReadyResp(coll);
+          } else {
+            *nextWakeup = std::min(*nextWakeup, coll->startTime+coll->timeout+RAS_COLLECTIVE_EXTRA_TIMEOUT);
+          }
+        } // conn->nFwdRecv < conn->nFwdSent
+      } else {
+        *nextWakeup = std::min(*nextWakeup, coll->startTime+coll->timeout);
+      }
+    } // if (coll->timeout > 0)
+
+    coll = collNext;
+  } // for (coll)
 }
 
 
@@ -476,15 +490,16 @@ void rasCollsHandleTimeouts(int64_t now, int64_t* nextWakeup) {
 // For this particular collective, we keep some reduced statistical data (min/max/avg travel time) as well
 // as connection-specific info in case we observed a negative min travel time (which, ideally, shouldn't happen,
 // but the system clocks may not be perfectly in sync).
-static ncclResult_t rasCollConnsInit(char** pData, int* pNData) {
+static ncclResult_t rasCollConnsInit(struct rasCollRequest** pReq, size_t* pReqLen, char** pData, int* pNData) {
   struct rasCollConns connsData = {.travelTimeMin = INT64_MAX, .travelTimeMax = INT64_MIN};
   struct rasCollConns* pConnsData;
 
+  *pReqLen = rasCollDataLength(RAS_COLL_CONNS);
+
   // Update the statistical data first and in the process also calculate how much connection-specific space we
   // will need.
-  for (int i = 0; i < nRasConns; i++) {
-    struct rasConnection* conn = rasConns+i;
-    if (conn->inUse && conn->travelTimeCount > 0) {
+  for (struct rasConnection* conn = rasConnsHead; conn; conn = conn->next) {
+    if (conn->travelTimeCount > 0) {
       if (connsData.travelTimeMin > conn->travelTimeMin)
         connsData.travelTimeMin = conn->travelTimeMin;
       if (connsData.travelTimeMax < conn->travelTimeMax)
@@ -502,9 +517,9 @@ static ncclResult_t rasCollConnsInit(char** pData, int* pNData) {
   pConnsData = (struct rasCollConns*)*pData;
   memcpy(pConnsData, &connsData, sizeof(*pConnsData));
   if (connsData.nNegativeMins > 0) {
-    for (int i = 0, negMinsIdx = 0; i < nRasConns; i++) {
-      struct rasConnection* conn = rasConns+i;
-      if (conn->inUse && conn->travelTimeMin < 0) {
+    int negMinsIdx = 0;
+    for (struct rasConnection* conn = rasConnsHead; conn; conn = conn->next) {
+      if (conn->travelTimeMin < 0) {
         struct rasCollConns::negativeMin* negativeMin = pConnsData->negativeMins+negMinsIdx;
         memcpy(&negativeMin->source, &rasNetListeningSocket.addr, sizeof(negativeMin->source));
         memcpy(&negativeMin->dest, &conn->addr, sizeof(negativeMin->dest));
@@ -560,10 +575,26 @@ static ncclResult_t rasCollConnsMerge(struct rasCollective* coll, struct rasMsg*
 // Initializes the accumulated data with just the local data for now.
 // For this particular collective, we keep for every communicator information about every rank, to help identify
 // the missing ones and the discrepancies between the ones that did respond.
-static ncclResult_t rasCollCommsInit(char** pData, int* pNData) {
+// For any new (previously unseen) communicator we also save the basic identification data about every rank that is
+// "missing" (i.e., not part of this process).  During merging, this should be replaced by the actual data from
+// those ranks, if they are responsive.  We want to provide this information to the user (so that we can say more
+// than "rank xyz missing").
+// Every "new" communicator is also recorded in the (updated) request, so that when that request is forwarded to our
+// peers, those peers don't needlessly send us the same data.
+static ncclResult_t rasCollCommsInit(struct rasCollRequest** pReq, size_t* pReqLen, char** pData, int* pNData) {
+  ncclResult_t ret = ncclSuccess;
   struct rasCollComms* commsData;
-  int nComms = 0, nRanks = 0;
+  int nComms = 0, nRanks = 0, nMissingRanks = 0;
+  bool skipMissing = false;
   std::lock_guard<std::mutex> lock(ncclCommsMutex);
+  struct rasCollComms::comm* comm;
+  struct rasCollRequest* req = nullptr;
+  struct rasPeerInfo** peersReSorted = nullptr;
+  int firstNewSkipMissingIdx = -1;
+
+  *pReqLen = rasCollDataLength(RAS_COLL_COMMS) +
+    (*pReq)->comms.nSkipMissingRanksComms * sizeof(*(*pReq)->comms.skipMissingRanksComms);
+  *pData = nullptr;
 
   // Start by counting the communicators so that we know how much space to allocate.
   // We also need to sort the comms array, to make the subsequent merging easier, both between the ranks (in case
@@ -572,77 +603,152 @@ static ncclResult_t rasCollCommsInit(char** pData, int* pNData) {
     qsort(ncclComms, nNcclComms, sizeof(*ncclComms), &ncclCommsCompare);
     ncclCommsSorted = true;
   }
-  for (int i = 0; i < nNcclComms; i++) {
-    if (ncclComms[i] == nullptr) // nullptr's are always at the end after sorting.
+  for (int commIdx = 0; commIdx < nNcclComms; commIdx++) {
+    if (ncclComms[commIdx] == nullptr) // nullptr's are always at the end after sorting.
       break;
-    if (i == 0) {
-      nComms = 1;
-    } else if (ncclComms[i]->commHash != ncclComms[i-1]->commHash) {
-      nComms++;
-    }
-    nRanks++;
-  }
+    // A process may manage multiple GPUs and thus have multiple communicators with the same commHash.
+    // Comparing just the commHash is OK though within communicators that are part of the same process.
+    if (commIdx == 0 || ncclComms[commIdx]->commHash != ncclComms[commIdx-1]->commHash) {
+      skipMissing = rasCollCommsSkipMissing(*pReq, ncclComms[commIdx]);
+      if (!skipMissing) {
+        // Add this communicator to the request so that the processes we forward the request to know not to fill in
+        // the missing rank info.
+        struct rasCommId* skipComm;
+        if (req == nullptr) {
+          // We pessimistically allocate space for all the remaining communicators so that we don't need to reallocate.
+          int newSize = *pReqLen + (nNcclComms-commIdx) * sizeof(*req->comms.skipMissingRanksComms);
+          NCCLCHECKGOTO(ncclCalloc((char**)&req, newSize), ret, fail);
+          memcpy(req, *pReq, *pReqLen);
+          *pReq = req;
+          firstNewSkipMissingIdx = req->comms.nSkipMissingRanksComms;
+        }
+        skipComm = req->comms.skipMissingRanksComms + req->comms.nSkipMissingRanksComms++;
+        skipComm->commHash = ncclComms[commIdx]->commHash;
+        skipComm->hostHash = ncclComms[commIdx]->peerInfo->hostHash;
+        skipComm->pidHash = ncclComms[commIdx]->peerInfo->pidHash;
 
-  // rasNetCollCommsData has nested variable-length arrays, which makes the size calculation and subsequent
+        nMissingRanks += ncclComms[commIdx]->nRanks;
+      } // if (!skipMissing)
+      nComms++;
+    } // if encountered a new communicator
+    nRanks++;
+    if (!skipMissing)
+      nMissingRanks--;
+  } // for (commIdx)
+
+  // rasCollComms has nested variable-length arrays, which makes the size calculation and subsequent
   // pointer manipulations somewhat unwieldy...
-  *pNData = sizeof(*commsData) + nComms * sizeof(*commsData->comms) + nRanks * sizeof(*commsData->comms[0].ranks);
-  NCCLCHECK(ncclCalloc(pData, *pNData));
+  // This is extra complicated because of the "hidden" array of struct rasCollCommsMissingRank following the
+  // ranks array for each communicator.
+  *pNData = sizeof(*commsData) + nComms * sizeof(*commsData->comms) + nRanks * sizeof(*commsData->comms[0].ranks) +
+    nMissingRanks * sizeof(struct rasCollCommsMissingRank);
+  NCCLCHECKGOTO(ncclCalloc(pData, *pNData), ret, fail);
   commsData = (struct rasCollComms*)*pData;
   commsData->nComms = nComms;
 
   // comm points at the space in the accumulated data where the info about the current communicator is to be stored.
-  struct rasCollComms::comm* comm = commsData->comms;
-  for (int i = 0; i < nNcclComms; i++) {
-    struct rasCollComms::comm::rank* rank;
-    ncclResult_t asyncError;
-    if (ncclComms[i] == nullptr)
-      break;
-    if (i == 0 || ncclComms[i]->commHash != ncclComms[i-1]->commHash) {
-      if (i > 0)
-        comm = (struct rasCollComms::comm*)(((char*)(comm+1)) + comm->nRanks * sizeof(*comm->ranks));
-      comm->commHash = ncclComms[i]->commHash;
-      comm->commNRanks = ncclComms[i]->nRanks;
-      comm->nRanks = 0;
-    } else if (ncclComms[i]->nRanks != ncclComms[i-1]->nRanks) {
-      INFO(NCCL_RAS, "RAS encountered inconsistent communicator data: size %d != %d -- "
-           "possible commHash collision (0x%lx)", ncclComms[i-1]->nRanks, ncclComms[i]->nRanks, comm->commHash);
-      continue; // Short of failing, the best we can do is skip...
-    } else if (ncclComms[i]->rank == ncclComms[i-1]->rank) {
-      INFO(NCCL_RAS, "RAS encountered duplicate data for rank %d -- possible commHash collision (0x%lx)",
-           ncclComms[i]->rank, comm->commHash);
-      continue; // Short of failing, the best we can do is skip...
-    }
-    if (comm->nRanks == comm->commNRanks) {
-      INFO(NCCL_RAS,
-           "RAS encountered more ranks than the communicator size (%d) -- possible commHash collision (0x%lx)",
-           comm->commNRanks, comm->commHash);
-      continue; // Short of failing, the best we can do is skip...
-    }
-    rank = comm->ranks+comm->nRanks;
-    rank->commRank = ncclComms[i]->rank;
-    // rasNetSendCollReq initializes coll->peers[0] to our rasNetListeningSocket.addr, so peerIdx is initially
-    // always 0.  It will increase after we send this response back to the peer we got the request from.
-    rank->peerIdx = 0;
-    rank->collOpCount = ncclComms[i]->collOpCount;
-    rank->status.initState = ncclComms[i]->initState;
-    if (ncclCommGetAsyncError(ncclComms[i], &asyncError) == ncclSuccess)
-      rank->status.asyncError = asyncError;
-    rank->status.finalizeCalled = (ncclComms[i]->finalizeCalled != 0);
-    rank->status.destroyFlag = (ncclComms[i]->destroyFlag != 0);
-    rank->status.abortFlag = (__atomic_load_n(ncclComms[i]->abortFlag, __ATOMIC_ACQUIRE) != 0);
-    rank->cudaDev = ncclComms[i]->cudaDev;
-    rank->nvmlDev = ncclComms[i]->nvmlDev;
-    comm->nRanks++;
-  }
-  assert(nComms == 0 || ((char*)(comm->ranks+comm->nRanks)) - (char*)commsData <= *pNData);
+  comm = commsData->comms;
+  // collCommIdx counts rasCollComms::comm (comm); commIdx indexes ncclComms.
+  for (int collCommIdx = 0, commIdx = 0; collCommIdx < nComms; collCommIdx++) {
+    struct ncclComm* ncclComm = ncclComms[commIdx];
 
-  return ncclSuccess;
+    comm->commId.commHash = ncclComm->commHash;
+    comm->commId.hostHash = ncclComm->peerInfo->hostHash;
+    comm->commId.pidHash = ncclComm->peerInfo->pidHash;
+    comm->commNRanks = ncclComm->nRanks;
+    comm->nRanks = comm->nMissingRanks = 0;
+
+    // Fill in the comm->ranks array.
+    for (; commIdx < nNcclComms && ncclComms[commIdx] && ncclComms[commIdx]->commHash == comm->commId.commHash;
+         commIdx++) {
+      ncclComm = ncclComms[commIdx];
+      struct rasCollComms::comm::rank* rank = comm->ranks+comm->nRanks;
+      ncclResult_t asyncError;
+      rank->commRank = ncclComm->rank;
+      // rasNetSendCollReq initializes coll->peers[0] to our rasNetListeningSocket.addr, so peerIdx is initially
+      // always 0.  It will increase after we send this response back to the peer we got the request from.
+      rank->peerIdx = 0;
+      memcpy(rank->collOpCounts, ncclComm->seqNumber, sizeof(rank->collOpCounts));
+      rank->status.initState = ncclComm->initState;
+      if (ncclCommGetAsyncError(ncclComm, &asyncError) == ncclSuccess)
+        rank->status.asyncError = asyncError;
+      rank->status.finalizeCalled = (ncclComm->finalizeCalled != 0);
+      rank->status.destroyFlag = (ncclComm->destroyFlag != 0);
+      rank->status.abortFlag = (__atomic_load_n(ncclComm->abortFlag, __ATOMIC_ACQUIRE) != 0);
+      rank->cudaDev = ncclComm->cudaDev;
+      rank->nvmlDev = ncclComm->nvmlDev;
+      comm->nRanks++;
+    } // for (commIdx)
+
+    if (firstNewSkipMissingIdx != -1 &&
+        memcmp(req->comms.skipMissingRanksComms+firstNewSkipMissingIdx, &comm->commId, sizeof(comm->commId)) == 0) {
+      // Fill in the missingRanks array that follows the comm->ranks.
+      struct rasCollCommsMissingRank* missingRanks = (struct rasCollCommsMissingRank*)(comm->ranks+comm->nRanks);
+
+      if (peersReSorted == nullptr) {
+        // Create a lookup table to rasPeers that is sorted by hostHash and pidHash, to reduce the complexity of the
+        // lookups in the missingRankIdx loop below.
+        NCCLCHECKGOTO(ncclCalloc(&peersReSorted, nRasPeers), ret, fail);
+        for (int peerIdx = 0; peerIdx < nRasPeers; peerIdx++)
+          peersReSorted[peerIdx] = rasPeers+peerIdx;
+        qsort(peersReSorted, nRasPeers, sizeof(*peersReSorted), peersHashesCompare);
+      }
+
+      comm->nMissingRanks = comm->commNRanks - comm->nRanks;
+      for (int missingRankIdx = 0, rankIdx = 0; missingRankIdx < comm->nMissingRanks; missingRankIdx++) {
+        struct rasCollCommsMissingRank* missingRank;
+        struct ncclPeerInfo* info;
+        struct rasPeerInfo** peer;
+        uint64_t key[2];
+        // Look for the next "hole" in the ranks array.
+        while (rankIdx < comm->nRanks && comm->ranks[rankIdx].commRank == rankIdx+missingRankIdx)
+          rankIdx++;
+
+        missingRank = missingRanks + missingRankIdx;
+        missingRank->commRank = rankIdx + missingRankIdx;
+        info = ncclComm->peerInfo + missingRank->commRank;
+        key[0] = info->hostHash - ncclComm->commHash;
+        key[1] = info->pidHash - ncclComm->commHash;
+        peer = (struct rasPeerInfo**)bsearch(key, peersReSorted, nRasPeers, sizeof(*peersReSorted), peersHashesSearch);
+        if (peer)
+          memcpy(&missingRank->addr, &(*peer)->addr, sizeof(missingRank->addr));
+        missingRank->cudaDev = info->cudaDev;
+        missingRank->nvmlDev = info->nvmlDev;
+      } // for (missingRankIdx)
+
+      if (++firstNewSkipMissingIdx == req->comms.nSkipMissingRanksComms)
+        firstNewSkipMissingIdx = -1;
+    } // if need to fill in the missingRanks
+
+    comm = (struct rasCollComms::comm*)(((char*)(comm+1)) + comm->nRanks * sizeof(*comm->ranks) +
+                                        comm->nMissingRanks * sizeof(struct rasCollCommsMissingRank));
+  } // for (collCommIdx)
+  assert(((char*)comm) - (char*)commsData <= *pNData);
+
+  if (req) {
+    // Finish updating the request.
+    *pReqLen = rasCollDataLength(RAS_COLL_COMMS) +
+      req->comms.nSkipMissingRanksComms * sizeof(*req->comms.skipMissingRanksComms);
+    qsort(req->comms.skipMissingRanksComms, req->comms.nSkipMissingRanksComms,
+          sizeof(*req->comms.skipMissingRanksComms), rasCommIdCompare);
+  }
+ret:
+  free(peersReSorted);
+  return ret;
+fail:
+  if (req) {
+    free(req);
+    *pReq = nullptr;
+  }
+  free(*pData);
+  *pData = nullptr;
+  goto ret;
 }
 
 // Merges incoming collective RAS_COLL_COMMS response message into the local accumulated data.
 static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg* msg) {
-  struct rasCollComms* collData;
-  struct rasCollComms* msgData;
+  struct rasCollComms* collData; // Data previously stored (locally) by our process.
+  struct rasCollComms* msgData; // Data just received from another process.
   int dataOffset = rasMsgLength(RAS_MSG_COLLRESP) + msg->collResp.nPeers*sizeof(*msg->collResp.peers);
   ALIGN_SIZE(dataOffset, alignof(int64_t));
 
@@ -650,7 +756,7 @@ static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg*
   collData = (struct rasCollComms*)coll->data;
 
   if (msgData->nComms > 0) {
-    struct rasCollComms* newData = nullptr;
+    struct rasCollComms* newData = nullptr; // Destination buffer for the merged data.
 
     // Allocate the new buffer pessimistically (sized as the sum of the two old ones).
     NCCLCHECK(ncclCalloc((char**)&newData, coll->nData + msg->collResp.nData));
@@ -661,25 +767,28 @@ static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg*
     for (int collIdx = 0, msgIdx = 0; collIdx < collData->nComms || msgIdx < msgData->nComms; newData->nComms++) {
       int cmp;
       if (collIdx < collData->nComms && msgIdx < msgData->nComms)
-        cmp = (collComm->commHash < msgComm->commHash ? -1 : (collComm->commHash > msgComm->commHash ? 1 : 0));
+        cmp = rasCommIdCompare(&collComm->commId, &msgComm->commId);
       else
         cmp = (collIdx < collData->nComms ? -1 : 1);
 
       if (cmp == 0 && collComm->commNRanks != msgComm->commNRanks) {
         INFO(NCCL_RAS, "RAS encountered inconsistent communicator data: size %d != %d -- "
-             "possible commHash collision (0x%lx)", collComm->commNRanks, msgComm->commNRanks, collComm->commHash);
+             "possible hash collision (0x%lx, 0x%lx, 0x%lx)", collComm->commNRanks, msgComm->commNRanks,
+             collComm->commId.commHash, collComm->commId.hostHash, collComm->commId.pidHash);
         cmp = (collComm->commNRanks < msgComm->commNRanks ? -1 : 1);
-        // We try to preserve both separately, although the input data might already be messed up anyway...
+        // We try to preserve them both separately...
       }
 
       if (cmp == 0) {
         // Merge the comms.
-        newComm->commHash = collComm->commHash;
+        memcpy(&newComm->commId, &collComm->commId, sizeof(newComm->commId));
         newComm->commNRanks = collComm->commNRanks;
         if (collComm->nRanks + msgComm->nRanks > collComm->commNRanks) {
           INFO(NCCL_RAS,
-               "RAS encountered more ranks (%d) than the communicator size (%d) -- possible commHash collision (0x%lx)",
-               collComm->nRanks + msgComm->nRanks, newComm->commNRanks, newComm->commHash);
+               "RAS encountered more ranks (%d) than the communicator size (%d) -- possible hash collision "
+               "(0x%lx, 0x%lx, 0x%lx)", collComm->nRanks + msgComm->nRanks, newComm->commNRanks,
+               collComm->commId.commHash, collComm->commId.hostHash, collComm->commId.pidHash);
+          newComm->nRanks = newComm->commNRanks;
           // We'll skip the extras in the loop below.
         } else {
           newComm->nRanks = collComm->nRanks + msgComm->nRanks;
@@ -691,16 +800,18 @@ static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg*
           int cmpRank;
           if (newRankIdx == newComm->commNRanks)
             break; // Short of failing, the best we can do is skip...
-          if (collRankIdx < collComm->nRanks && msgRankIdx < msgComm->nRanks)
+          if (collRankIdx < collComm->nRanks && msgRankIdx < msgComm->nRanks) {
             cmpRank = (collComm->ranks[collRankIdx].commRank < msgComm->ranks[msgRankIdx].commRank ? -1 :
                        (collComm->ranks[collRankIdx].commRank > msgComm->ranks[msgRankIdx].commRank ? 1 : 0));
-          else
+          } else {
             cmpRank = (collRankIdx < collComm->nRanks ? -1 : 1);
+          }
 
           // There shouldn't be any overlaps in ranks between different sources.
           if (cmpRank == 0) {
-            INFO(NCCL_RAS, "RAS encountered duplicate data for rank %d -- possible commHash collision (0x%lx)",
-                 collComm->ranks[collRankIdx].commRank, newComm->commHash);
+            INFO(NCCL_RAS, "RAS encountered duplicate data for rank %d -- possible hash collision "
+                 "(0x%lx, 0x%lx, 0x%lx)", collComm->ranks[collRankIdx].commRank,
+                 newComm->commId.commHash, newComm->commId.hostHash, newComm->commId.pidHash);
             msgRankIdx++; // Short of failing, the best we can do is skip...
           }
           memcpy(newComm->ranks+newRankIdx, (cmpRank <= 0 ? collComm->ranks+collRankIdx++ :
@@ -708,23 +819,63 @@ static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg*
           if (cmpRank > 0) {
             // peerIdx values from msgComm need to shift after merge.
             newComm->ranks[newRankIdx].peerIdx += coll->nPeers;
-          }
+
+            if (collComm->nMissingRanks > 0) {
+              // Remove the corresponding entry from missingRanks.
+              struct rasCollCommsMissingRank* missingRank;
+              missingRank = (struct rasCollCommsMissingRank*)bsearch(&newComm->ranks[newRankIdx].commRank,
+                                                                     collComm->ranks+collComm->nRanks,
+                                                                     collComm->nMissingRanks,
+                                                                     sizeof(struct rasCollCommsMissingRank),
+                                                                     rasCollCommsMissingRankSearch);
+              if (missingRank) {
+                // Mark the entry as no longer needed.
+                memset(&missingRank->addr, '\0', sizeof(missingRank->addr));
+              } else {
+                INFO(NCCL_RAS, "RAS failed to find missingRank data -- internal error?");
+              }
+            } // if (collComm->nMissingRanks > 0)
+          } // if (cmpRank > 0)
         } // for (newRankIdx)
-        newComm = (struct rasCollComms::comm*)(((char*)(newComm+1)) + newComm->nRanks * sizeof(*newComm->ranks));
-        collComm = (struct rasCollComms::comm*)(((char*)(collComm+1)) + collComm->nRanks * sizeof(*collComm->ranks));
+        if (collComm->nMissingRanks > 0) {
+          // Copy the missingRanks to newComm, skipping over any no longer needed entries.
+          union ncclSocketAddress emptyAddr;
+          struct rasCollCommsMissingRank* collMissingRanks;
+          struct rasCollCommsMissingRank* newMissingRanks;
+          int newRankIdx;
+
+          memset(&emptyAddr, '\0', sizeof(emptyAddr));
+          collMissingRanks = (struct rasCollCommsMissingRank*)(collComm->ranks+collComm->nRanks);
+          newMissingRanks = (struct rasCollCommsMissingRank*)(newComm->ranks+newComm->nRanks);
+          newRankIdx = 0;
+          for (int collRankIdx = 0; collRankIdx < collComm->nMissingRanks; collRankIdx++) {
+            if (memcmp(&collMissingRanks[collRankIdx].addr, &emptyAddr, sizeof(emptyAddr))) {
+              memcpy(newMissingRanks + newRankIdx++, collMissingRanks + collRankIdx, sizeof(*newMissingRanks));
+            }
+          }
+          newComm->nMissingRanks = newRankIdx;
+          assert(newComm->nRanks + newComm->nMissingRanks == newComm->commNRanks);
+        }
+        newComm = (struct rasCollComms::comm*)(((char*)(newComm+1)) + newComm->nRanks * sizeof(*newComm->ranks) +
+                                               newComm->nMissingRanks * sizeof(struct rasCollCommsMissingRank));
+        collComm = (struct rasCollComms::comm*)(((char*)(collComm+1)) + collComm->nRanks * sizeof(*collComm->ranks) +
+                                                collComm->nMissingRanks * sizeof(struct rasCollCommsMissingRank));
         collIdx++;
-        msgComm = (struct rasCollComms::comm*)(((char*)(msgComm+1)) + msgComm->nRanks * sizeof(*msgComm->ranks));
+        msgComm = (struct rasCollComms::comm*)(((char*)(msgComm+1)) + msgComm->nRanks * sizeof(*msgComm->ranks) +
+                                               msgComm->nMissingRanks * sizeof(struct rasCollCommsMissingRank));
         msgIdx++;
       } else if (cmp < 0) {
         // Copy from collComm.
-        int commSize = sizeof(*collComm) + collComm->nRanks * sizeof(*collComm->ranks);
+        int commSize = sizeof(*collComm) + collComm->nRanks * sizeof(*collComm->ranks) +
+          collComm->nMissingRanks * sizeof(struct rasCollCommsMissingRank);
         memcpy(newComm, collComm, commSize);
         newComm = (struct rasCollComms::comm*)(((char*)(newComm)) + commSize);
         collComm = (struct rasCollComms::comm*)(((char*)(collComm)) + commSize);
         collIdx++;
       } else { // cmp > 0
         // Copy from msgComm.
-        int commSize = sizeof(*msgComm) + msgComm->nRanks * sizeof(*msgComm->ranks);
+        int commSize = sizeof(*msgComm) + msgComm->nRanks * sizeof(*msgComm->ranks) +
+          msgComm->nMissingRanks * sizeof(struct rasCollCommsMissingRank);
         memcpy(newComm, msgComm, commSize);
         for (int i = 0; i < newComm->nRanks; i++) {
           // peerIdx values from msgComm need to shift after merge.
@@ -745,18 +896,87 @@ static ncclResult_t rasCollCommsMerge(struct rasCollective* coll, struct rasMsg*
   return ncclSuccess;
 }
 
+// Checks if a given communicator is in the skipMissingRanksComms array of the request.
+static bool rasCollCommsSkipMissing(const struct rasCollRequest* req, struct ncclComm* comm) {
+  struct rasCommId id;
+  id.commHash = comm->commHash;
+  id.hostHash = comm->peerInfo->hostHash;
+  id.pidHash = comm->peerInfo->pidHash;
+  return (bsearch(&id, req->comms.skipMissingRanksComms, req->comms.nSkipMissingRanksComms,
+                  sizeof(*req->comms.skipMissingRanksComms), rasCommIdCompare) != nullptr);
+}
+
 // Sorting callback for the ncclComms array.
 static int ncclCommsCompare(const void* p1, const void* p2) {
-  const ncclComm** pc1 = (const ncclComm**)p1;
-  const ncclComm** pc2 = (const ncclComm**)p2;
+  const ncclComm* comm1 = *(const ncclComm**)p1;
+  const ncclComm* comm2 = *(const ncclComm**)p2;
 
   // Put nullptr's at the end.
-  if (*pc1 == nullptr || *pc2 == nullptr)
-    return (*pc1 != nullptr ? -1 : (*pc2 != nullptr ? 1 : 0));
+  if (comm1 == nullptr || comm2 == nullptr)
+    return (comm1 != nullptr ? -1 : (comm2 != nullptr ? 1 : 0));
 
-  if ((*pc1)->commHash == (*pc2)->commHash) {
-    return ((*pc1)->rank < (*pc2)->rank ? -1 : ((*pc1)->rank > (*pc2)->rank ? 1 : 0));
+  if (comm1->commHash == comm2->commHash) {
+    return (comm1->rank < comm2->rank ? -1 : (comm1->rank > comm2->rank ? 1 : 0));
   } else {
-    return ((*pc1)->commHash < (*pc2)->commHash ? -1 : 1);
+    return (comm1->commHash < comm2->commHash ? -1 : 1);
   }
 }
+
+// Sorting callback for a lookup table to rasPeers.  Sorts by the hostHash (primary) and pidHash (secondary).
+static int peersHashesCompare(const void* p1, const void* p2) {
+  const struct rasPeerInfo* pi1 = *(const struct rasPeerInfo**)p1;
+  const struct rasPeerInfo* pi2 = *(const struct rasPeerInfo**)p2;
+
+  if (pi1->hostHash == pi2->hostHash) {
+    return (pi1->pidHash < pi2->pidHash ? -1 : (pi1->pidHash > pi2->pidHash ? 1 : 0));
+  } else {
+    return (pi1->hostHash < pi2->hostHash ? -1 : 1);
+  }
+}
+
+// Search callback for a lookup table to rasPeers.  Searches by the hostHash and pidHash.  The key is an array
+// containing the hostHash at index 0 and the pidHash at index 1.
+static int peersHashesSearch(const void* k, const void* e) {
+  const uint64_t* key = (const uint64_t*)k;
+  const struct rasPeerInfo* elem = *(const struct rasPeerInfo**)e;
+
+  if (key[0] == elem->hostHash) {
+    return (key[1] < elem->pidHash ? -1 : (key[1] > elem->pidHash ? 1 : 0));
+  } else {
+    return (key[0] < elem->hostHash ? -1 : 1);
+  }
+}
+
+// Sorting/searching callback for struct rasCommId.  Sorts by commHash, then hostHash, then pidHash.
+static int rasCommIdCompare(const void* p1, const void* p2) {
+  const struct rasCommId* i1 = (const struct rasCommId*)p1;
+  const struct rasCommId* i2 = (const struct rasCommId*)p2;
+  if (i1->commHash == i2->commHash) {
+    if (i1->hostHash == i2->hostHash) {
+      return (i1->pidHash < i2->pidHash ? -1 : (i1->pidHash > i2->pidHash ? 1 : 0));
+    } else {
+      return (i1->hostHash < i2->hostHash ? -1 : 1);
+    }
+  } else {
+    return (i1->commHash < i2->commHash ? -1 : 1);
+  }
+}
+
+// Search callback for rasCollComms::comm rasCollCommsMissingRank array.  The key is the commRank.
+static int rasCollCommsMissingRankSearch(const void* k, const void* e) {
+  int key = *(const int*)k;
+  const struct rasCollCommsMissingRank* elem = (const struct rasCollCommsMissingRank*)e;
+
+  return (key < elem->commRank ? -1 : (key > elem->commRank ? 1 : 0));
+}
+
+// Invoked during RAS termination to release all the allocated resources.
+void rasCollectivesTerminate() {
+  for (struct rasCollective* coll = rasCollectivesHead; coll;) {
+    struct rasCollective* collNext = coll->next;
+    rasCollFree(coll);
+    coll = collNext;
+  }
+
+  // rasCollectivesHead and rasCollectivesTail are taken care of by rasCollFree().
+}

src/ras/peers.cc

@@ -40,10 +40,11 @@ static ncclResult_t rasRanksConvertToPeers(struct rasRankInit* ranks, int nranks
 static ncclResult_t rasPeersUpdate(struct rasPeerInfo* rankPeers, int* nRankPeers, int newNRasPeers = -1);
 
 static ncclResult_t rasNetUpdatePeers(const struct rasPeerInfo* newPeers, int nNewPeers, bool updateDeadPeers,
-                                      struct rasRankInit* ranks = nullptr, int nranks = 0, int fromConnIdx = -1);
+                                      struct rasRankInit* ranks = nullptr, int nranks = 0,
+                                      struct rasConnection* fromConn = nullptr);
 static ncclResult_t rasLinkPropagateUpdate(struct rasLink* link, const struct rasPeerInfo* newPeers, int nNewPeers,
                                            bool updateDeadPeers, struct rasRankInit* ranks, int nranks,
-                                           int fromConnIdx);
+                                           struct rasConnection* fromConn);
 static ncclResult_t rasConnPropagateUpdate(struct rasConnection* conn, const struct rasPeerInfo* newPeers,
                                            int nNewPeers, bool updateDeadPeers, struct rasRankInit* ranks, int nranks);
 ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock);
@@ -146,6 +147,8 @@ static ncclResult_t rasRanksConvertToPeers(struct rasRankInit* ranks, int nranks
     rankPeer->pid = rank->pid;
     rankPeer->cudaDevs = (1UL << rank->cudaDev);
     rankPeer->nvmlDevs = (1UL << rank->nvmlDev);
+    rankPeer->hostHash = rank->hostHash;
+    rankPeer->pidHash = rank->pidHash;
     rankPeerIdx++;
 
     // Also check if there is already an entry with that address in the global rasPeers so that the caller can know how
@@ -357,12 +360,12 @@ int rasPeerFind(const union ncclSocketAddress* addr) {
 // ranks -- if provided -- lists all the peers who are already aware of this update (because they are the members
 // of the new communicator being established), and who thus don't need to be notified.  updatedDeadPeers can
 // be used, however, to request at least the propagation of rasDeadPeers to such peers.
-// fromConnIdx -- if provided -- identified the connection used to receive this update; there's no need to
+// fromConn -- if provided -- identifies the connection used to receive this update; there's no need to
 // propagate the update back through it.
 // Reconfigures the RAS network to accommodate the newly added peers, by modifying the links and establishing new
 // connections as needed.
 static ncclResult_t rasNetUpdatePeers(const struct rasPeerInfo* newPeers, int nNewPeers, bool updateDeadPeers,
-                                      struct rasRankInit* ranks, int nranks, int fromConnIdx) {
+                                      struct rasRankInit* ranks, int nranks, struct rasConnection* fromConn) {
   ncclResult_t ret = ncclSuccess;
 
   // Do we actually have anything to do?
@@ -371,8 +374,8 @@ static ncclResult_t rasNetUpdatePeers(const struct rasPeerInfo* newPeers, int nN
 
   // Start by propagating the update through the RAS network links.  We consider any errors during this process
   // to be non-fatal (we can re-sync later around a keep-alive exchange).
-  (void)rasLinkPropagateUpdate(&rasNextLink, newPeers, nNewPeers, updateDeadPeers, ranks, nranks, fromConnIdx);
-  (void)rasLinkPropagateUpdate(&rasPrevLink, newPeers, nNewPeers, updateDeadPeers, ranks, nranks, fromConnIdx);
+  (void)rasLinkPropagateUpdate(&rasNextLink, newPeers, nNewPeers, updateDeadPeers, ranks, nranks, fromConn);
+  (void)rasLinkPropagateUpdate(&rasPrevLink, newPeers, nNewPeers, updateDeadPeers, ranks, nranks, fromConn);
 
   // Calculate new link peers and open new connections if needed.
   NCCLCHECKGOTO(rasLinkReinitConns(&rasNextLink), ret, fail);
@@ -388,15 +391,13 @@ fail:
 // for the explanation of the function arguments.
 static ncclResult_t rasLinkPropagateUpdate(struct rasLink* link, const struct rasPeerInfo* newPeers, int nNewPeers,
                                            bool updateDeadPeers, struct rasRankInit* ranks, int nranks,
-                                           int fromConnIdx) {
-  for (int i = 0; i < link->nConns; i++) {
-    struct rasLinkConn* linkConn = link->conns+i;
+                                           struct rasConnection* fromConn) {
+  for (struct rasLinkConn* linkConn = link->conns; linkConn; linkConn = linkConn->next) {
     // Note that we don't send the update via the connection that we received this notification from in the first
     // place (while it wouldn't loop indefinitely, it would add a needless extra exchange).
-    if (linkConn->connIdx != -1 && linkConn->connIdx != fromConnIdx) {
-      struct rasConnection* conn = rasConns+linkConn->connIdx;
+    if (linkConn->conn && linkConn->conn != fromConn) {
       // Failed propagations are not considered fatal (we will retry after a keep-alive).
-      (void)rasConnPropagateUpdate(conn, newPeers, nNewPeers, updateDeadPeers, ranks, nranks);
+      (void)rasConnPropagateUpdate(linkConn->conn, newPeers, nNewPeers, updateDeadPeers, ranks, nranks);
     }
   }
 
@@ -407,7 +408,7 @@ static ncclResult_t rasLinkPropagateUpdate(struct rasLink* link, const struct ra
 // arguments.
 static ncclResult_t rasConnPropagateUpdate(struct rasConnection* conn, const struct rasPeerInfo* newPeers,
                                            int nNewPeers, bool updateDeadPeers, struct rasRankInit* ranks, int nranks) {
-  if (conn->sockIdx != -1 && rasSockets[conn->sockIdx].status == RAS_SOCK_READY) {
+  if (conn->sock && conn->sock->status == RAS_SOCK_READY) {
     // If we have the rank info, check if the peer on the other side of this connection has participated in the new
     // communicator.
     int connRank = -1;
@@ -462,7 +463,8 @@ ncclResult_t rasConnSendPeersUpdate(struct rasConnection* conn, const struct ras
   msg->peersUpdate.deadPeersHash = rasDeadPeersHash;
   msg->peersUpdate.nDeadPeers = nDeadPeers;
   memcpy(msg->peersUpdate.peers, peers, nPeers * sizeof(msg->peersUpdate.peers[0]));
-  memcpy(((char*)msg)+deadPeersOffset, rasDeadPeers, nDeadPeers * sizeof(*rasDeadPeers));
+  if (nDeadPeers > 0)
+    memcpy(((char*)msg)+deadPeersOffset, rasDeadPeers, nDeadPeers * sizeof(*rasDeadPeers));
 
   if (nPeers > 0)
     conn->lastSentPeersHash = rasPeersHash;
@@ -485,8 +487,7 @@ ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock)
   ncclResult_t ret = ncclSuccess;
   struct rasMsg* newMsg = nullptr;
   int newMsgLen = 0;
-  assert(sock->connIdx != -1);
-  struct rasConnection* conn = rasConns+sock->connIdx;
+  assert(sock->conn);
   int nPeers, nDeadPeers;
   int deadPeersOffset = 0;
   bool updatePeers, updateDeadPeers;
@@ -496,8 +497,8 @@ ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock)
        msg->peersUpdate.nPeers, msg->peersUpdate.nDeadPeers);
   INFO(NCCL_RAS, "RAS my old rasPeersHash 0x%lx, rasDeadPeersHash 0x%lx, nRasPeers %d, nRasDeadPeers %d",
        rasPeersHash, rasDeadPeersHash, nRasPeers, nRasDeadPeers);
-  conn->lastRecvPeersHash = msg->peersUpdate.peersHash;
-  conn->lastRecvDeadPeersHash = msg->peersUpdate.deadPeersHash;
+  sock->conn->lastRecvPeersHash = msg->peersUpdate.peersHash;
+  sock->conn->lastRecvDeadPeersHash = msg->peersUpdate.deadPeersHash;
 
   // Prepare ours to send back.  We don't enqueue it right away because we want to make sure first that we need
   // to send it.  We'll find out by comparing the hash values after the merge.
@@ -545,15 +546,15 @@ ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock)
       rasDeadPeersDump();
 
     // If post-merge the hashes are still different, send our (dead) peers back.
-    updatePeers = (conn->lastSentPeersHash != rasPeersHash && conn->lastRecvPeersHash != rasPeersHash);
-    updateDeadPeers = (conn->lastSentDeadPeersHash != rasDeadPeersHash &&
-                       conn->lastRecvDeadPeersHash != rasDeadPeersHash);
+    updatePeers = (sock->conn->lastSentPeersHash != rasPeersHash && sock->conn->lastRecvPeersHash != rasPeersHash);
+    updateDeadPeers = (sock->conn->lastSentDeadPeersHash != rasDeadPeersHash &&
+                       sock->conn->lastRecvDeadPeersHash != rasDeadPeersHash);
     if (updatePeers || updateDeadPeers) {
       newMsg->peersUpdate.peersHash = rasPeersHash;
       newMsg->peersUpdate.deadPeersHash = rasDeadPeersHash;
       if (updatePeers) {
         assert(nPeers > 0);
-        conn->lastSentPeersHash = rasPeersHash;
+        sock->conn->lastSentPeersHash = rasPeersHash;
       } else {
         // If hashes match, make sure that we don't send the rasPeers back.
         newMsg->peersUpdate.nPeers = 0;
@@ -564,14 +565,14 @@ ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock)
 
       if (updateDeadPeers) {
         assert(nRasDeadPeers > 0);
-        conn->lastSentDeadPeersHash = rasDeadPeersHash;
+        sock->conn->lastSentDeadPeersHash = rasDeadPeersHash;
 
         ALIGN_SIZE(newMsgLen, alignof(union ncclSocketAddress));
         deadPeersOffset = newMsgLen;
         newMsgLen += nRasDeadPeers*sizeof(*rasDeadPeers);
 
         memcpy(((char*)newMsg)+deadPeersOffset, rasDeadPeers, nDeadPeers * sizeof(*rasDeadPeers));
-        conn->lastSentDeadPeersHash = rasDeadPeersHash;
+        sock->conn->lastSentDeadPeersHash = rasDeadPeersHash;
         newMsg->peersUpdate.nDeadPeers = nRasDeadPeers;
       } else {
         newMsg->peersUpdate.nDeadPeers = 0;
@@ -580,13 +581,13 @@ ncclResult_t rasMsgHandlePeersUpdate(struct rasMsg* msg, struct rasSocket* sock)
       INFO(NCCL_RAS, "RAS sending back a peersUpdate (nPeers %d, nDeadPeers %d)",
            newMsg->peersUpdate.nPeers, newMsg->peersUpdate.nDeadPeers);
 
-      rasConnEnqueueMsg(conn, newMsg, newMsgLen);
+      rasConnEnqueueMsg(sock->conn, newMsg, newMsgLen);
       newMsg = nullptr;
     } // if (updatePeers || updateDeadPeers)
 
     // Propagate the changes through our RAS network links.
     NCCLCHECKGOTO(rasNetUpdatePeers(msg->peersUpdate.peers, msg->peersUpdate.nPeers, updateDeadPeers, nullptr, 0,
-                                    sock->connIdx), ret, fail);
+                                    sock->conn), ret, fail);
   }
 
 exit:
@@ -603,7 +604,7 @@ fail:
 
 // Reinitializes the connection(s) of a particular link, following a peers update.
 // Adding new peers can affect the calculation of the link's primary connection and also the fallbacks.
-// The newly added peers could also shift all the existing peerIdx values, invalidating the values in RasLinkConn
+// The newly added peers could also shift all the existing peerIdx values, invalidating the values in rasLinkConn
 // structures, so it's better to drop it all and recalculate from scratch.
 // We recalculate the primary peer; if an active connection to it already exists, then we're done.  If there
 // is no connection, we create one.  If a connection exists but is experiencing delays then we add a fallback and
@@ -611,77 +612,51 @@ fail:
 // External conns are dropped from the links as well (they will be re-created via keepAlive messages as needed).
 static ncclResult_t rasLinkReinitConns(struct rasLink* link) {
   struct rasLinkConn* linkConn;
-  struct rasConnection* conn = nullptr;
   int newPeerIdx = myPeerIdx;
 
-  if (link->connsSize == 0) {
-    link->connsSize = RAS_INCREMENT;
-    NCCLCHECK(ncclCalloc(&link->conns, link->connsSize));
+  if (link->conns) {
+    // Free the old contents but keep the first entry for convenience (though wipe it).
+    for (struct rasLinkConn* linkConn = link->conns->next; linkConn;) {
+      struct rasLinkConn* linkConnNext = linkConn->next;
+      free(linkConn);
+      linkConn = linkConnNext;
+    }
+    memset(link->conns, '\0', sizeof(*link->conns));
+    link->lastUpdatePeersTime = 0;
+  } else { // link->conns == nullptr
+    NCCLCHECK(ncclCalloc(&link->conns, 1));
   }
-  link->nConns = 0;
 
-  // Establish a connection for this link.  We iterate as long as the connections we find are experiencing delays.
-  while (newPeerIdx != -1) {
-    if (link->nConns == link->connsSize) {
-      NCCLCHECK(ncclRealloc(&link->conns, link->connsSize, link->connsSize+RAS_INCREMENT));
-      link->connsSize += RAS_INCREMENT;
-    }
+  // Fill in the entry for the primary connection.
+  linkConn = link->conns;
+  linkConn->peerIdx = newPeerIdx = rasLinkCalculatePeer(link, myPeerIdx, /*isFallback*/false);
+  linkConn->conn = (newPeerIdx != -1 ? rasConnFind(&rasPeers[newPeerIdx].addr) : nullptr);
+  linkConn->external = false;
 
-    newPeerIdx = rasLinkCalculatePeer(link, newPeerIdx, /*isFallback*/link->nConns > 1);
-    if (newPeerIdx == -1) {
-      INFO(NCCL_RAS, "RAS link %d: no more fallbacks to add (nConns %d)", link->direction, link->nConns);
-      if (link->nConns > 0)
-        break;
-    }
-    linkConn = link->conns+link->nConns;
-    linkConn->peerIdx = newPeerIdx;
-    linkConn->connIdx = (newPeerIdx != -1 ? rasConnFind(&rasPeers[newPeerIdx].addr) : -1);
-    linkConn->external = false;
-
-    // If the calculated connection does not exist, then we are at the end of the chain and this is the last iteration.
-    // Depending on the circumstances, we may first need to create that connection.
-    if (linkConn->connIdx == - 1) {
-      if (link->nConns == 0) {
-        if (linkConn->peerIdx != -1) {
-          INFO(NCCL_RAS, "RAS link %d: %s primary connection with %s",
-               link->direction, (myPeerIdx < linkConn->peerIdx ? "opening new" : "calculated deferred"),
-               ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
-          // We try to initiate primary connections from the side with a lower address (and thus an earlier peer index)
-          // to avoid races and the creation of duplicate connections.
-          if (myPeerIdx < linkConn->peerIdx) {
-            NCCLCHECK(rasConnCreate(&rasPeers[linkConn->peerIdx].addr, &linkConn->connIdx));
-          }
-          else { // If we didn't initiate the connection, start the timeout.
-            link->lastUpdatePeersTime = clockNano();
-          }
-        } // if (linkConn->peerIdx != -1)
-      } else { // link->nConns > 0
-        INFO(NCCL_RAS, "RAS link %d: opening new fallback connection %d with %s",
-             link->direction, link->nConns, ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
-        NCCLCHECK(rasConnCreate(&rasPeers[newPeerIdx].addr, &linkConn->connIdx));
-      } // link->nConns > 0
-    } else { // linkConn->connIdx != -1
-      if (link->nConns == 0) {
-        INFO(NCCL_RAS, "RAS link %d: calculated existing primary connection with %s",
-             link->direction, ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
-      } else {
-        INFO(NCCL_RAS, "RAS link %d: calculated existing fallback connection %d with %s",
-             link->direction, link->nConns, ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
+  if (linkConn->conn == nullptr) {
+    if (linkConn->peerIdx != -1) {
+      // We try to initiate primary connections from the side with a lower address (and thus an earlier peer index)
+      // to avoid races and the creation of duplicate connections.
+      INFO(NCCL_RAS, "RAS link %d: %s primary connection with %s",
+           link->direction, (myPeerIdx < linkConn->peerIdx ? "opening new" : "calculated deferred"),
+           ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
+      if (myPeerIdx < linkConn->peerIdx) {
+        NCCLCHECK(rasConnCreate(&rasPeers[linkConn->peerIdx].addr, &linkConn->conn));
       }
-    }
-    link->nConns++;
-    if (linkConn->connIdx == -1)
-      break;
-    conn = rasConns+linkConn->connIdx;
-
-    // We check if the connection already went through the fallback calculation; if so, we'll need to create a new
-    // fallback in the next iteration, to ensure that RAS will keep retrying.
-    if (!conn->experiencingDelays)
-      break;
+      else { // If we didn't initiate the connection, start the timeout.
+        link->lastUpdatePeersTime = clockNano();
+      }
+    } // if (linkConn->peerIdx != -1)
+  } else { // linkConn->conn
+    INFO(NCCL_RAS, "RAS link %d: calculated existing primary connection with %s",
+         link->direction, ncclSocketToString(&rasPeers[linkConn->peerIdx].addr, rasLine));
+  } // linkConn->conn
 
+  if (linkConn->conn && linkConn->conn->experiencingDelays) {
     INFO(NCCL_RAS, "RAS connection experiencingDelays %d, startRetryTime %.2fs, socket status %d",
-         conn->experiencingDelays, (clockNano()-conn->startRetryTime)/1e9,
-         (conn->sockIdx == -1 ? -1 : rasSockets[conn->sockIdx].status));
+         linkConn->conn->experiencingDelays, (clockNano()-linkConn->conn->startRetryTime)/1e9,
+         (linkConn->conn->sock ? linkConn->conn->sock->status : - 1));
+    NCCLCHECK(rasLinkAddFallback(link, linkConn->conn));
   }
 
   return ncclSuccess;
@@ -701,39 +676,37 @@ int rasLinkCalculatePeer(const struct rasLink* link, int peerIdx, bool isFallbac
     if (isFallback && !ncclSocketsSameNode(&rasPeers[peerIdx].addr, &rasNetListeningSocket.addr)) {
       // peerIdx is a fallback and it is not running on the same node as us.
       int tryPeerIdx = newPeerIdx;
-      int tryConnIdx = -1;
+      struct rasConnection* tryConn = nullptr;
 
       // Try to skip the remaining peers on the same node as peerIdx.  We may end up skipping over some peers that
       // are alive, which is fine -- they will still have connectivity with the rest of the RAS network, just a
       // little suboptimal one.
       while (ncclSocketsSameNode(&rasPeers[tryPeerIdx].addr, &rasPeers[peerIdx].addr)) {
         if (!rasPeerIsDead(&rasPeers[tryPeerIdx].addr)) {
-          tryConnIdx = rasConnFind(&rasPeers[tryPeerIdx].addr);
-          if (tryConnIdx != -1) {
-            struct rasConnection* tryConn = rasConns+tryConnIdx;
+          tryConn = rasConnFind(&rasPeers[tryPeerIdx].addr);
+          if (tryConn) {
             // Check if the connection is fully established and operational, i.e., if the underlying socket
             // is ready and there's been recent communication on it.
-            if (tryConn->sockIdx != -1 && rasSockets[tryConn->sockIdx].status == RAS_SOCK_READY &&
-                !tryConn->experiencingDelays) {
+            if (tryConn->sock && tryConn->sock->status == RAS_SOCK_READY && !tryConn->experiencingDelays) {
               // We convinced ourselves that the node is not down.  We don't adjust newPeerIdx in
               // this case.  This is the only case when tryConnIdx != -1 after this loop.
               break;
             }
-          } // if (tryConnIdx != -1)
+          } // if (tryConn)
         } // if (!rasPeerIsDead(&rasPeers[tryPeerIdx].addr))
 
-        tryConnIdx = -1;
-        tryPeerIdx = (tryPeerIdx + nRasPeers + link->direction) % nRasPeers;
+        tryConn = nullptr;
+        tryPeerIdx = (tryPeerIdx + link->direction + nRasPeers) % nRasPeers;
         if (tryPeerIdx == myPeerIdx)
           break;
       }
 
-      if (tryConnIdx == -1)
+      if (tryConn == nullptr)
         newPeerIdx = tryPeerIdx;
       if (tryPeerIdx == myPeerIdx)
         break;
     } // if (isFallback && !ncclSocketsSameNode(&rasPeers[peerIdx].addr, &rasNetListeningSocket.addr))
-    
+
     if (rasPeerIsDead(&rasPeers[newPeerIdx].addr)) {
       newPeerIdx = (newPeerIdx + nRasPeers + link->direction) % nRasPeers;
     }
@@ -932,7 +905,8 @@ bool ncclSocketsSameNode(const union ncclSocketAddress* a1, const union ncclSock
 static void rasPeersDump() {
   for (int p = 0; p < nRasPeers; p++) {
     const struct rasPeerInfo* peer = rasPeers+p;
-    INFO(NCCL_RAS, "RAS peer %d: %s%s", p, rasPeerDump(peer, rasLine, sizeof(rasLine)), (p == myPeerIdx ? " [this process]" : ""));
+    INFO(NCCL_RAS, "RAS peer %d: %s%s", p, rasPeerDump(peer, rasLine, sizeof(rasLine)),
+         (p == myPeerIdx ? " [this process]" : ""));
   }
   if (nRasPeers > 0)
     INFO(NCCL_RAS, "RAS peersHash 0x%lx", rasPeersHash);
@@ -958,3 +932,17 @@ static char* rasPeerDump(const struct rasPeerInfo* peer, char* result, size_t nr
            rasGpuDevsToString(peer->cudaDevs, peer->nvmlDevs, line2, sizeof(line2)));
   return result;
 }
+
+// Invoked during RAS termination to release all the allocated resources.
+void rasPeersTerminate() {
+  free(rasPeers);
+  rasPeers = nullptr;
+  nRasPeers = 0;
+  rasPeersHash = 0;
+  myPeerIdx = -1;
+
+  free(rasDeadPeers);
+  rasDeadPeers = nullptr;
+  nRasDeadPeers = rasDeadPeersSize = 0;
+  rasDeadPeersHash = 0;
+}

src/ras/ras.cc

@@ -4,8 +4,10 @@
  * See LICENSE.txt for license information
  ************************************************************************/
 
-#define NDEBUG // Comment out during development only!
-#include <cassert>
+// Workaround for libstdc++ trying to force public visibility of std:: symbols.  We don't want to do that in libnccl.so.
+#include <bits/c++config.h>
+#undef _GLIBCXX_VISIBILITY
+#define _GLIBCXX_VISIBILITY(V)
 #include <cstddef>
 #include <mutex>
 #include <poll.h>
@@ -65,8 +67,8 @@ int nNcclComms = 0;
 bool ncclCommsSorted = false; // Whether the array is currently sorted. We sort by the comms' commHash and rank.
 
 static ncclResult_t rasLocalNotify(const struct rasNotification* msg);
-static ncclResult_t rasLocalHandle();
-static void rasLocalHandleTerminate();
+static ncclResult_t rasLocalHandle(bool* terminate);
+static void rasThreadCleanup();
 
 static ncclResult_t rasMsgHandleConnInit(const struct rasMsg* msg, struct rasSocket* sock);
 static ncclResult_t rasMsgHandleConnInitAck(const struct rasMsg* msg, struct rasSocket* sock);
@@ -74,6 +76,8 @@ static ncclResult_t rasNetSendNack(struct rasSocket* sock);
 
 static void* rasThreadMain(void*);
 
+static void rasTerminate() __attribute__((destructor));
+
 NCCL_PARAM(RasTimeoutFactor, "RAS_TIMEOUT_FACTOR", 1);
 
 //////////////////////////////////////////////////
@@ -105,7 +109,6 @@ ncclResult_t ncclRasCommInit(struct ncclComm* comm, struct rasRankInit* myRank)
 
       PTHREADCHECKGOTO(pthread_create(&rasThread, nullptr, &rasThreadMain, nullptr), "pthread_create", ret, fail);
       ncclSetThreadName(rasThread, "NCCL RAS");
-      (void)pthread_detach(rasThread);
 
       rasInitialized = true;
     }
@@ -157,18 +160,27 @@ ncclResult_t ncclRasCommFini(const struct ncclComm* comm) {
       }
     }
   }
-  if (ncclAtomicRefCountDecrement(&rasInitRefCount) == 0) {
-    struct rasNotification msg;
-    msg.type = RAS_TERMINATE;
-    NCCLCHECK(rasLocalNotify(&msg));
-  }
+  ncclAtomicRefCountDecrement(&rasInitRefCount);
   return ncclSuccess;
 }
 
+// Global destructor.  Notifies the RAS thread to release all the resources
+// and terminate.  Waits for the thread to terminate.
+static void rasTerminate() {
+  struct rasNotification msg;
+  if (!rasInitialized)
+    return;
+  memset(&msg, '\0', sizeof(msg));
+  msg.type = RAS_TERMINATE;
+  if (rasLocalNotify(&msg) == ncclSuccess)
+    (void)pthread_join(rasThread, nullptr);
+}
+
 // Invoked by regular NCCL threads on every (non-split) comm initialization.  Provides info on all the ranks within
 // the communicator.
 ncclResult_t ncclRasAddRanks(struct rasRankInit* ranks, int nranks) {
   struct rasNotification msg;
+  memset(&msg, '\0', sizeof(msg));
   msg.type = RAS_ADD_RANKS;
   msg.addRanks.ranks = ranks;
   msg.addRanks.nranks = nranks;
@@ -199,7 +211,7 @@ static ncclResult_t rasLocalNotify(const struct rasNotification* msg) {
 /////////////////////////////////////////////////////////////////////////////////
 
 // Handles asynchronous local notifications arriving from regular NCCL threads.
-static ncclResult_t rasLocalHandle() {
+static ncclResult_t rasLocalHandle(bool* terminate) {
   struct rasNotification msg;
 
   size_t done = 0;
@@ -212,9 +224,11 @@ static ncclResult_t rasLocalHandle() {
   }
 
   if (msg.type == RAS_ADD_RANKS) {
-    NCCLCHECK(rasLocalHandleAddRanks(msg.addRanks.ranks, msg.addRanks.nranks));
+    (void)rasLocalHandleAddRanks(msg.addRanks.ranks, msg.addRanks.nranks);
+    // Not great if the above fails, but it shouldn't be critical; better to keep going.
   } else if (msg.type == RAS_TERMINATE) {
-    rasLocalHandleTerminate();
+    INFO(NCCL_RAS, "RAS handling local termination request");
+    *terminate = true;
   } else {
     WARN("RAS received unknown notification type %d", msg.type);
     return ncclInternalError;
@@ -223,10 +237,35 @@ static ncclResult_t rasLocalHandle() {
   return ncclSuccess;
 }
 
-// Handles local RAS_TERMINATE notification.
-static void rasLocalHandleTerminate() {
-  INFO(NCCL_RAS, "RAS handling local termination request");
-  // For now we don't do anything.
+// Cleans up local RAS state, normally in response to a RAS_TERMINATE notification.
+static void rasThreadCleanup() {
+  rasClientSupportTerminate();
+  rasNetTerminate();
+  rasCollectivesTerminate();
+  rasPeersTerminate();
+
+  {
+    std::lock_guard<std::mutex> lock(rasInitMutex);
+    (void)close(rasNotificationPipe[1]);
+    (void)close(rasNotificationPipe[0]);
+    // rasClientListeningSocket is taken care of by rasClientSupportTerminate().
+    rasNotificationPipe[0] = rasNotificationPipe[1] = -1;
+    (void)ncclSocketClose(&rasNetListeningSocket);
+    rasInitRefCount = 0;
+    rasInitialized = false;
+  }
+
+  {
+    std::lock_guard<std::mutex> lock(ncclCommsMutex);
+    free(ncclComms);
+    ncclComms = nullptr;
+    nNcclComms = 0;
+    ncclCommsSorted = false;
+  }
+
+  free(rasPfds);
+  rasPfds = nullptr;
+  nRasPfds = 0;
 }
 
 
@@ -270,10 +309,10 @@ void rasConnEnqueueMsg(struct rasConnection* conn, struct rasMsg* msg, size_t ms
   else
     ncclIntruQueueEnqueue(&conn->sendQ, meta);
 
-  if (conn->sockIdx != -1) {
-    struct rasSocket* sock = rasSockets+conn->sockIdx;
-    if (sock->status == RAS_SOCK_READY || (sock->status == RAS_SOCK_HANDSHAKE && msg->type == RAS_MSG_CONNINIT)) {
-      rasPfds[sock->pfd].events |= POLLOUT;
+  if (conn->sock) {
+    if (conn->sock->status == RAS_SOCK_READY ||
+        (conn->sock->status == RAS_SOCK_HANDSHAKE && msg->type == RAS_MSG_CONNINIT)) {
+      rasPfds[conn->sock->pfd].events |= POLLOUT;
       ready = true;
     }
   }
@@ -283,31 +322,31 @@ void rasConnEnqueueMsg(struct rasConnection* conn, struct rasMsg* msg, size_t ms
          "(experiencingDelays %d, startRetryTime %.2fs, socket status %d)",
          msg->type, ncclSocketToString(&conn->addr, rasLine),
          conn->experiencingDelays, (conn->startRetryTime ? (clockNano()-conn->startRetryTime)/1e9 : 0.0),
-         (conn->sockIdx == -1 ? -1 : rasSockets[conn->sockIdx].status));
+         (conn->sock ? conn->sock->status : -1));
   }
 }
 
 // Attempts to send the queued RAS messages to another RAS thread.
 ncclResult_t rasConnSendMsg(struct rasConnection* conn, int* closed, bool* allSent) {
-  struct ncclSocket* sock = &rasSockets[conn->sockIdx].sock;
   struct rasMsgMeta* meta;
   *closed = 0;
   while ((meta = ncclIntruQueueHead(&conn->sendQ)) != nullptr) {
-    if (rasSockets[conn->sockIdx].status == RAS_SOCK_HANDSHAKE && meta->msg.type != RAS_MSG_CONNINIT) {
+    if (conn->sock->status == RAS_SOCK_HANDSHAKE && meta->msg.type != RAS_MSG_CONNINIT) {
       // We don't send anything beyond the handshake at this point.
       meta = nullptr;
       break;
     }
     if (meta->offset < sizeof(meta->length)) {
       // Send the length of the message.
-      NCCLCHECK(ncclSocketProgress(NCCL_SOCKET_SEND, sock, &meta->length, sizeof(meta->length), &meta->offset, closed));
+      NCCLCHECK(ncclSocketProgress(NCCL_SOCKET_SEND, &conn->sock->sock, &meta->length, sizeof(meta->length),
+                                   &meta->offset, closed));
       if (*closed)
         return ncclSuccess;
       if (meta->offset < sizeof(meta->length))
         break;
     }
     // Send the body of the message.
-    NCCLCHECK(ncclSocketProgress(NCCL_SOCKET_SEND, sock, ((char*)&meta->msg)-sizeof(meta->length),
+    NCCLCHECK(ncclSocketProgress(NCCL_SOCKET_SEND, &conn->sock->sock, ((char*)&meta->msg)-sizeof(meta->length),
                                  meta->length+sizeof(meta->length), &meta->offset, closed));
     if (*closed)
       return ncclSuccess;
@@ -377,7 +416,7 @@ ncclResult_t rasMsgHandle(struct rasMsg* msg, struct rasSocket* sock) {
 static ncclResult_t rasMsgHandleConnInit(const struct rasMsg* msg, struct rasSocket* sock) {
   ncclResult_t ret = ncclSuccess;
   struct rasConnection* conn = nullptr;
-  int connIdx, peerIdx;
+  int peerIdx;
   struct rasMsg* newMsg = nullptr;
   int newMsgLen;
   char line[SOCKET_NAME_MAXLEN+1];
@@ -406,19 +445,16 @@ static ncclResult_t rasMsgHandleConnInit(const struct rasMsg* msg, struct rasSoc
   }
 
   // Check for any existing connection with that RAS thread (could happen due to a network issue, or possibly a race).
-  connIdx = rasConnFind(&msg->connInit.listeningAddr);
-  if (connIdx != -1) {
-    conn = rasConns+connIdx;
-
+  conn = rasConnFind(&msg->connInit.listeningAddr);
+  if (conn) {
     INFO(NCCL_RAS,
          "RAS found a matching existing connection (sendQ %sempty, experiencingDelays %d, startRetryTime %.2fs)",
          (ncclIntruQueueEmpty(&conn->sendQ) ? "" : "not "),
          conn->experiencingDelays, (conn->startRetryTime ? (clockNano()-conn->startRetryTime)/1e9 : 0.0));
 
-    if (conn->sockIdx != -1) {
-      struct rasSocket* connSock = rasSockets+conn->sockIdx;
+    if (conn->sock) {
       INFO(NCCL_RAS, "RAS found an alternative existing socket (status %d, createTime %.2fs)",
-           connSock->status, (clockNano()-connSock->createTime)/1e9);
+           conn->sock->status, (clockNano()-conn->sock->createTime)/1e9);
       // In general we prefer to keep the newer connection, but "newer" can be a relative term: we may have
       // a race where both sides attempt to establish a connection at roughly the same time, so the other side's
       // incoming connection ends up looking newer than the locally-initiated one -- for *both* of them.
@@ -433,21 +469,19 @@ static ncclResult_t rasMsgHandleConnInit(const struct rasMsg* msg, struct rasSoc
         goto exit;
       } else {
         INFO(NCCL_RAS, "RAS keeping the new socket and terminating the existing one");
-        rasSocketTerminate(connSock);
+        rasSocketTerminate(conn->sock);
       }
     }
-  }
-  if (!conn) {
+  } else { // conn == nullptr
     NCCLCHECK(getNewConnEntry(&conn));
     memcpy(&conn->addr, &msg->connInit.listeningAddr, sizeof(conn->addr));
-    connIdx = conn - rasConns;
   }
 
   sock->status = RAS_SOCK_READY;
   // rasConnResume will reset any experiencingDelays, startRetryTime, etc.
 
-  conn->sockIdx = sock-rasSockets;
-  sock->connIdx = connIdx;
+  conn->sock = sock;
+  sock->conn = conn;
   memcpy(&sock->sock.addr, &msg->connInit.listeningAddr, sizeof(sock->sock.addr));
 
   // Make sure that the connection is part of the right links forming the RAS network.  At this point we only
@@ -456,8 +490,8 @@ static ncclResult_t rasMsgHandleConnInit(const struct rasMsg* msg, struct rasSoc
   // Note: it's possible for peerIdx to be -1 at this point if, due to races, the connInit arrives before
   // the peers update.
   if (peerIdx != -1) {
-    (void)rasLinkUpdateConn(&rasNextLink, connIdx, peerIdx);
-    (void)rasLinkUpdateConn(&rasPrevLink, connIdx, peerIdx);
+    (void)rasLinkConnUpdate(&rasNextLink, conn, peerIdx);
+    (void)rasLinkConnUpdate(&rasPrevLink, conn, peerIdx);
   }
 
   // Send a confirmation to the server that requested the connection (so that the resilience code can mark
@@ -504,12 +538,13 @@ static ncclResult_t rasMsgHandleConnInitAck(const struct rasMsg* msg, struct ras
 }
 
 // Handles the deadPeer broadcast.
-void rasMsgHandleBCDeadPeer(const struct rasCollRequest* req, bool* pDone) {
-  INFO(NCCL_RAS, "RAS handling deadPeer (addr %s)", ncclSocketToString(&req->deadPeer.addr, rasLine));
+void rasMsgHandleBCDeadPeer(struct rasCollRequest** pReq, size_t* pReqLen, bool* pDone) {
+  INFO(NCCL_RAS, "RAS handling deadPeer (addr %s)", ncclSocketToString(&(*pReq)->deadPeer.addr, rasLine));
 
-  if (!rasPeerIsDead(&req->deadPeer.addr)) {
-    rasConnDisconnect(&req->deadPeer.addr);
-    (void)rasPeerDeclareDead(&req->deadPeer.addr);
+  *pReqLen = rasCollDataLength(RAS_BC_DEADPEER);
+  if (!rasPeerIsDead(&(*pReq)->deadPeer.addr)) {
+    rasConnDisconnect(&(*pReq)->deadPeer.addr);
+    (void)rasPeerDeclareDead(&(*pReq)->deadPeer.addr);
     *pDone = false;
   } else {
     INFO(NCCL_RAS, "RAS already knew it was dead");
@@ -530,6 +565,7 @@ static ncclResult_t rasNetSendNack(struct rasSocket* sock) {
 
   INFO(NCCL_RAS, "RAS sending NACK to %s", ncclSocketToString(&sock->sock.addr, rasLine));
 
+  memset(&msg, '\0', sizeof(msg));
   msg.type = RAS_MSG_CONNINITACK;
   msg.connInitAck.nack = 1;
   offset = 0;
@@ -557,16 +593,16 @@ static void* rasThreadMain(void*) {
   INFO(NCCL_RAS, "RAS thread started");
 
   // Initialize the global pollfd with the file descriptors we already have (the pipe and the listening socket).
-  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, fail);
+  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, exit);
   rasPfds[pfd].fd = rasNotificationPipe[0];
   rasPfds[pfd].events = POLLIN;
 
-  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, fail);
-  NCCLCHECKGOTO(ncclSocketGetFd(&rasNetListeningSocket, &rasNetListeningSocketFd), ret, fail);
+  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, exit);
+  NCCLCHECKGOTO(ncclSocketGetFd(&rasNetListeningSocket, &rasNetListeningSocketFd), ret, exit);
   rasPfds[pfd].fd = rasNetListeningSocketFd;
   rasPfds[pfd].events = POLLIN;
 
-  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, fail);
+  NCCLCHECKGOTO(rasGetNewPollEntry(&pfd), ret, exit);
   rasPfds[pfd].fd = rasClientListeningSocket;
   rasPfds[pfd].events = POLLIN;
 
@@ -595,32 +631,37 @@ static void* rasThreadMain(void*) {
       if (rasPfds[pollIdx].revents) {
         nEvents--;
         if (rasPfds[pollIdx].fd == rasNotificationPipe[0]) {
-          (void)rasLocalHandle();
+          bool terminate = false;
+          NCCLCHECKGOTO(rasLocalHandle(&terminate), ret, exit);
+          if (terminate)
+            goto exit;
         } else if (rasPfds[pollIdx].fd == rasNetListeningSocketFd) {
           (void)rasNetAcceptNewSocket();
         } else if (rasPfds[pollIdx].fd == rasClientListeningSocket) {
           (void)rasClientAcceptNewSocket();
         } else {
           // Check if it's one of the RAS sockets.
-          int sockIdx;
-          for (sockIdx = 0; sockIdx < nRasSockets; sockIdx++) {
-            struct rasSocket* sock = rasSockets+sockIdx;
-            if (sock->status != RAS_SOCK_CLOSED && rasPfds[pollIdx].fd == sock->sock.fd) {
-              rasSockEventLoop(sockIdx, pollIdx);
+          struct rasSocket* sock;
+          for (sock = rasSocketsHead; sock;) {
+            struct rasSocket* sockNext = sock->next;
+            if (rasPfds[pollIdx].fd == sock->sock.fd) {
+              rasSockEventLoop(sock, pollIdx);
               break;
             }
-          } // for (sockIdx)
+            sock = sockNext;
+          } // for (sock)
 
-          if (sockIdx == nRasSockets) {
+          if (sock == nullptr) {
             // Try a client socket instead.
-            for (int clientIdx = 0; clientIdx < nRasClients; clientIdx++) {
-              struct rasClient* client = rasClients+clientIdx;
-              if (client->status != RAS_CLIENT_CLOSED && rasPfds[pollIdx].fd == client->sock) {
-                rasClientEventLoop(clientIdx, pollIdx);
+            for (struct rasClient* client = rasClientsHead; client;) {
+              struct rasClient* clientNext = client->next;
+              if (rasPfds[pollIdx].fd == client->sock) {
+                rasClientEventLoop(client, pollIdx);
                 break;
               }
-            } // for (clientIdx)
-          } // if (sockIdx == nRasSockets)
+              client = clientNext;
+            } // for (client)
+          } // if (sock == nullptr)
         } // dynamic fds
       } // if (revents)
     } // for (pollIdx)
@@ -636,14 +677,9 @@ static void* rasThreadMain(void*) {
     rasCollsHandleTimeouts(now, &nextWakeup);
   } // for (;;)
 
-fail:
-  WARN("fatal error - RAS thread terminating");
-  std::lock_guard<std::mutex> lock(rasInitMutex);
-  (void)close(rasNotificationPipe[1]);
-  (void)close(rasNotificationPipe[0]);
-  (void)close(rasClientListeningSocket);
-  (void)ncclSocketClose(&rasNetListeningSocket);
-  rasInitialized = false;
+exit:
+  rasThreadCleanup();
+  INFO(NCCL_RAS, "RAS thread terminating");
   return nullptr;
 }
 

src/ras/ras_internal.h

@@ -42,6 +42,14 @@ typedef enum {
   RAS_COLL_COMMS = 1002, // Collect data about all communicators.
 } rasCollectiveType;
 
+// Unique communicator identifier.  commHash by itself is definitely not guaranteed to be unique.
+// Combined with the two other hashes, the chance is much better...
+// All three fields are used for sorting.
+struct rasCommId {
+  uint64_t commHash;
+  uint64_t hostHash, pidHash; // These are the hashes of the *first* rank (comm->peerInfo[0]).
+};
+
 // Payload of a collective request message (RAS_MSG_COLLREQ).
 struct rasCollRequest {
   union ncclSocketAddress rootAddr;
@@ -56,6 +64,10 @@ struct rasCollRequest {
     struct {
     } conns;
     struct {
+      int nSkipMissingRanksComms; // Number of elements in the array below.
+      // Communicators for which we do *not* need the missingRanks data in the responses
+      // (see struct rasCollCommsMissingRank later).
+      struct rasCommId skipMissingRanksComms[0]; // Variable length, sorted.
     } comms;
   };
 };
@@ -69,8 +81,8 @@ struct rasCollResponse {
   int nPeers;
   int nData; // Size of data in bytes.
   union ncclSocketAddress peers[0]; // Variable length.
-  // The peersAddrs array is followed by:
-  //alignas(int64_t) char data[0]; // Variable length, collective-dependent.
+  // The peers array is followed by:
+  // alignas(int64_t) char data[0]; // Variable length, collective-dependent.
 };
 
 // Describes a peer NCCL process.  Every RAS thread keeps an (identical) array of them, one entry for each
@@ -80,6 +92,8 @@ struct rasPeerInfo {
   pid_t pid;
   uint64_t cudaDevs; // Bitmask.  This is for local devices so 64 bits is enough.
   uint64_t nvmlDevs; // Same, but not affected by CUDA_VISIBLE_DEVICES.
+  uint64_t hostHash, pidHash; // Taken from ncclComm, but with the commHash subtracted to make it
+                              // communicator-independent.
 };
 
 // Describes a RAS message.  Every message is preceded by a (32-bit) message length.  All data in the host
@@ -112,7 +126,7 @@ struct rasMsg {
       int nPeers;
       int nDeadPeers;
       struct rasPeerInfo peers[0]; // Variable length.
-      // The peers array is followed by the following:
+      // The peers array is followed by:
       //union ncclSocketAddress deadPeers[0]; // Variable length.
     } peersUpdate;
     struct {
@@ -218,6 +232,9 @@ struct rasMsgMeta {
 // Describes an ongoing collective RAS operation (apart from broadcasts, which don't need a response).
 // For every collective operation, each participating RAS thread will create its own.
 struct rasCollective {
+  struct rasCollective* next;
+  struct rasCollective* prev;
+
   union ncclSocketAddress rootAddr;
   uint64_t rootId;
 
@@ -227,15 +244,16 @@ struct rasCollective {
   bool timeoutWarned;
 
   int64_t startTime; // For timeout calculations.
-  int fromConnIdx; // The connection we received the request from.
+  struct rasConnection* fromConn; // The connection we received the request from.
 
-  int* fwdConns; // Indices of the connections we forwarded the request to; replaced by -1 as the responses arrive.
+  struct rasConnection** fwdConns; // Connections we forwarded the request to; replaced by nullptr's as the
+                                   // responses arrive.
   int nFwdSent; // Count of the above (local process only).
   int nFwdRecv; // Count of the responses received or timeouts (local process only).
 
   int nLegTimeouts; // Collective (from this process and the responses we received).
 
-  union ncclSocketAddress* peers; // Collective (from this process and the responses we received).
+  union ncclSocketAddress* peers; // Collective (from this process and the responses we received).  Unsorted.
   int nPeers;
 
   char* data; // Collective (from this process and the responses we received).
@@ -261,13 +279,14 @@ struct rasCollConns {
 struct rasCollComms {
   int nComms;
   struct comm {
-    uint64_t commHash;
-    int commNRanks;
-    int nRanks; // number of elements in the array below, *not* in the communicator.
+    struct rasCommId commId;
+    int commNRanks; // >= nRanks + nMissingRanks
+    int nRanks; // Number of elements in the ranks array below, *not* in the communicator.
+    int nMissingRanks; // Number of elements in the missingRanks array below.
     struct rank {
       int commRank;
       int peerIdx; // Index within rasCollective->peers, *not* rasPeers.
-      uint64_t collOpCount;
+      uint64_t collOpCounts[NCCL_NUM_FUNCTIONS];
       struct {
         ncclResult_t initState:4;
         ncclResult_t asyncError:4;
@@ -278,34 +297,47 @@ struct rasCollComms {
       char cudaDev;
       char nvmlDev;
     } ranks[0]; // Variable length. Sorted by commRank.  Optimized for 1 GPU/process.
-  } comms[0]; // Variable length. Sorted by commHash.
+    // The ranks array is followed by:
+    // struct rasCollCommsMissingRank missingRanks[0]; // Variable length.  Sorted by commRank.
+  } comms[0]; // Variable length.  Sorted by commId.
+};
+
+// Provides info about missing ranks.  An array of these structures can be part of struct rasCollComms above.
+// Because the arrays are of variable length, we can't describe them in C.  To ensure that adding
+// rasCollCommsMissingRank structures doesn't mess up the alignment, we explicitly request one.
+struct alignas(struct rasCollComms) rasCollCommsMissingRank {
+  int commRank;
+  union ncclSocketAddress addr;
+  // We don't need pid here as we can look it up in rasPeers via addr.
+  char cudaDev;
+  char nvmlDev;
 };
 
 // Holds data needed to keep track of a connection belonging to a RAS network link (either the primary one
 // or one of the fallbacks).
 struct rasLinkConn {
+  struct rasLinkConn* next;
   int peerIdx; // Index in the rasPeers array of the peer this entry describes.  Could be -1 (an entry initiated
                // by an as of yet unknown peer -- should be a temporary situation that resolves via peer updates).
-  int connIdx; // Index in the rasConns array of the connection to the above peer.  Could be -1 (a placeholder
-               // for a connection to be started by the remote peer).
+  struct rasConnection* conn; // The connection to the above peer.  Could be nullptr (a placeholder for a connection
+                              // to be started by the remote peer).
   bool external; // true if the entry exists only due to an external request (requested by a remote peer, most
                  // likely as part of fault recovery).  Such connections are kept as fallbacks even if there's a
                  // valid primary connection, in order to ensure that keep-alive messages are sent.
 };
 
 // Describes a link that forms the backbone of the RAS network.  Links focus on direction (previous/next in
-// case of 1-D topology) rather than a particular destination.  The are implemented using rasConnections, but
+// case of 1-D topology) rather than a particular destination.  They are implemented using rasConnections, but
 // they are persistent through the life of the RAS threads, whereas rasConnections can be terminated if the RAS
 // network is reconfigured or a peer dies.
 struct rasLink {
   int direction; // 1 for nextLink, -1 for prevLink.
 
-  // Index 0 is the primary connection; any additional ones are fallbacks (that get created if we are having
-  // problems with the primary connection).  The elements are de-facto ordered (highest-preference ones have
-  // the lowest indices).
+  // First element is the primary connection; any additional ones are fallbacks (that get created if we are having
+  // problems with the primary connection).  The highest-preference elements come first; the list is de-facto sorted
+  // by peerIdx, though peerIdx values can wrap around (given the ring/torus topology) and they can also be -1
+  // (the latter are stored at the end).
   struct rasLinkConn* conns;
-  int nConns;
-  int connsSize; // Array size; could be larger than nConns.
 
   // Keep track of a timeout in case we did not create a connection during the last peers update (because we expect
   // the peer on the other side to do so) but that peer failed to initiate.
@@ -315,15 +347,15 @@ struct rasLink {
 // Describes a connection to another peer on the RAS network.  It is meant to be more persistent than a volatile
 // socket (described by the rasSocket structure), which can be affected by transient network issues.
 struct rasConnection {
-  bool inUse;
+  struct rasConnection* next;
+  struct rasConnection* prev;
 
   union ncclSocketAddress addr;
 
-  // Index of the current rasSocket in the rasSockets array.  Note that multiple rasSocket entries may point back
+  // Pointer to the current rasSocket.  Note that multiple rasSocket entries may point back
   // to a single entry here, for sockets that are in the process of being terminated and re-established.
-  // We use indices, not pointers, because the arrays holding these structures can be re-alloced at run time.
-  // -1 if there is no such socket.
-  int sockIdx;
+  // nullptr if there is no such socket.
+  struct rasSocket* sock;
 
   // We keep the rasPeersHash of remote connections to minimize the number of needless exchanges.
   // There is a subtle difference in the meaning of lastSentPeersHash and lastRecvPeersHash.
@@ -371,16 +403,18 @@ typedef enum {
 
 // Describes a socket implementing communication between two peers.
 struct rasSocket {
+  struct rasSocket* next;
+  struct rasSocket* prev;
+
   struct ncclSocket sock;
 
   rasSocketStatus status;
 
   int pfd; // Index in the rasPfds array.
 
- // Index of the corresponding entry in the rasConns array.
-  // We use indices, not pointers, because the arrays holding these structures can be re-alloced at run time.
-  // -1 if there is no connection (normal condition on the accept side before the connInit message).
-  int connIdx;
+  // Pointer to the corresponding entry in the rasConns array.
+  // nullptr if there is no connection (a normal condition on the accept side before the connInit message).
+  struct rasConnection* conn;
 
   int64_t createTime;
   int64_t lastSendTime;
@@ -404,7 +438,10 @@ typedef enum {
 
 // Describes a RAS client.
 struct rasClient {
-  int sock;
+  struct rasClient* next;
+  struct rasClient* prev;
+
+  int sock; // File descriptor
 
   rasClientStatus status;
 
@@ -420,7 +457,7 @@ struct rasClient {
   int64_t timeout;
 
   // State stored during asynchronous operations such as collectives.
-  int collIdx; // Index to the onging rasCollective.
+  struct rasCollective* coll;
 };
 
 
@@ -440,31 +477,33 @@ void rasConnEnqueueMsg(struct rasConnection* conn, struct rasMsg* msg, size_t ms
 ncclResult_t rasConnSendMsg(struct rasConnection* conn, int* closed, bool* allSent);
 ncclResult_t rasMsgRecv(struct rasSocket* sock, struct rasMsg** msg, int* closed);
 ncclResult_t rasMsgHandle(struct rasMsg* msg, struct rasSocket* sock);
-void rasMsgHandleBCDeadPeer(const struct rasCollRequest* req, bool* pDone);
+void rasMsgHandleBCDeadPeer(struct rasCollRequest** pReq, size_t* pReqLen, bool* pDone);
 ncclResult_t rasGetNewPollEntry(int* index);
 
 
 // rasnet.cc
 extern struct rasLink rasNextLink, rasPrevLink;
-extern struct rasConnection* rasConns;
-extern int nRasConns;
-extern struct rasSocket *rasSockets;
-extern int nRasSockets;
+extern struct rasConnection* rasConnsHead;
+extern struct rasConnection* rasConnsTail;
+extern struct rasSocket *rasSocketsHead;
+extern struct rasSocket *rasSocketsTail;
 
 ncclResult_t getNewConnEntry(struct rasConnection** pConn);
-ncclResult_t rasConnCreate(const union ncclSocketAddress* addr, int* pConnIdx);
-int rasConnFind(const union ncclSocketAddress* addr);
+ncclResult_t rasConnCreate(const union ncclSocketAddress* addr, struct rasConnection** pConn);
+struct rasConnection* rasConnFind(const union ncclSocketAddress* addr);
 void rasConnsHandleTimeouts(int64_t now, int64_t* nextWakeup);
 void rasConnDisconnect(const union ncclSocketAddress* addr);
 ncclResult_t rasNetAcceptNewSocket();
 void rasSocksHandleTimeouts(int64_t now, int64_t* nextWakeup);
 void rasSocketTerminate(struct rasSocket* sock, bool finalize = false, uint64_t startRetryOffset = 0,
                         bool retry = true);
-void rasSockEventLoop(int sockIdx, int pollIdx);
+void rasSockEventLoop(struct rasSocket* sock, int pollIdx);
 void rasNetHandleTimeouts(int64_t now, int64_t* nextWakeup);
 ncclResult_t rasMsgHandleKeepAlive(const struct rasMsg* msg, struct rasSocket* sock);
-ncclResult_t rasLinkUpdateConn(struct rasLink* link, int connIdx, int peerIdx, bool external = false,
-                               bool insert = false, bool pretend = false, int* pLinkIdx = nullptr);
+ncclResult_t rasLinkAddFallback(struct rasLink* link, const struct rasConnection* conn);
+ncclResult_t rasLinkConnUpdate(struct rasLink* link, struct rasConnection* conn, int peerIdx);
+void rasNetTerminate();
+
 
 // peers.cc
 extern struct rasPeerInfo* rasPeers;
@@ -483,29 +522,35 @@ ncclResult_t rasPeerDeclareDead(const union ncclSocketAddress* addr);
 bool rasPeerIsDead(const union ncclSocketAddress* addr);
 int ncclSocketsCompare(const void* p1, const void* p2);
 bool ncclSocketsSameNode(const union ncclSocketAddress* a1, const union ncclSocketAddress* a2);
+void rasPeersTerminate();
 
 
 // collectives.cc
-extern struct rasCollective* rasCollectives;
+extern struct rasCollective* rasCollectivesHead;
+extern struct rasCollective* rasCollectivesTail;
 
 void rasCollReqInit(struct rasCollRequest* req);
-ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, size_t reqLen, bool* pAllDone = nullptr,
-                               int* pCollIdx = nullptr, int fromConnIdx = -1);
+ncclResult_t rasNetSendCollReq(const struct rasCollRequest* req, bool* pAllDone = nullptr,
+                               struct rasCollective** pColl = nullptr, struct rasConnection* fromConn = nullptr);
 ncclResult_t rasMsgHandleCollReq(struct rasMsg* msg, struct rasSocket* sock);
 ncclResult_t rasMsgHandleCollResp(struct rasMsg* msg, struct rasSocket* sock);
-void rasCollsPurgeConn(int connIdx);
+void rasCollsPurgeConn(struct rasConnection* conn);
 void rasCollFree(struct rasCollective* coll);
 void rasCollsHandleTimeouts(int64_t now, int64_t* nextWakeup);
+void rasCollectivesTerminate();
+
 
 // client_support.cc
 extern int rasClientListeningSocket;
-extern struct rasClient* rasClients;
-extern int nRasClients;
+extern struct rasClient* rasClientsHead;
+extern struct rasClient* rasClientsTail;
+
 ncclResult_t rasClientInitSocket();
 ncclResult_t rasClientAcceptNewSocket();
 ncclResult_t rasClientResume(struct rasCollective* coll);
-void rasClientEventLoop(int clientIdx, int pollIdx);
+void rasClientEventLoop(struct rasClient* client, int pollIdx);
 const char* rasGpuDevsToString(uint64_t cudaDevs, uint64_t nvmlDevs, char* buf, size_t size);
+void rasClientSupportTerminate();
 
 #endif // !NCCL_RAS_CLIENT