fix(rccl): correct IB-CAST network telemetry counters on bnxt_re (AICOMNET-432)

projects/rccl/docs/how-to/network-telemetry.md

@@ -47,7 +47,8 @@ fatal.
 * If the JSON cannot be written at exit, the reason and the path are printed.
 * If a QP or channel cannot be given a stats slot, it is counted in
   `num_qp_untracked` rather than dropped silently, and that count is in the
-  JSON next to the counters it is missing from.
+  JSON next to the counters it is missing from. The first such shortfall is
+  also printed on stderr.
 * A hardware counter that cannot be read is emitted as `-1`, never as `0`, so
   "not available" is never mistaken for "no events".
 
@@ -67,7 +68,7 @@ After the run, inspect any `rccl_telemetry_*.json` in the output directory.
 ```
 {
   "version", "host_name", "process_name", "process_id",
-  "start_time", "end_time", "transport",
+  "start_time", "end_time", "transport",   // "none" if no device registered
   "latency_sample_interval",   // only when RCCL_TELEMETRY_LATENCY_SAMPLE > 1
   "devices": [
     {
@@ -222,7 +223,7 @@ channel (aggregated) and per QP.
 |---|---|
 | `num_wqe_sent` | Send WQEs posted on this QP. |
 | `num_recv_wqe` | Receive WQEs posted (`ibv_post_recv`). |
-| `num_wqe_rcvd` | Completions drained from the CQ, send and receive alike. |
+| `num_wqe_rcvd` | Send, receive and CTS completions drained from the CQ. Not counted: error completions (see `num_cq_errors`), GPU flush completions, and completions resiliency matches to no active request. |
 | `num_wqe_completed` | Completions that matched a tracked posting, i.e. those with a recorded post timestamp so latency is computable. |
 | `num_cts_sent` | CTS (clear-to-send) messages posted. |
 | `num_cts_sent_signalled` / `num_cts_sent_unsignalled` | Split of `num_cts_sent` by whether the WR was signalled. Their sum equals `num_cts_sent`. |
@@ -251,7 +252,7 @@ Three counters look similar but mean different things, and the difference is the
 usual source of confusion:
 
 - `num_recv_wqe` counts receive WQEs **posted**, not completed.
-- `num_wqe_rcvd` counts **all** completions drained from the CQ (send and recv).
+- `num_wqe_rcvd` counts send, receive and CTS completions drained from the CQ.
 - `num_wqe_completed` counts only the subset of those completions that were
   matched to a tracked posting.
 
@@ -271,10 +272,10 @@ whole run: once on first use (baseline) and once at exit (final). A rank
 registers every NIC it can enumerate but normally drives only one or two, and
 each read is a direct `ioctl(SIOCETHTOOL)` (using `<linux/ethtool.h>` and
 `<net/if.h>`, no shell) plus a few IB sysfs reads, so devices the rank never
-connects over are not read at all and report `-1`. The periodic `hw_samples`
-time series is produced only when `RCCL_TELEMETRY_SAMPLE_MS > 0`, and even then
-it reads only IB sysfs files. The data path itself never reads sysfs or the NIC
-counters.
+connects over are not read at all and are left out of the JSON. The periodic
+`hw_samples` time series is produced only when `RCCL_TELEMETRY_SAMPLE_MS > 0`,
+and even then it reads only IB sysfs files. The data path itself never reads
+sysfs or the NIC counters.
 
 ## Known limitations
 
@@ -283,3 +284,18 @@ counters.
   completion, so `rx_bytes` cannot be measured and reads 0. This is expected,
   not a lost count. When the receiver does take completions,
   `IBV_WR_RDMA_WRITE_WITH_IMM` is used and `rx_bytes` is populated.
+- With pre-posted receive WQEs (`NCCL_IB_PREPOST_RECEIVE_WORK_REQUESTS`,
+  resiliency or OOO RQ), receive completions carry no post timestamp: they count
+  in `num_wqe_rcvd` but not in `num_wqe_completed` or the latency histogram.
+- thor2 (`bnxt_re`): counter names differ across driver versions, so the table
+  falls back to older names. The IB sysfs byte/packet counters can update
+  late, so `delta_*` is only reliable over the whole run. They are treated as
+  48 bits wide: one wrap during the run is accounted for, and a counter reset
+  reads as a wrap. Some firmware reports `pfc_*_pause_us` as 0 while PFC
+  frames are counted.
+- Any other HW counter that goes down during the run (a wrap or a reset)
+  reads -1.
+- With QP sharing, each communicator has its own QP slots, so a shared QP is
+  counted once per communicator using it.
+- In `hw_samples`, a column the NIC cannot read falls back to a related counter:
+  `local_ack_timeout_err` reports `to_retransmits` on ainic and thor2.

projects/rccl/src/include/net_telemetry.h

@@ -1065,6 +1065,15 @@ static inline void rcclTelemetryWqeSize(int devIdx, uint64_t bytes) {
  */
 int rcclTelemetrySetupChannel(int devIdx, int chIdx, int numQps, int* numSlots);
 
+/**
+ * Set up the n QPs of one device on a channel and resolve their handles.
+ * Entries with the same key (e.g. the ibv_qp a QP-sharing secondary borrows)
+ * share one slot, so posts and completions on that QP meet in the same place.
+ * handles[q] is NULL for an untracked entry. Returns the slots granted.
+ */
+int rcclTelemetrySetupQps(int devIdx, int chIdx, const void* const* keys, const int* isDataQp, int n,
+                          RcclQpStats** handles);
+
 /**
  * Mark a device's channel ids as not real NCCL channel ids.
  *
@@ -1102,6 +1111,51 @@ static inline void rcclTelemetrySetQpRole(int devIdx, int chIdx, int qpIdx, int
     __atomic_fetch_add(&ch->num_cts_qp, 1, __ATOMIC_RELAXED);
 }
 
+/* ------------------------------------------------------------------ */
+/* Internal to net_telemetry.cc, declared here for the unit test.      */
+/* Transports must not use anything below.                             */
+/* ------------------------------------------------------------------ */
+
+/* Longest `ethtool -S` statistic name we ever ask for. The kernel bounds a
+ * statistic name by ETH_GSTRING_LEN (32), and the per-priority names we format
+ * stay well inside that, so this has room to spare. */
+#define RCCL_TEL_ETHTOOL_KEY_MAX 64
+
+typedef struct {
+  char     key[RCCL_TEL_ETHTOOL_KEY_MAX];
+  int64_t* target;
+} RcclEthtoolWantedEx;
+
+/* First match in `wanted` order wins, so a primary key queued before its
+ * fallback is preferred. */
+void rcclTelemetryEthtoolResolve(const char* names, size_t nameStride, const uint64_t* values, unsigned n,
+                                 RcclEthtoolWantedEx* wanted, int nwanted);
+
+/* Root of the sysfs/debugfs tree telemetry reads, "/sys" outside tests. */
+extern const char* rcclTelemetrySysRoot;
+
+/* Delta of a HW counter vs. its baseline; -1 if either is unavailable. A
+ * counter that went down is one wrap at wrapBits (e.g. 48 for bnxt_re
+ * rx_bytes), or unknown (-1) when wrapBits is 0 or it cannot be a wrap. */
+int64_t rcclTelemetryCounterDelta(int64_t cur, int64_t init, int wrapBits);
+
+/* Samples that retry a sampled column's primary row before it locks to the
+ * alternate. */
+#define RCCL_TEL_SAMPLE_PRIMARY_TRIES 10
+
+/* Which table row an hw_samples column reads on one device. With both a
+ * primary and an alternate row, the primary is tried on each sample and the
+ * choice locks on its first successful read, else to the alternate after
+ * RCCL_TEL_SAMPLE_PRIMARY_TRIES samples. */
+typedef struct {
+  int nameIdx, altIdx;
+  int locked; /* row in use once settled, -2 while still trying the primary */
+  int tries;
+} RcclTelSampledCol;
+
+void rcclTelemetrySampledColInit(int devIdx, int sampled, RcclTelSampledCol* col);
+int64_t rcclTelemetrySampledColRead(int devIdx, RcclTelSampledCol* col);
+
 #ifdef __cplusplus
 }
 #endif

projects/rccl/src/transport/net_ib_cast/connect.cc

@@ -1289,6 +1289,7 @@ static ncclResult_t IbCastQpSharingSenderSetup(
       // with QP sharing; the non-offload signaling path uses devIndex instead.
       comm->base.qps[q].qp = slot->qp;
       comm->base.qps[q].devIndex = slot->devIndex;
+      comm->base.qps[q].isDataQp = true;
       comm->base.activeQps[q] = &comm->base.qps[q];
 
       // Populate metadata with shared QP info
@@ -1368,6 +1369,29 @@ static ncclResult_t IbCastQpSharingSenderRegisterPrimary(
   return ncclSuccess;
 }
 
+// Telemetry slots for every QP of a comm, one per ibv_qp: a QP-sharing
+// secondary can map several qps[] entries to one shared QP. Untracked QPs keep
+// telQpStats NULL (IbCastQpCreate() or the zeroed comm).
+static void IbCastTelSetupQps(struct ncclIbNetCommBase* base, int channelId, bool channelIdKnown,
+                              const int* isDataQp) {
+  for (int i = 0; i < base->vProps.ndevs; i++) {
+    int ibDevN = base->vProps.devs[i];
+    const void* keys[NCCL_IB_MAX_QPS];
+    int roles[NCCL_IB_MAX_QPS], idx[NCCL_IB_MAX_QPS];
+    RcclQpStats* handles[NCCL_IB_MAX_QPS];
+    int n = 0;
+    for (int q = 0; q < base->nqps; q++) {
+      if (base->qps[q].devIndex != i) continue;
+      keys[n] = base->qps[q].qp;
+      roles[n] = isDataQp[q];
+      idx[n++] = q;
+    }
+    rcclTelemetrySetupQps(ibDevN, channelId, keys, roles, n, handles);
+    if (!channelIdKnown) rcclTelemetryMarkChannelsUnknown(ibDevN);
+    for (int k = 0; k < n; k++) base->qps[idx[k]].telQpStats = handles[k];
+  }
+}
+
 ncclResult_t IbCastConnectImpl(void* ctx, int dev, void* opaqueHandle, void** sendComm,
                                ncclNetDeviceHandle_t** sendDevComm, int envTrafficClass) {
   ncclResult_t ret = ncclSuccess;
@@ -1543,36 +1567,21 @@ ncclResult_t IbCastConnectImpl(void* ctx, int dev, void* opaqueHandle, void** se
 
   if (sharingRole != QP_SHARING_SECONDARY) {
     NCCLCHECKGOTO(IbCastSenderQpsCreate(comm, &meta, channelId, depthMult), ret, fail);
-    comm->telChId = channelId;
-    comm->telChStats = NULL;
-
-    // Telemetry-only: skip when off. IbCastQpCreate() left every telQpStats NULL,
-    // the untracked value the hooks expect.
-    if (rcclTelemetryOn()) {
-      for (int i = 0; i < comm->base.vProps.ndevs; i++) {
-        int ibDevN = comm->base.vProps.devs[i];
-        int numQpsForDev = 0;
-        for (int q = 0; q < comm->base.nqps; q++)
-          if (comm->base.qps[q].devIndex == i) numQpsForDev++;
-        int numSlots = 0;
-        int startSlot = rcclTelemetrySetupChannel(ibDevN, channelId, numQpsForDev, &numSlots);
-        if (!telChannelIdKnown) rcclTelemetryMarkChannelsUnknown(ibDevN);
-        int slotOffset = 0;
-        for (int q = 0; q < comm->base.nqps && slotOffset < numSlots; q++) {
-          if (comm->base.qps[q].devIndex != i) continue;
-          // QPs past the granted slots keep telQpStats == NULL (untracked).
-          int telSlot = startSlot + slotOffset++;
-          rcclTelemetrySetQpRole(ibDevN, channelId, telSlot, comm->base.qps[q].isDataQp);
-          comm->base.qps[q].telQpStats = rcclTelemetryResolveQp(ibDevN, channelId, telSlot);
-        }
-      }
-      // Request completions are charged to the comm's first device.
-      comm->telChStats = rcclTelemetryResolveChannel(comm->base.vProps.devs[0], channelId);
-    }
+  }
+  comm->telChId = channelId;
+  comm->telChStats = NULL;
 
-    if (sharingRole == QP_SHARING_PRIMARY) {
-      NCCLCHECKGOTO(IbCastQpSharingSenderRegisterPrimary(comm, handle, &remoteVProps), ret, fail);
-    }
+  // Telemetry-only: skip when off.
+  if (rcclTelemetryOn()) {
+    int isDataQp[NCCL_IB_MAX_QPS];
+    for (int q = 0; q < comm->base.nqps; q++) isDataQp[q] = comm->base.qps[q].isDataQp;
+    IbCastTelSetupQps(&comm->base, channelId, telChannelIdKnown, isDataQp);
+    // Request completions are charged to the comm's first device.
+    comm->telChStats = rcclTelemetryResolveChannel(comm->base.vProps.devs[0], channelId);
+  }
+
+  if (sharingRole == QP_SHARING_PRIMARY) {
+    NCCLCHECKGOTO(IbCastQpSharingSenderRegisterPrimary(comm, handle, &remoteVProps), ret, fail);
   }
 
   // Populate remaining QP sharing metadata
@@ -2041,6 +2050,7 @@ ncclResult_t IbCastPostReceiveWorkRequestsOnQp(struct ncclIbRecvComm* recvComm,
        dataQp->qp->qp_num, recvComm);
   for (int j = 0; j < nRecvWorkRequestsPerQp; j++) {
     NCCLCHECK(IbCastPostRecvWorkRequest(dataQp->qp, &recvComm->ibRecvWorkRequest));
+    rcclTelemetryQpRecvPosted(dataQp->telQpStats);
   }
   return ncclSuccess;
 }
@@ -2525,36 +2535,23 @@ ncclResult_t IbCastAccept(void* listenComm, void** recvComm, ncclNetDeviceHandle
   meta.sharedGroupIdx = rComm->base.qpSharing.groupIdx;
   meta.commId = rComm->base.qpSharing.netIbCommId;
 
-  if (rComm->prepostReceiveWorkRequests) {
-    NCCLCHECKGOTO(IbCastReceiverPrePostReceiveWorkRequests(rComm), ret, fail);
-  }
-
   rComm->telChId = channelId;
   rComm->telChStats = NULL;
 
   // See IbCastConnect(): telemetry-only work, skipped when telemetry is off.
   if (rcclTelemetryOn()) {
-    for (int i = 0; i < rComm->base.vProps.ndevs; i++) {
-      int telIbDevN = rComm->base.vProps.devs[i];
-      int numQpsForDev = 0;
-      for (int q = 0; q < rComm->base.nqps; q++)
-        if (rComm->base.qps[q].devIndex == i) numQpsForDev++;
-      int numSlots = 0;
-      int startSlot = rcclTelemetrySetupChannel(telIbDevN, channelId, numQpsForDev, &numSlots);
-      if (!telChannelIdKnown) rcclTelemetryMarkChannelsUnknown(telIbDevN);
-      int slotOffset = 0;
-      for (int q = 0; q < rComm->base.nqps && slotOffset < numSlots; q++) {
-        if (rComm->base.qps[q].devIndex != i) continue;
-        // QPs past the granted slots keep telQpStats == NULL (untracked).
-        int telSlot = startSlot + slotOffset++;
-        // isDataQp is always false on receiver QPs; classify by CTS role.
-        rcclTelemetrySetQpRole(telIbDevN, channelId, telSlot, !IbCastRecvCommIsCtsQp(rComm, q));
-        rComm->base.qps[q].telQpStats = rcclTelemetryResolveQp(telIbDevN, channelId, telSlot);
-      }
-    }
+    // isDataQp is always false on receiver QPs; classify by CTS role.
+    int isDataQp[NCCL_IB_MAX_QPS];
+    for (int q = 0; q < rComm->base.nqps; q++) isDataQp[q] = !IbCastRecvCommIsCtsQp(rComm, q);
+    IbCastTelSetupQps(&rComm->base, channelId, telChannelIdKnown, isDataQp);
     rComm->telChStats = rcclTelemetryResolveChannel(rComm->base.vProps.devs[0], channelId);
   }
 
+  // After telemetry setup so the pre-posted WQEs are counted.
+  if (rComm->prepostReceiveWorkRequests) {
+    NCCLCHECKGOTO(IbCastReceiverPrePostReceiveWorkRequests(rComm), ret, fail);
+  }
+
   // Store the remote CTS FIFO info provided by the remote peer
   rComm->remCtsFifo.addr = remMeta.addr;
   for (int i = 0; i < rComm->base.nRemDevs; i++) {

projects/rccl/src/transport/net_ib_cast/p2p.cc

@@ -248,6 +248,7 @@ ncclResult_t IbCastMultiSend(struct ncclIbSendComm* comm, int slot, int nqps, in
         comm->wrs[r].sg_list = NULL;
         comm->wrs[r].num_sge = 0;
         length = 0;
+        rcclTelemetryWqeSize(telDevIdx, 0);
       } else {
         // Select proper lkey
         comm->sges[r].lkey = reqs[r]->send.lkeys[devIndex];
@@ -526,7 +527,14 @@ ncclResult_t IbCastIsend(void* sendComm, void* data, size_t size, int tag, void*
     TIME_START(0);
     NCCLCHECK(IbCastMultiSend(comm, slot, nqps, startQpIndex, wrrSched, useWriteOp));
 
-    rcclTelemetryBytes(comm->base.vProps.devs[0], 1, (uint64_t)size);
+    // Reached once per multi-send, when sendReqsCnt[slot] == nreqs, so every
+    // reqs[i] is set; fifoHead then moves on. Resiliency replays call
+    // IbCastMultiSend directly and are not counted again.
+    if (rcclTelemetryOn()) {
+      uint64_t sentBytes = 0;
+      for (int i = 0; i < nreqs; i++) sentBytes += reqs[i]->send.size;
+      rcclTelemetryBytes(comm->base.vProps.devs[0], 1, sentBytes);
+    }
 
     comm->base.fifoHead++;
     TIME_STOP(0);
@@ -1101,6 +1109,7 @@ static ncclResult_t IbCastCompletionEventByOrder(struct ncclIbNetCommBase* commB
       req->events[devIndex]--;
     } else if (wc->opcode == IBV_WC_RDMA_WRITE) {
       TRACE(NCCL_NET, "NET/IB: %s: Got completion for a CTS (devIndex=%d, qp_num=%u)", __func__, devIndex, wc->qp_num);
+      IbCastTelemetryWqeComplete(commBase, wc, devIndex, 0);
       req->events[devIndex]--;
       return ncclSuccess;
     } else {
@@ -1212,6 +1221,7 @@ static inline ncclResult_t IbCastCompletionEventProcess(struct ncclIbNetCommBase
         NCCLCHECK(IbCastCommBaseGetQpByQpNum(commBase, devIndex, wc->qp_num, &qp, &qpIndex));
         req->recv.cmplsRecords->completions[qpIndex] = 1;
         NCCLCHECK(IbCastPostRecvWorkRequest(qp->qp, &recvComm->ibRecvWorkRequest));
+        rcclTelemetryQpRecvPosted(qp->telQpStats);
       } else {
         // In the prepost path wr_id is UINT64_MAX (sentinel); only decrement rxPosts
         // in the non-prepost path where wr_id is a valid slot index.
@@ -1253,6 +1263,7 @@ static inline ncclResult_t IbCastCompletionEventProcess(struct ncclIbNetCommBase
              be32toh(wc->imm_data));
         return ncclSuccess;
       }
+      IbCastTelemetryWqeComplete(commBase, wc, devIndex, 0);
     } else {
       WARN("NET/IB: %s: Unknown completion (req=%p, comm=%p, id=%ld, devIndex=%d, req->type=%s, wc.wr_id=0x%lx, "
            "wc.opcode=%s(%d), wc.qp_num=%u, wc.imm_data=0x%x)",