intercept_timer.cpp

/*
 * intercept_timer — an example libfab-intercepter consumer that measures the
 * latency of each initiator-side data-path op (issue ... CQ completion) and
 * attributes it to the directed NIC->NIC link it ran on.
 *
 * Timed ops: fi_send, fi_tsend, fi_tsendmsg, fi_tsenddata, fi_write, fi_writemsg,
 * the whole fi_read family (fi_read / fi_readv / fi_readmsg), and fi_trecv
 * (receive-side; see "Receiver-side timing" below). The tagged sendmsg/senddata
 * variants are how Cray MPICH sends; the fi_read family is how it services
 * MPI_Get. (There is no fi_readdata op -- libfabric carries remote CQ data only
 * on the write side; a read that wants it goes through fi_readmsg with
 * FI_REMOTE_CQ_DATA, which is already covered here.)
 *
 * How it works
 * ------------
 *   - On each issue (the send/write/read shims): claim a slot in a fixed ring,
 *     stamp the start time, the op kind, the transfer size, the issuing
 *     endpoint and the peer fi_addr_t, stash the application's original
 *     context, then forward the op with the context redirected at our slot so
 *     we can find it again at completion. For the plain ops `context` is a
 *     parameter we repoint directly; for the `msg` ops it lives in a const
 *     struct, so we forward a copy.
 *   - on_ep_bind: when the app binds an address vector to an endpoint, remember
 *     ep->av so we can turn an fi_addr_t (an AV-table index for cxi) back into a
 *     human-readable NIC address at report time.
 *   - cq_open: remember each CQ's entry format, so we know where the op_context
 *     lives in the completion structs cq_read fills in.
 *   - cq_read: for every completion whose op_context points back into our ring,
 *     stamp the end time, fold the latency into the per-link / per-(kind,size)
 *     stats, and restore the application's original context before it sees the
 *     completion. The first completion on a link resolves its local/remote NIC
 *     address strings (the endpoint and AV are still alive at that point).
 *   - cq_readerr: an op that failed completes here instead; we free its slot,
 *     restore the context and count it as an error rather than timing it.
 *
 * Output: each rank writes its own report file at exit so the per-rank reports
 * never interleave. Path is
 *   ${INTERCEPT_TIMER_DIR:-.}/intercept_timer.<host>.r<rank>.p<pid>.txt
 * (rank taken from SLURM_PROCID / OMPI_COMM_WORLD_RANK / PMIX_RANK / PMI_RANK /
 * RANK). A single line to stderr points at each file. Set
 * INTERCEPT_TIMER_DIR=stderr (or "-") to write straight to stderr instead
 * (which will interleave across ranks). Each report contains:
 *   - Per NIC (local endpoint): a concurrency-aware achieved bandwidth =
 *     completed bytes / the wall-clock time the NIC had >=1 op in flight (the
 *     union of all its in-flight intervals). This is the number to read for
 *     "am I saturating the link"; it counts overlapping ops once, not summed.
 *     Attribution is per NIC, not per link, because when several peers share a
 *     NIC their intervals overlap and can't be split per link.
 *   - Per link (local NIC -> remote NIC): every distinct (op, message size)
 *     with its op count, latency (avg/min/max) and an implied per-op bandwidth,
 *     so you can see how close an individual large transfer gets to peak.
 *
 * Measurement caveat: "latency" is issue -> the moment we observe the
 * completion in cq_read, so it includes however long the entry sat in the CQ
 * before the app polled it. It is a faithful op turnaround time, not a
 * hardware-only number; the per-link implied bandwidth is per-op (one op at a
 * time), while the per-NIC achieved bandwidth is the concurrency-aware figure.
 *
 * Receiver-side timing (fi_trecv): post -> completion. Unlike the send-side
 * ops this includes the wait for the peer to send at all, so its avg/min/max
 * are NOT a network latency. The number to read is the [imm N%] tag on each
 * fi_trecv row: the fraction of recvs that completed almost immediately
 * (< 50 us) after being posted, i.e. the message had ALREADY arrived and was
 * sitting in the provider's unexpected/overflow queue -- direct evidence the
 * receiver posted late. trecvs are deliberately excluded from the per-NIC
 * busy/achieved aggregate (their in-flight time is mostly idle wait), and the
 * row's size is the actual bytes received (from the CQ entry), not the posted
 * buffer size, when the CQ format carries a length.
 *
 * CQ poll gaps: every fi_cq_read/fi_cq_readfrom invocation is timestamped
 * (one steady_clock read per poll) and the gap to the SAME CQ's previous poll
 * is folded into a per-CQ histogram, reported as the "CQ#n" lines. A progress
 * thread that is descheduled or stuck shows up as gaps in the ms+ buckets and
 * a large max (with its time offset since the first fabric op, to correlate
 * with benchmark phases). Legitimate idle phases (between benchmark sizes,
 * init/teardown) also produce large gaps -- interpret counts, not totals, and
 * compare nodes against nodes. Set INTERCEPT_TIMER_CQGAP=0 to disable the
 * per-poll timestamping entirely.
 *
 * Per-op CSV mode (opt-in): logs EVERY completed op -- start, end, duration, op
 * kind, bytes, and source->target NIC -- as a CSV at exit
 * (intercept_timer_ops.<host>.r<rank>.p<pid>.csv in INTERCEPT_TIMER_DIR). The
 * dumped op kinds are exactly the hooked initiator-side ops plus tagged recv:
 * fi_send, fi_tsend, fi_tsendmsg, fi_tsenddata, fi_write, fi_writemsg, fi_read,
 * fi_readv, fi_readmsg, fi_trecv. Two
 * ways to scope it (set one):
 *   - INTERCEPT_TIMER_OPS_RANKS=<list/range> (e.g. "0,8,16" or "0-7,512"):
 *     dump every op on those ranks.
 *   - INTERCEPT_TIMER_OPS_NIC=<nic substring> (e.g. "0x01ad0407", taken from a
 *     NIC line of the aggregate report): dump on ALL ranks, but only ops
 *     touching that NIC -- sends/writes/reads aimed at it (remote NIC) and
 *     recvs posted by it (local NIC). The match is decided once per link, at
 *     address-resolution time, so the hot path stays a cached bool. If both
 *     vars are set, the NIC filter wins.
 * Either way, only ops of at least INTERCEPT_TIMER_OPS_MIN_BYTES bytes are
 * logged (default 65536, which keeps RCCL's 64/128 KiB data chunks and drops
 * the small control/signaling messages that dominate op counts at scale; set
 * 131072 for 128 KiB chunks only, 0 to log everything). Ops that complete in
 * error are always logged regardless of size. The aggregate report is
 * unaffected -- the filter scopes only the per-op CSV.
 * Samples are buffered in a RAM buffer malloc'd once at startup for active
 * ranks (INTERCEPT_TIMER_OPS_MB, default 1024 = 1 GiB/rank; ~33M ops; virtual
 * until written, so ranks that match nothing pay ~no physical memory) and
 * appended lock-free (one atomic bump + struct store), formatted to CSV only at
 * exit. If the buffer fills, further ops are dropped and counted (raise the MB
 * knob).
 *
 * NIC hardware-counter time series (on by default): a background thread samples
 * the Cassini per-NIC counters every INTERCEPT_TIMER_NIC_MS ms (default 100) and
 * writes them as a CSV at exit:
 *   ${INTERCEPT_TIMER_DIR:-.}/intercept_timer_nic.<host>.r<rank>.p<pid>.csv
 * One row per sample: t_ns (ns since this rank's anchor -- the SAME anchor as the
 * per-op CSV, so they line up), wall_ns (CLOCK_REALTIME, for cross-rank
 * alignment), then one column per counter. Two sources are read: the sysfs
 * telemetry under /sys/class/cxi/<dev>/device/telemetry/ (e.g. hni_sts_*_octets,
 * pct_spt_timeouts, hni_pause_sent, ixe_pool_ecn_pkts_N) and the cxi_rh retry
 * handler stats under /run/cxi/<dev>/ (e.g. nacks, parked_nids, pkts_cancelled_o),
 * the latter as "rh:<name>" columns. The default selection favors the telemetry:
 * the retry-handler timeouts/NACKs/occupancy all have pct_* twins that advance
 * with the hardware instead of lagging the FUSE daemon that exports the rh: side,
 * so by default only the rh stats WITHOUT a hardware twin (NID/switch parking,
 * cancellations) are sampled; pass INTERCEPT_TIMER_NIC_COUNTERS=rh: for the full
 * retry-handler set. By default each rank samples the NIC it drives
 * (cxi$SLURM_LOCALID). The thread starts on the first fabric activity, not at
 * load, so forked non-fabric children don't each emit a file. Knobs:
 * INTERCEPT_TIMER_NIC=0 (off), INTERCEPT_TIMER_NIC_MS, INTERCEPT_TIMER_NIC_DEV
 * (cxi list | all), INTERCEPT_TIMER_NIC_COUNTERS (all | rh: | <prefix>* | names;
 * default = a wide curated telemetry set + a curated rh subset),
 * INTERCEPT_TIMER_NIC_MAX_MB (default 256).
 *
 * Build (linked into your own LD_PRELOAD .so, see README.md):
 *   g++ -std=c++17 -fPIC -pthread -shared -I$LIBFABRIC_PREFIX/include \
 *       intercept_timer.cpp libfabintercept.a -o libtimer_preload.so -ldl
 */
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>

#include <algorithm>
#include <atomic>
#include <cctype>
#include <chrono>
#include <condition_variable>
#include <ctime>
#include <limits>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <unordered_set>
#include <vector>

#include <dirent.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>  // gethostname, getpid

#include "libfabintercept.h"
#include "rdma/fabric.h"
#include "rdma/fi_cm.h"  // fi_getname

using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;

namespace {

enum op_kind {
  OP_SEND,
  OP_TSEND,
  OP_WRITE,
  OP_WRITEMSG,
  OP_READ,
  OP_READMSG,
  OP_TRECV,
  // Appended (not inserted) so existing op_kind numeric values stay stable. The
  // tagged-send variants Cray MPICH uses for its rendezvous/large path: bare
  // fi_tsend (also hooked) never showed up, so its sends go through these.
  OP_TSENDMSG,
  OP_TSENDDATA,
  // Completes the fi_read family: the scatter/gather read. Cray MPICH issues it
  // when an MPI_Get lands on a non-contiguous origin buffer.
  OP_READV,
  OP_KIND_COUNT
};

const char* op_name(op_kind k) {
  switch (k) {
    case OP_SEND:      return "fi_send";
    case OP_TSEND:     return "fi_tsend";
    case OP_WRITE:     return "fi_write";
    case OP_WRITEMSG:  return "fi_writemsg";
    case OP_READ:      return "fi_read";
    case OP_READMSG:   return "fi_readmsg";
    case OP_TRECV:     return "fi_trecv";
    case OP_TSENDMSG:  return "fi_tsendmsg";
    case OP_TSENDDATA: return "fi_tsenddata";
    case OP_READV:     return "fi_readv";
    default:           return "?";
  }
}

// Best-effort rank id for naming the per-rank output file. We have no MPI here,
// so read whichever launcher set its rank in the environment (srun, Open MPI,
// PMIx/PMI, torchrun); fall back to "NA" if none is present.
std::string detect_rank() {
  static const char* const keys[] = {"SLURM_PROCID", "OMPI_COMM_WORLD_RANK",
                                      "PMIX_RANK", "PMI_RANK", "RANK"};
  for (const char* k : keys)
    if (const char* v = getenv(k)) return v;
  return "NA";
}

// ---- Per-op CSV logging (opt-in) ----------------------------------------
// One fixed-size sample per completed op, appended to a preallocated RAM buffer
// when this rank is selected by INTERCEPT_TIMER_OPS_RANKS and written out as CSV
// at exit. This is ADDITIVE to the aggregate report and entirely off otherwise.
struct op_sample {
  uint64_t start_ns;  // since state.anchor
  uint64_t dur_ns;    // issue -> completion; end = start_ns + dur_ns
  uint32_t bytes;     // bytes moved (received len for trecv)
  uint32_t link_id;   // -> (local_nic, remote_nic), resolved at dump time
  uint16_t kind;      // op_kind
  uint16_t error;     // 1 if the op completed in error, else 0
};

// Membership test for a rank spec like "0,8,16" or "0-7,16,32-40".
bool rank_selected(const char* spec, long rank) {
  if (!spec || !*spec || rank < 0) return false;
  for (const char* p = spec; *p;) {
    char* e = nullptr;
    long lo = std::strtol(p, &e, 10);
    if (e == p) break;  // not a number: stop
    long hi = lo;
    if (*e == '-') {
      const char* q = e + 1;
      hi = std::strtol(q, &e, 10);
      if (e == q) hi = lo;
    }
    if (rank >= lo && rank <= hi) return true;
    while (*e == ',' || *e == ' ') ++e;
    p = e;
  }
  return false;
}

// True iff INTERCEPT_TIMER_OPS_RANKS is set and selects this rank. A rank string
// that is not a number (e.g. "NA" when no launcher env is present) is never
// selected -- we can't tell which process it is.
bool ops_selected_for(const std::string& rank_str) {
  const char* spec = getenv("INTERCEPT_TIMER_OPS_RANKS");
  if (!spec || !*spec) return false;
  char* e = nullptr;
  long r = std::strtol(rank_str.c_str(), &e, 10);
  if (e == rank_str.c_str()) return false;
  return rank_selected(spec, r);
}

// Capacity (in samples) of the per-rank buffer. INTERCEPT_TIMER_OPS_MB sizes it
// in MiB; default 1 GiB. The reservation is virtual (malloc), so the unused
// tail costs no physical memory.
uint64_t ops_capacity() {
  uint64_t mb = 1024;
  if (const char* v = getenv("INTERCEPT_TIMER_OPS_MB")) {
    char* e = nullptr;
    long m = std::strtol(v, &e, 10);
    if (e != v && m > 0) mb = static_cast<uint64_t>(m);
  }
  return (mb * 1024ULL * 1024ULL) / sizeof(op_sample);
}

// NIC-targeted filter: the substring an op's NIC address must contain to be
// logged (empty = mode off). See ops_on / record().
std::string ops_nic_target() {
  const char* v = getenv("INTERCEPT_TIMER_OPS_NIC");
  return v ? std::string(v) : std::string();
}

// Size floor for the per-op log: ops smaller than this many bytes are not
// logged (errors always are). INTERCEPT_TIMER_OPS_MIN_BYTES, default 64 KiB.
uint64_t ops_min_bytes() {
  uint64_t b = 65536;
  if (const char* v = getenv("INTERCEPT_TIMER_OPS_MIN_BYTES")) {
    char* e = nullptr;
    long long m = std::strtoll(v, &e, 10);
    if (e != v && m >= 0) b = static_cast<uint64_t>(m);
  }
  return b;
}

// One in-flight op we are timing. While the op is outstanding a pointer to this
// record lives in the op's context; `user_ctx` keeps the value we displaced so
// we can put it back before the app sees the completion. `busy` marks the slot
// as occupied from claim until its completion is processed, so claim never
// reuses a slot whose op is still outstanding (which would corrupt both the
// timing and the context we restore to the app). `ep`/`peer` carry the link
// identity from issue to completion.
//
// `scratch` MUST be the first member. Providers that negotiate FI_CONTEXT /
// FI_CONTEXT2 (aws-ofi-nccl requests both) treat the `context` pointer we hand
// them as provider-owned scratch and write up to a full fi_context2 (64B) into
// it for the lifetime of the op. By redirecting context at `&rec`, that scratch
// lands in `scratch` and leaves our bookkeeping fields below it intact; without
// it the provider would clobber `busy`/`user_ctx`/... and run off the end of the
// record into the next ring slot.
struct op_record {
  fi_context2 scratch;
  std::atomic<bool> busy{false};
  void* user_ctx = nullptr;
  struct fid_ep* ep = nullptr;
  fi_addr_t peer = FI_ADDR_UNSPEC;
  size_t bytes = 0;
  op_kind kind = OP_SEND;
  time_point start{};
};

// Fixed-size ring of in-flight records. A slot is claimed per op by an atomic
// increment of `seq` (lock-free, thread-safe) and only handed out if it is not
// still busy from an earlier, not-yet-completed op. Size the ring above the
// workload's queue depth so this normally never collides; if it does, the op is
// timed-through untouched and counted in `dropped` rather than overwriting a
// live slot.
//
// Caveat: a slot is freed only when its completion (success or error) is seen,
// so this assumes every timed op produces a CQ entry. Ops with suppressed
// completions (FI_INJECT / selective completion) would leak their slot -- but
// those wouldn't be timed meaningfully anyway.
// Sized for recvs too: NCCL preposts up to NCCL_STEPS recvs on every active
// connection (1000+ peers at scale), and each preposted trecv holds its slot
// until its message arrives.
constexpr size_t kMaxInflight = 16384;

// A trecv that completes faster than this after being posted found its message
// already waiting in the unexpected queue -> the receiver posted late.
constexpr double kImmNs = 50e3;  // 50 us

// Latency stats for one (op kind, message size) within one link.
struct bucket {
  uint64_t count = 0;
  uint64_t errors = 0;
  uint64_t imm = 0;  // completions < kImmNs after issue (read for fi_trecv)
  double total_ns = 0.0;
  double min_ns = std::numeric_limits<double>::infinity();
  double max_ns = 0.0;

  void add(double ns) {
    ++count;
    if (ns < kImmNs) ++imm;
    total_ns += ns;
    if (ns < min_ns) min_ns = ns;
    if (ns > max_ns) max_ns = ns;
  }
};

struct bucket_key {
  op_kind kind;
  size_t bytes;
  bool operator==(const bucket_key& o) const {
    return kind == o.kind && bytes == o.bytes;
  }
};
struct bucket_key_hash {
  size_t operator()(const bucket_key& k) const {
    return std::hash<size_t>()(k.bytes) * 1099511628211ULL ^
           static_cast<size_t>(k.kind);
  }
};

// One directed link: a local endpoint sending/reading to/from a remote peer.
// Its address strings are resolved lazily, once, on the first completion seen
// for the link (the endpoint and AV are still alive at data time).
struct link_data {
  std::mutex mtx;
  bool resolved = false;
  uint32_t id = 0;  // stable per-link index for per-op CSV logging
  // NIC-targeted mode (ops_nic): cached once at resolve time. A send/write/read
  // on this link "targets" the NIC if remote == ops_nic; a recv is "posted by"
  // the NIC if local == ops_nic.
  bool nic_send = false;
  bool nic_recv = false;
  std::string local = "?";
  std::string remote = "?";
  std::unordered_map<bucket_key, bucket, bucket_key_hash> buckets;
};

struct link_id {
  struct fid_ep* ep;
  fi_addr_t peer;
  bool operator==(const link_id& o) const {
    return ep == o.ep && peer == o.peer;
  }
};
struct link_id_hash {
  size_t operator()(const link_id& k) const {
    return std::hash<const void*>()(k.ep) * 1099511628211ULL ^
           std::hash<uint64_t>()(static_cast<uint64_t>(k.peer));
  }
};

// Concurrency-aware aggregate for one local NIC (endpoint). Per-op latency
// can't be summed into throughput, but the wall-clock time the NIC has >=1 op
// in flight can: achieved bandwidth = bytes / busy_ns. `inflight` rises on issue
// and falls on completion; `busy_start` is stamped on the 0->1 transition and a
// closed interval is folded into `busy_ns` on the 1->0 transition, so
// overlapping ops are counted once (union of their intervals), not summed.
// Attribution is per NIC rather than per link on purpose: when several peers
// share a NIC their intervals overlap and can't be split per link, but the
// NIC's own union is unambiguous.
struct nic_stats {
  std::mutex mtx;
  bool resolved = false;
  std::string local = "?";
  uint64_t inflight = 0;
  time_point busy_start{};
  double busy_ns = 0.0;  // wall time with >=1 op in flight
  uint64_t bytes = 0;    // bytes of successfully completed ops
  uint64_t ops = 0;
};

// Poll-gap accounting for one CQ. Lock-free: the polling hot path does one
// steady_clock read and a few relaxed atomics per fi_cq_read call. `last_ns`
// is the previous poll's timestamp (ns since the state's anchor; 0 = none yet);
// the gap between consecutive polls of the same CQ is folded into a fixed
// histogram. Multiple threads polling one CQ measure their combined cadence,
// which is the semantics we want ("was anyone servicing this CQ?").
struct cq_gap {
  std::atomic<struct fid_cq*> cq{nullptr};
  std::atomic<uint64_t> polls{0};
  std::atomic<uint64_t> last_ns{0};
  std::atomic<uint64_t> hist[6]{};  // <10us <100us <1ms <10ms <100ms >=100ms
  std::atomic<uint64_t> max_gap_ns{0};
  std::atomic<uint64_t> max_gap_at_ns{0};  // when the max gap ENDED (anchor-rel)
};
constexpr size_t kMaxCqs = 64;
constexpr uint64_t kGapEdgeNs[5] = {10'000, 100'000, 1'000'000, 10'000'000,
                                    100'000'000};

bool cqgap_env_on() {
  const char* e = getenv("INTERCEPT_TIMER_CQGAP");
  return !(e && std::strcmp(e, "0") == 0);
}

// ===========================================================================
// NIC hardware-counter sampler
// ===========================================================================
//
// A background thread that snapshots Cassini per-NIC counters on a fixed cadence
// (default 100 ms) into a RAM time series and writes it as a CSV at exit, so the
// op-level timeline above can be lined up against what the hardware and the
// retry handler were doing (pause, NACKs, spt_timeouts, resource exhaustion).
//
// Two counter sources, both read the same way -- one unsigned integer at the
// head of the file; strtoull stops at the telemetry '@<timestamp>' suffix or the
// rh stat's trailing newline:
//   * sysfs telemetry  /sys/class/cxi/<dev>/device/telemetry/<name>
//                      "<count>@<unix_sec>.<nsec>"   (pct_*, hni_*, ixe_*, cq_*)
//   * retry handler    /run/cxi/<dev>/<name>         (cxi_rh FUSE, plain int:
//                      nacks, parked_nids, pkts_cancelled_o, ...)
//                      -> emitted as column "rh:<name>"
// The rh source is exported by a userspace FUSE daemon and updates a beat behind
// the hardware, so most of its stats (timeouts, NACK types, *_in_use occupancy)
// are sampled from their pct_* telemetry twins instead; the default rh subset is
// only what the hardware doesn't expose (kCuratedRh -- NID/switch parking,
// cancellations). See kCuratedTelem / kCuratedRh.
//
// Which NIC: by default the one this rank drives, cxi$SLURM_LOCALID (a node runs
// one rank per NIC), so each rank's CSV covers a distinct NIC on that rank's own
// clock anchor -- the SAME anchor as its per-op CSV, so the two line up directly.
// If the local id is unknown or its device is absent, every cxi present is
// sampled. Override with INTERCEPT_TIMER_NIC_DEV=cxi0,cxi2 | all.
//
// The thread is started LAZILY, on the first fabric activity we see (cq_open /
// ep_bind), not at library load: the preload is inherited by every process the
// job forks (ninja, python workers, the launcher); starting at load would have
// each of them spin a sampler and emit a duplicate counter file. A process that
// never touches the fabric never starts one.
//
// Env:
//   INTERCEPT_TIMER_NIC=0            disable entirely (default on)
//   INTERCEPT_TIMER_NIC_MS=<ms>      sample interval, default 100
//   INTERCEPT_TIMER_NIC_DEV=<list>   cxi devices (default cxi$LOCALID, else every
//                                    cxi present); "all" = every cxi present
//   INTERCEPT_TIMER_NIC_COUNTERS=... comma-separated selection (default = a wide
//                                    curated telemetry set + a curated rh subset:
//                                    only the rh stats with no hardware twin, the
//                                    rest superseded by their pct_* counters which
//                                    don't lag the FUSE daemon).
//                                    Tokens: "all" (every telemetry counter +
//                                    every rh stat), "rh:" / "rh:*" (every rh
//                                    stat), "rh:<name>" (one rh stat),
//                                    "<prefix>*" (telemetry prefix glob),
//                                    "<name>" (one telemetry counter).
//   INTERCEPT_TIMER_NIC_MAX_MB=<mb>  cap the RAM time series, default 256

// A wide, research-oriented default telemetry set. Names that are absent on a
// given NIC are silently skipped at open time, so this can over-list safely. The
// retry-handler side (NACKs, timeouts, resource occupancy) is covered here by the
// pct_* counters -- the hardware twins of the cxi_rh stats, read from sysfs so
// they don't lag the FUSE daemon -- and the default rh subset (kCuratedRh) adds
// back only what has no hardware equivalent.
static const char* const kCuratedTelem[] = {
    // throughput: bytes (octets), tx flits, packets per traffic class
    "hni_sts_tx_ok_octets", "hni_sts_rx_ok_octets", "cq_cq_oxe_num_flits",
    "cq_cq_tou_num_flits", "hni_pkts_sent_by_tc_0", "hni_pkts_sent_by_tc_1",
    "hni_pkts_sent_by_tc_2", "hni_pkts_sent_by_tc_3", "hni_pkts_sent_by_tc_4",
    "hni_pkts_sent_by_tc_5", "hni_pkts_sent_by_tc_6", "hni_pkts_sent_by_tc_7",
    "hni_pkts_recv_by_tc_0", "hni_pkts_recv_by_tc_1", "hni_pkts_recv_by_tc_2",
    "hni_pkts_recv_by_tc_3", "hni_pkts_recv_by_tc_4", "hni_pkts_recv_by_tc_5",
    "hni_pkts_recv_by_tc_6", "hni_pkts_recv_by_tc_7",
    // pause / fine-grain flow control
    "hni_pause_sent", "hni_pause_refresh", "hni_fgfc_event_xoff",
    "hni_fgfc_event_xon", "hni_fgfc_discard",
    // ECN marking
    "ixe_pool_ecn_pkts_0", "ixe_pool_ecn_pkts_1", "ixe_pool_ecn_pkts_2",
    "ixe_pool_ecn_pkts_3", "ixe_pool_no_ecn_pkts_0", "ixe_pool_no_ecn_pkts_1",
    "ixe_pool_no_ecn_pkts_2", "ixe_pool_no_ecn_pkts_3",
    // link-level retry (LLR)
    "hni_llr_tx_replay_event", "hni_llr_rx_replay_event",
    "hni_llr_tx_nack_ctl_os", "hni_llr_rx_nack_ctl_os",
    "hni_llr_rx_ack_nack_seq_err",
    // PCT timeouts / NACKs / retries. These are the hardware-side equivalents of
    // the cxi_rh retry-handler stats (rh:spt_timeouts, rh:nack_sequence_error,
    // rh:nack_resource_busy, ...) -- read straight from the NIC's sysfs telemetry,
    // so they advance with the hardware instead of lagging behind the FUSE daemon
    // that exports the rh: side. Prefer these over the rh: columns.
    "pct_spt_timeouts", "pct_sct_timeouts", "pct_tct_timeouts",
    "pct_rsp_dropped_timeout", "pct_bad_seq_nacks", "pct_no_mst_nacks",
    "pct_no_tct_nacks", "pct_no_trs_nacks", "pct_resource_busy",
    "pct_trs_rsp_nack_drops",
    "pct_trs_replay_pend_drops", "pct_req_blocked_retry", "pct_req_no_response",
    "pct_req_blocked_clearing", "pct_req_blocked_closing",
    "pct_retry_srb_requests", "pct_retry_mst_get", "pct_retry_trs_put",
    // PCT connection lifecycle & retry-recovery -- the edge that ENDS a stall and
    // the replay-after-timeout path, the half the timeout/NACK counters above don't
    // show. sct_stall_state = an ordered request (RCCL rendezvous) hit a stalled
    // SCT and had to wait (head-of-line-stall onset); conn_*_open / close_sent /
    // clear_sent = the teardown+reopen that recovers; clear_close_drop = the
    // recovery path itself dropped to outbound-FIFO congestion (same family as
    // trs_rsp_nack_drops, would explain a freeze outlasting one timeout);
    // no_matching_tct / err_no_matching_{trs,mst} = a replay that landed after the
    // target state was freed; req_tct_tmout_drops = the target's view of the freeze.
    "pct_sct_stall_state", "pct_conn_sct_open", "pct_conn_tct_open",
    "pct_close_sent", "pct_clear_sent", "pct_clear_close_drop",
    "pct_no_matching_tct", "pct_err_no_matching_trs", "pct_err_no_matching_mst",
    "pct_req_tct_tmout_drops",
    // PCT resource occupancy (instantaneous + high-water). The hardware twins of
    // the rh:*_in_use stats (spt/mst/sct/trs/tct/smt/srb).
    "pct_prf_spt_status_spt_in_use", "pct_prf_spt_status_max_spt_in_use",
    "pct_prf_mst_status_mst_in_use", "pct_prf_mst_status_max_mst_in_use",
    "pct_prf_sct_status_sct_in_use", "pct_prf_sct_status_max_sct_in_use",
    "pct_prf_trs_status_trs_in_use", "pct_prf_trs_status_max_trs_in_use",
    "pct_prf_tct_status_tct_in_use", "pct_prf_tct_status_max_tct_in_use",
    "pct_prf_smt_status_smt_in_use", "pct_prf_smt_status_max_smt_in_use",
    "pct_prf_srb_status_srb_in_use", "pct_prf_srb_status_max_srb_in_use",
    // inbound drops
    "ixe_rx_pkt_drop_pct", "hni_discard_cntr_0", "hni_discard_cntr_1",
    "hni_discard_cntr_2", "hni_discard_cntr_3", "hni_discard_cntr_4",
    "hni_discard_cntr_5", "hni_discard_cntr_6", "hni_discard_cntr_7",
};

// The cxi_rh retry-handler stats worth keeping by default: the ones with NO
// hardware-telemetry twin. Timeouts, NACK types and resource occupancy are all
// covered by the pct_* counters above (which advance with the hardware instead
// of lagging the FUSE daemon, so we prefer them); what is left here -- the retry
// handler PARKING destinations after persistent failure, and its cancellations --
// is the retry handler's own reaction to congestion and exists nowhere else.
// Emitted as "rh:<name>" columns. The full rh set is still reachable on demand
// via INTERCEPT_TIMER_NIC_COUNTERS=rh: (or =all).
static const char* const kCuratedRh[] = {
    "parked_nids",     "max_parked_nids",      "parked_switches",
    "max_parked_switches", "connections_cancelled", "pkts_cancelled_o",
    "pkts_cancelled_u", "nacks",
};

bool path_is_dir(const std::string& p) {
  struct stat st;
  return stat(p.c_str(), &st) == 0 && S_ISDIR(st.st_mode);
}

// Sorted regular-file names directly under `dir` (no recursion; subdirectories
// such as /run/cxi/<dev>/config are skipped). Empty if `dir` can't be opened.
std::vector<std::string> list_regular_files(const std::string& dir) {
  std::vector<std::string> out;
  DIR* d = opendir(dir.c_str());
  if (!d) return out;
  while (struct dirent* e = readdir(d)) {
    if (e->d_name[0] == '.') continue;
    if (path_is_dir(dir + "/" + e->d_name)) continue;
    out.emplace_back(e->d_name);
  }
  closedir(d);
  std::sort(out.begin(), out.end());
  return out;
}

// The cxi devices present, "cxi<N>" (digits only, excludes e.g. cxi_user).
std::vector<std::string> list_cxi_devices() {
  std::vector<std::string> out;
  DIR* d = opendir("/sys/class/cxi");
  if (!d) return out;
  while (struct dirent* e = readdir(d)) {
    const char* n = e->d_name;
    if (std::strncmp(n, "cxi", 3) != 0 || !n[3]) continue;
    bool digits = true;
    for (const char* p = n + 3; *p; ++p)
      if (!std::isdigit(static_cast<unsigned char>(*p))) { digits = false; break; }
    if (digits) out.emplace_back(n);
  }
  closedir(d);
  std::sort(out.begin(), out.end());
  return out;
}

class nic_sampler {
 public:
  nic_sampler() {
    const char* e = getenv("INTERCEPT_TIMER_NIC");
    enabled_ = !(e && std::strcmp(e, "0") == 0);
    if (const char* v = getenv("INTERCEPT_TIMER_NIC_MS")) {
      char* end;
      long m = std::strtol(v, &end, 10);
      if (end != v && m > 0) interval_ms_ = static_cast<int>(m);
    }
    if (const char* v = getenv("INTERCEPT_TIMER_NIC_MAX_MB")) {
      char* end;
      long m = std::strtol(v, &end, 10);
      if (end != v && m > 0) max_mb_ = static_cast<uint64_t>(m);
    }
  }

  // Start the sampling thread exactly once, sharing the tool's clock anchor so
  // sample timestamps line up with the per-op CSV. Idempotent and thread-safe:
  // the first caller wins the CAS and spawns the thread; everyone else returns.
  void start(time_point anchor, uint64_t anchor_wall_ns) {
    bool expected = false;
    if (!started_.compare_exchange_strong(expected, true)) return;
    if (!enabled_) return;
    anchor_ = anchor;
    anchor_wall_ns_ = anchor_wall_ns;
    open_counters();
    if (labels_.empty()) return;
    // Reserve the WHOLE cap up front rather than growing. At the default 100 ms
    // the growth reallocs are invisible, but at a 1 ms cadence the series passes
    // 16k samples in 16 s and every later realloc copies (and so touches) up to
    // MAX_MB of pages inside a sample period -- a self-inflicted gap in the very
    // time series you are sampling. The reservation is virtual; only the samples
    // actually taken are ever touched.
    rows_.reserve(max_samples_ * stride_);
    thread_ = std::thread([this] { run(); });
  }

  ~nic_sampler() {
    if (thread_.joinable()) {
      {
        std::lock_guard<std::mutex> lk(mtx_);
        stop_ = true;
      }
      cv_.notify_all();
      thread_.join();
    }
    dump();
    for (int fd : fds_)
      if (fd >= 0) close(fd);
  }

 private:
  static uint64_t realtime_ns() {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return static_cast<uint64_t>(ts.tv_sec) * 1000000000ULL + ts.tv_nsec;
  }

  uint64_t ns_since_anchor(time_point t) {
    return static_cast<uint64_t>(
        std::chrono::duration_cast<std::chrono::nanoseconds>(t - anchor_)
            .count());
  }

  static std::string read_first_line(const std::string& path) {
    int fd = open(path.c_str(), O_RDONLY);
    if (fd < 0) return "";
    char buf[128];
    ssize_t n = read(fd, buf, sizeof(buf) - 1);
    close(fd);
    if (n <= 0) return "";
    buf[n] = '\0';
    if (char* nl = std::strchr(buf, '\n')) *nl = '\0';
    return buf;
  }

  static long local_id() {
    static const char* const keys[] = {
        "SLURM_LOCALID", "MPI_LOCALRANKID", "OMPI_COMM_WORLD_LOCAL_RANK",
        "PMI_LOCAL_RANK", "FLUX_TASK_LOCAL_ID"};
    for (const char* k : keys)
      if (const char* v = getenv(k)) {
        char* e;
        long r = std::strtol(v, &e, 10);
        if (e != v) return r;
      }
    return -1;
  }

  static std::vector<std::string> split_csv(const std::string& s) {
    std::vector<std::string> out;
    size_t i = 0;
    while (i < s.size()) {
      size_t j = s.find(',', i);
      if (j == std::string::npos) j = s.size();
      size_t a = i, b = j;
      while (a < b && std::isspace(static_cast<unsigned char>(s[a]))) ++a;
      while (b > a && std::isspace(static_cast<unsigned char>(s[b - 1]))) --b;
      if (b > a) out.push_back(s.substr(a, b - a));
      i = j + 1;
    }
    return out;
  }

  std::vector<std::string> resolve_devices() {
    std::vector<std::string> all = list_cxi_devices();
    const char* spec = getenv("INTERCEPT_TIMER_NIC_DEV");
    if (spec && *spec) {
      std::string s = spec;
      if (s == "all") return all;
      std::vector<std::string> out;
      for (const std::string& tok : split_csv(s))
        if (path_is_dir("/sys/class/cxi/" + tok)) out.push_back(tok);
      if (!out.empty()) return out;
      return all;  // nothing matched: fall back to every NIC present
    }
    long lid = local_id();
    if (lid >= 0) {
      std::string dev = "cxi" + std::to_string(lid);
      if (path_is_dir("/sys/class/cxi/" + dev)) return {dev};
    }
    return all;  // no local id (or its device absent): sample every NIC present
  }

  // Open one counter file and register its column. Deduped by label, and skipped
  // silently if the file is missing/unreadable (a curated name absent on this
  // NIC, an rh stat with the daemon down, ...).
  void add_counter(const std::string& label, const std::string& path) {
    if (!seen_.insert(label).second) return;
    int fd = open(path.c_str(), O_RDONLY);
    if (fd < 0) return;
    labels_.push_back(label);
    fds_.push_back(fd);
  }

  void open_counters() {
    devices_ = resolve_devices();
    const char* spec = getenv("INTERCEPT_TIMER_NIC_COUNTERS");
    std::string sel = spec ? spec : "";
    for (const std::string& dev : devices_) {
      const std::string tdir = "/sys/class/cxi/" + dev + "/device/telemetry";
      const std::string rdir = "/run/cxi/" + dev;
      if (sel.empty()) {  // default: wide curated telemetry + curated rh stats
        for (const char* name : kCuratedTelem)
          add_counter(dev + ":" + name, tdir + "/" + name);
        for (const char* name : kCuratedRh)
          add_counter(dev + ":rh:" + name, rdir + "/" + name);
        continue;
      }
      for (const std::string& tok : split_csv(sel)) {
        if (tok == "all") {
          for (const std::string& n : list_regular_files(tdir))
            add_counter(dev + ":" + n, tdir + "/" + n);
          for (const std::string& n : list_regular_files(rdir))
            add_counter(dev + ":rh:" + n, rdir + "/" + n);
        } else if (tok == "rh:" || tok == "rh:*") {
          for (const std::string& n : list_regular_files(rdir))
            add_counter(dev + ":rh:" + n, rdir + "/" + n);
        } else if (tok.rfind("rh:", 0) == 0) {
          add_counter(dev + ":" + tok, rdir + "/" + tok.substr(3));
        } else if (tok.back() == '*') {
          std::string pre = tok.substr(0, tok.size() - 1);
          for (const std::string& n : list_regular_files(tdir))
            if (n.rfind(pre, 0) == 0) add_counter(dev + ":" + n, tdir + "/" + n);
        } else {
          add_counter(dev + ":" + tok, tdir + "/" + tok);
        }
      }
    }
    stride_ = labels_.size() + 2;  // t_ns, wall_ns, then one column per counter
    uint64_t cap_bytes = max_mb_ << 20;
    max_samples_ =
        stride_ ? std::max<uint64_t>(1, cap_bytes / (stride_ * sizeof(uint64_t)))
                : 0;
  }

  void sample_once() {
    if (nsamples_ >= max_samples_) {
      ++dropped_;
      return;
    }
    uint64_t t = ns_since_anchor(clock_type::now());
    uint64_t w = realtime_ns();
    size_t base = rows_.size();
    rows_.resize(base + stride_);
    rows_[base] = t;
    rows_[base + 1] = w;
    char buf[64];
    for (size_t i = 0; i < fds_.size(); ++i) {
      uint64_t v = 0;
      ssize_t n = pread(fds_[i], buf, sizeof(buf) - 1, 0);
      if (n > 0) {
        buf[n] = '\0';
        v = std::strtoull(buf, nullptr, 10);  // stops at '@' or '\n'
      }
      rows_[base + 2 + i] = v;
    }
    ++nsamples_;
  }

  void run() {
    auto period = std::chrono::milliseconds(interval_ms_);
    auto next = clock_type::now();
    std::unique_lock<std::mutex> lk(mtx_);
    while (!stop_) {
      lk.unlock();
      sample_once();  // only this thread writes the series; no lock needed
      lk.lock();
      next += period;
      cv_.wait_until(lk, next, [this] { return stop_; });
    }
  }

  void dump() {
    if (!enabled_ || labels_.empty() || nsamples_ == 0) return;

    char host[65] = {0};
    if (gethostname(host, sizeof(host) - 1) != 0) std::strcpy(host, "?");
    std::string rank = detect_rank();
    long pid = static_cast<long>(getpid());

    const char* dir = getenv("INTERCEPT_TIMER_DIR");
    bool to_stderr =
        dir && (std::strcmp(dir, "stderr") == 0 || std::strcmp(dir, "-") == 0);
    char path[600];
    snprintf(path, sizeof(path), "%s/intercept_timer_nic.%s.r%s.p%ld.csv",
             (dir && !to_stderr) ? dir : ".", host, rank.c_str(), pid);
    FILE* f = fopen(path, "w");
    if (!f) {
      fprintf(stderr, "[intercept_timer] could not open %s for NIC counters\n",
              path);
      return;
    }
    static char iobuf[1 << 20];
    setvbuf(f, iobuf, _IOFBF, sizeof(iobuf));

    std::string devs;
    for (size_t i = 0; i < devices_.size(); ++i)
      devs += (i ? "," : "") + devices_[i];
    fprintf(f, "# intercept_timer NIC counters host=%s rank=%s pid=%ld\n", host,
            rank.c_str(), pid);
    // Achieved cadence: one sample costs a pread per counter (each sysfs
    // telemetry read is a fresh hardware read), so a small interval_ms_ can be
    // unmeetable. Compare requested vs achieved before reading the series as a
    // fixed-rate signal -- if achieved > requested the thread is running flat
    // out and the real cadence is whatever this line says.
    double span_ms = nsamples_ > 1
                         ? (rows_[(nsamples_ - 1) * stride_] - rows_[0]) / 1e6
                         : 0.0;
    fprintf(f,
            "# interval_ms=%d achieved_mean_ms=%.3f devices=%s counters=%zu "
            "samples=%llu\n",
            interval_ms_, nsamples_ > 1 ? span_ms / (nsamples_ - 1) : 0.0,
            devs.c_str(), labels_.size(),
            static_cast<unsigned long long>(nsamples_));
    fprintf(f,
            "# t_ns = ns since this rank's clock anchor (same anchor as the "
            "per-op CSV); wall_ns = CLOCK_REALTIME for cross-rank alignment\n");
    fprintf(f,
            "# anchor_mono_ns=%llu anchor_wall_ns=%llu  (t_ns clock = "
            "std::chrono::steady_clock; t_ns = steady_ns - anchor_mono_ns; "
            "wall = steady_ns + (anchor_wall_ns - anchor_mono_ns))\n",
            static_cast<unsigned long long>(
                std::chrono::duration_cast<std::chrono::nanoseconds>(
                    anchor_.time_since_epoch())
                    .count()),
            static_cast<unsigned long long>(anchor_wall_ns_));
    for (const std::string& dev : devices_) {
      std::string na =
          read_first_line("/sys/class/cxi/" + dev + "/device/properties/nic_addr");
      std::string nid =
          read_first_line("/sys/class/cxi/" + dev + "/device/properties/nid");
      fprintf(f, "# dev %s nic_addr=%s nid=%s\n", dev.c_str(),
              na.empty() ? "?" : na.c_str(), nid.empty() ? "?" : nid.c_str());
    }

    fputs("t_ns,wall_ns", f);
    for (const std::string& l : labels_) {
      fputc(',', f);
      fputs(l.c_str(), f);
    }
    fputc('\n', f);

    for (uint64_t s = 0; s < nsamples_; ++s) {
      const uint64_t* row = &rows_[s * stride_];
      fprintf(f, "%llu,%llu", static_cast<unsigned long long>(row[0]),
              static_cast<unsigned long long>(row[1]));
      for (size_t i = 0; i < labels_.size(); ++i)
        fprintf(f, ",%llu", static_cast<unsigned long long>(row[2 + i]));
      fputc('\n', f);
    }
    fclose(f);

    fprintf(stderr,
            "[intercept_timer] %s rank %s NIC counters -> %s "
            "(%llu samples x %zu counters)\n",
            host, rank.c_str(), path,
            static_cast<unsigned long long>(nsamples_), labels_.size());
    if (dropped_)
      fprintf(stderr,
              "[intercept_timer] WARNING: %llu NIC samples dropped (RAM cap); "
              "raise INTERCEPT_TIMER_NIC_MAX_MB\n",
              static_cast<unsigned long long>(dropped_));
  }

  // ---- config (read in the constructor) ----
  bool enabled_ = true;
  int interval_ms_ = 100;
  uint64_t max_mb_ = 256;

  // ---- resolved at start() ----
  time_point anchor_{};
  uint64_t anchor_wall_ns_ = 0;
  std::vector<std::string> devices_;
  std::vector<std::string> labels_;  // column names, parallel to fds_
  std::vector<int> fds_;
  std::unordered_set<std::string> seen_;  // label dedup during open
  size_t stride_ = 0;                     // uint64s per sample (2 + #counters)
  uint64_t max_samples_ = 0;

  // ---- time series: sample-major, stride_ uint64s each (t_ns, wall_ns, ...) ----
  std::vector<uint64_t> rows_;
  uint64_t nsamples_ = 0;
  uint64_t dropped_ = 0;

  // ---- thread control ----
  std::atomic<bool> started_{false};
  bool stop_ = false;
  std::mutex mtx_;
  std::condition_variable cv_;
  std::thread thread_;
};

struct state {
  std::atomic<uint64_t> seq{0};
  std::atomic<uint64_t> dropped{0};
  op_record ring[kMaxInflight];

  // CQ poll-gap registry. Slots are appended (never freed) by cq_open or on
  // first sight of an unknown CQ in a poll; lookup is a short linear scan.
  const time_point anchor = clock_type::now();
  // CLOCK_REALTIME captured at (essentially) the same instant as `anchor`. We
  // publish BOTH the steady-clock anchor (anchor_mono_ns) and this wall value in
  // the CSV headers -- "recording the anchor" -- so a SAME-PROCESS consumer that
  // stamps events with the identical std::chrono::steady_clock (e.g. the
  // iter_timeline all-to-all boundaries) maps them onto t_ns by exact subtraction:
  //   t_ns(event) = event_steady_ns - anchor_mono_ns()
  // and onto the global wall clock (cross-rank) by adding the constant
  //   (anchor_wall_ns - anchor_mono_ns). No NIC-sample fitting, no NTP slew.
  const uint64_t anchor_wall_ns = [] {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return static_cast<uint64_t>(ts.tv_sec) * 1000000000ULL + ts.tv_nsec;
  }();
  uint64_t anchor_mono_ns() const {
    return static_cast<uint64_t>(
        std::chrono::duration_cast<std::chrono::nanoseconds>(
            anchor.time_since_epoch())
            .count());
  }
  const bool cqgap_on = cqgap_env_on();
  cq_gap cq_gaps[kMaxCqs];
  std::atomic<size_t> ncqs{0};

  uint64_t ns_since_anchor(time_point t) {
    return static_cast<uint64_t>(
        std::chrono::duration_cast<std::chrono::nanoseconds>(t - anchor)
            .count());
  }

  cq_gap* gap_for(struct fid_cq* cq) {
    size_t n = ncqs.load(std::memory_order_acquire);
    for (size_t i = 0; i < n; ++i)
      if (cq_gaps[i].cq.load(std::memory_order_relaxed) == cq)
        return &cq_gaps[i];
    // Unknown CQ: append (first poll of a CQ races benignly; one slot wins,
    // losers find it on their next scan and this poll goes uncounted).
    size_t i = ncqs.load(std::memory_order_relaxed);
    if (i >= kMaxCqs) return nullptr;
    struct fid_cq* expected = nullptr;
    if (cq_gaps[i].cq.compare_exchange_strong(expected, cq,
                                              std::memory_order_relaxed)) {
      ncqs.store(i + 1, std::memory_order_release);
      return &cq_gaps[i];
    }
    return nullptr;
  }

  // The per-poll hot path: stamp this poll, histogram the gap since the last.
  void poll_mark(struct fid_cq* cq) {
    if (!cqgap_on) return;
    cq_gap* g = gap_for(cq);
    if (!g) return;
    uint64_t now = ns_since_anchor(clock_type::now());
    uint64_t prev = g->last_ns.exchange(now, std::memory_order_relaxed);
    g->polls.fetch_add(1, std::memory_order_relaxed);
    if (prev == 0) return;  // first observed poll: no gap yet
    uint64_t gap = now - prev;
    int b = 5;
    for (int e = 0; e < 5; ++e)
      if (gap < kGapEdgeNs[e]) { b = e; break; }
    g->hist[b].fetch_add(1, std::memory_order_relaxed);
    uint64_t m = g->max_gap_ns.load(std::memory_order_relaxed);
    while (gap > m) {
      if (g->max_gap_ns.compare_exchange_weak(m, gap,
                                              std::memory_order_relaxed)) {
        g->max_gap_at_ns.store(now, std::memory_order_relaxed);
        break;
      }
    }
  }

  std::mutex fmt_mtx;
  std::unordered_map<fid_cq*, fi_cq_format> cq_format;

  // ep -> the address vector bound to it (captured in on_ep_bind), used to
  // resolve fi_addr_t peers to readable NIC addresses.
  std::mutex ep_av_mtx;
  std::unordered_map<struct fid_ep*, struct fid_av*> ep_av;

  // The per-link aggregates. unique_ptr so a link_data's address (and its mutex)
  // is stable across rehashes of the registry.
  std::mutex links_mtx;
  std::unordered_map<link_id, std::unique_ptr<link_data>, link_id_hash> links;

  // Per-NIC (per local endpoint) aggregate for concurrency-aware bandwidth.
  std::mutex nics_mtx;
  std::unordered_map<struct fid_ep*, std::unique_ptr<nic_stats>> nics;

  // ---- Per-op CSV logging (opt-in via INTERCEPT_TIMER_OPS_RANKS) ----------
  // Config resolved once at construction; the buffer (selected ranks only) is
  // malloc'd once in the constructor, which runs at library load -- before the
  // app makes its fabric or any polling thread exists -- so `ops_buf` is a plain
  // pointer needing no lock/atomic to publish. A 1GiB malloc is virtual until
  // written, so a forked child that inherits the preload but does no fabric I/O
  // pays no physical memory. The per-op append is lock-free: one atomic bump of
  // `ops_idx` + a store; samples are formatted to CSV once, at exit.
  const std::string ops_rank = detect_rank();
  const std::string ops_nic = ops_nic_target();
  const bool ops_nic_on = !ops_nic.empty();
  const uint64_t ops_min_b = ops_min_bytes();
  // NIC-targeted mode runs on EVERY rank (filtered per op to ops touching
  // ops_nic); otherwise fall back to the rank-list selection.
  const bool ops_on = ops_nic_on || ops_selected_for(ops_rank);
  const uint64_t ops_cap = ops_capacity();
  op_sample* ops_buf = nullptr;
  std::atomic<uint64_t> ops_idx{0};
  std::atomic<uint64_t> ops_dropped{0};
  uint32_t next_link_id = 0;  // distinct-link counter, guarded by links_mtx

  // Background NIC hardware-counter sampler. Constructed here (reads its env),
  // but its thread is started lazily on the first fabric activity (see the
  // nic.start() calls in cq_open / ep_bind) so non-fabric forked children don't
  // each spin one. Destroyed with `state` at exit, which writes its CSV.
  nic_sampler nic;

  state() {
    if (ops_on)
      ops_buf = static_cast<op_sample*>(malloc(ops_cap * sizeof(op_sample)));
  }
  ~state() { dump(); }

  // Append one completed op. The caller (record) already holds ld->mtx, but this
  // append takes no lock of its own -- just an atomic cursor bump. `lid` is the
  // stable per-link index resolved to NIC strings at dump time.
  void log_op(uint32_t lid, op_record* rec, double ns, bool error) {
    if (!ops_buf) return;
    uint64_t i = ops_idx.fetch_add(1, std::memory_order_relaxed);
    if (i >= ops_cap) {
      ops_dropped.fetch_add(1, std::memory_order_relaxed);
      return;
    }
    op_sample& o = ops_buf[i];
    o.start_ns = ns_since_anchor(rec->start);
    o.dur_ns = ns > 0.0 ? static_cast<uint64_t>(ns) : 0;
    o.bytes = static_cast<uint32_t>(rec->bytes);
    o.link_id = lid;
    o.kind = static_cast<uint16_t>(rec->kind);
    o.error = error ? 1 : 0;
  }

  struct fid_av* av_for(struct fid_ep* ep) {
    std::lock_guard<std::mutex> lk(ep_av_mtx);
    auto it = ep_av.find(ep);
    return it == ep_av.end() ? nullptr : it->second;
  }

  link_data* link_for(struct fid_ep* ep, fi_addr_t peer) {
    std::lock_guard<std::mutex> lk(links_mtx);
    auto& slot = links[link_id{ep, peer}];
    if (!slot) {
      slot = std::make_unique<link_data>();
      slot->id = next_link_id++;
    }
    return slot.get();
  }

  nic_stats* nic_for(struct fid_ep* ep) {
    std::lock_guard<std::mutex> lk(nics_mtx);
    auto& slot = nics[ep];
    if (!slot) slot = std::make_unique<nic_stats>();
    return slot.get();
  }

  // Resolve an endpoint's local NIC address to a readable string. Off the hot
  // path; the endpoint and its AV are alive at data time.
  std::string local_addr(struct fid_ep* ep) {
    struct fid_av* av = av_for(ep);
    char raw[256];
    char str[128];
    if (ep) {
      size_t rl = sizeof(raw);
      if (fi_getname(&ep->fid, raw, &rl) == 0 && av) {
        size_t sl = sizeof(str);
        if (const char* s = fi_av_straddr(av, raw, str, &sl)) return s;
      }
    }
    char b[64];
    snprintf(b, sizeof(b), "ep@%p", static_cast<void*>(ep));
    return b;
  }

  std::string remote_addr(struct fid_ep* ep, fi_addr_t peer) {
    struct fid_av* av = av_for(ep);
    if (av && peer != FI_ADDR_UNSPEC) {
      char raw[256];
      char str[128];
      size_t rl = sizeof(raw);
      if (fi_av_lookup(av, peer, raw, &rl) == 0) {
        size_t sl = sizeof(str);
        if (const char* s = fi_av_straddr(av, raw, str, &sl)) return s;
      }
    }
    if (peer == FI_ADDR_UNSPEC) return "(unspec)";
    return "fi_addr#" + std::to_string(static_cast<uint64_t>(peer));
  }

  void record(op_record* rec, double ns, bool error) {
    link_data* ld = link_for(rec->ep, rec->peer);
    std::lock_guard<std::mutex> lk(ld->mtx);
    if (!ld->resolved) {
      ld->local = local_addr(rec->ep);
      ld->remote = remote_addr(rec->ep, rec->peer);
      if (ops_nic_on) {
        ld->nic_send = ld->remote.find(ops_nic) != std::string::npos;
        ld->nic_recv = ld->local.find(ops_nic) != std::string::npos;
      }
      ld->resolved = true;
    }
    bucket& b = ld->buckets[bucket_key{rec->kind, rec->bytes}];
    if (error)
      ++b.errors;
    else
      b.add(ns);
    // In NIC-targeted mode, log only ops touching ops_nic: sends/writes/reads
    // aimed at it (remote == nic) and recvs it posted (local == nic). Ops below
    // the size floor are skipped (signaling), unless they completed in error.
    if (ops_on && (rec->bytes >= ops_min_b || error) &&
        (!ops_nic_on || (rec->kind == OP_TRECV ? ld->nic_recv : ld->nic_send)))
      log_op(ld->id, rec, ns, error);
  }

  // Issue side: account a newly-submitted op against its NIC's busy interval.
  void nic_issue(struct fid_ep* ep, time_point start) {
    nic_stats* n = nic_for(ep);
    std::lock_guard<std::mutex> lk(n->mtx);
    if (n->inflight++ == 0) n->busy_start = start;
  }

  // Undo a nic_issue for an op the provider rejected (next() returned < 0): it
  // was never submitted, so no completion will close its interval.
  void nic_unissue(struct fid_ep* ep) {
    nic_stats* n = nic_for(ep);
    std::lock_guard<std::mutex> lk(n->mtx);
    if (n->inflight) --n->inflight;
  }

  // Completion side: close the NIC's busy interval when it drains, and add the
  // bytes of a successful op. Errors still decrement inflight (bytes unchanged).
  void nic_complete(struct fid_ep* ep, size_t bytes, time_point end, bool ok) {
    nic_stats* n = nic_for(ep);
    std::lock_guard<std::mutex> lk(n->mtx);
    if (!n->resolved) {
      n->local = local_addr(ep);
      n->resolved = true;
    }
    if (ok) {
      n->bytes += bytes;
      ++n->ops;
    }
    if (n->inflight && --n->inflight == 0)
      n->busy_ns +=
          std::chrono::duration<double, std::nano>(end - n->busy_start).count();
  }

  void dump() {
    // Skip processes that did no fabric I/O. The preload is inherited by every
    // child the app forks (JIT compilers, ninja, Python workers, the launcher),
    // all of which would otherwise emit an empty report -- and they share the
    // parent's SLURM_PROCID, so they'd collide on the same rank name too. A
    // process only has a nic/link entry if it actually issued/completed an op.
    {
      bool empty;
      {
        std::lock_guard<std::mutex> lk(nics_mtx);
        empty = nics.empty();
      }
      if (empty) {
        std::lock_guard<std::mutex> lk(links_mtx);
        empty = links.empty();
      }
      if (empty) return;
    }

    auto now = clock_type::now();

    // Pick the output stream. Each rank writes its own file by default so that
    // the (multi-line) reports of all ranks don't interleave on a shared stderr.
    //   INTERCEPT_TIMER_DIR=<dir>   -> <dir>/intercept_timer.<host>.r<rank>.p<pid>.txt
    //   INTERCEPT_TIMER_DIR unset   -> same, in the current directory
    //   INTERCEPT_TIMER_DIR=stderr  (or "-") -> write to stderr (interleaves!)
    char host[65] = {0};
    if (gethostname(host, sizeof(host) - 1) != 0) std::strcpy(host, "?");
    std::string rank = detect_rank();
    long pid = static_cast<long>(getpid());

    const char* dir = getenv("INTERCEPT_TIMER_DIR");
    bool to_stderr =
        dir && (std::strcmp(dir, "stderr") == 0 || std::strcmp(dir, "-") == 0);
    FILE* out = stderr;
    bool close_out = false;
    if (!to_stderr) {
      char path[600];
      snprintf(path, sizeof(path), "%s/intercept_timer.%s.r%s.p%ld.txt",
               dir ? dir : ".", host, rank.c_str(), pid);
      if (FILE* f = fopen(path, "w")) {
        out = f;
        close_out = true;
        // One short line to stderr is fine even interleaved; it just points at
        // the real report so you can find it.
        fprintf(stderr, "[intercept_timer] %s rank %s -> %s\n", host,
                rank.c_str(), path);
      }
    }

    fprintf(out, "[intercept_timer] host=%s rank=%s pid=%ld\n", host,
            rank.c_str(), pid);

    // Per-NIC concurrency-aware bandwidth: achieved = bytes / busy time (the
    // wall-clock union of all of the NIC's in-flight intervals).
    {
      std::vector<nic_stats*> ns;
      {
        std::lock_guard<std::mutex> lk(nics_mtx);
        for (auto& kv : nics) ns.push_back(kv.second.get());
      }
      std::sort(ns.begin(), ns.end(),
                [](nic_stats* a, nic_stats* b) { return a->local < b->local; });
      for (nic_stats* n : ns) {
        std::lock_guard<std::mutex> lk(n->mtx);
        double busy = n->busy_ns;
        if (n->inflight)  // an interval was still open at exit; close it now
          busy += std::chrono::duration<double, std::nano>(now - n->busy_start)
                      .count();
        if (n->ops == 0 && busy == 0.0) continue;
        double gbps = busy > 0.0 ? double(n->bytes) / busy : 0.0;
        fprintf(out,
                "[intercept_timer] NIC %-26s %10llu ops  %16llu bytes  "
                "busy %12.3f ms  achieved %8.3f GB/s\n",
                n->local.c_str(), static_cast<unsigned long long>(n->ops),
                static_cast<unsigned long long>(n->bytes), busy / 1e6, gbps);
      }
    }

    // Per-CQ poll-gap histograms (see header comment: counts of gaps between
    // consecutive polls of the same CQ; ms+ buckets are the stall signal).
    if (cqgap_on) {
      size_t n = ncqs.load(std::memory_order_acquire);
      for (size_t i = 0; i < n; ++i) {
        cq_gap& g = cq_gaps[i];
        uint64_t polls = g.polls.load(std::memory_order_relaxed);
        if (!polls) continue;
        fprintf(out,
                "[intercept_timer] CQ#%zu polls %llu  gaps: <10us %llu  "
                "<100us %llu  <1ms %llu  <10ms %llu  <100ms %llu  "
                ">=100ms %llu  max %.3f ms at t+%.3f s\n",
                i, static_cast<unsigned long long>(polls),
                static_cast<unsigned long long>(g.hist[0].load()),
                static_cast<unsigned long long>(g.hist[1].load()),
                static_cast<unsigned long long>(g.hist[2].load()),
                static_cast<unsigned long long>(g.hist[3].load()),
                static_cast<unsigned long long>(g.hist[4].load()),
                static_cast<unsigned long long>(g.hist[5].load()),
                g.max_gap_ns.load() / 1e6, g.max_gap_at_ns.load() / 1e9);
      }
    }

    // Stable order: sort links by (local, remote), buckets by (kind, size).
    std::vector<const std::pair<const link_id, std::unique_ptr<link_data>>*>
        order;
    {
      std::lock_guard<std::mutex> lk(links_mtx);
      for (auto& kv : links) order.push_back(&kv);
    }
    std::sort(order.begin(), order.end(), [](auto* a, auto* b) {
      if (a->second->local != b->second->local)
        return a->second->local < b->second->local;
      return a->second->remote < b->second->remote;
    });

    for (auto* kv : order) {
      link_data* ld = kv->second.get();
      std::lock_guard<std::mutex> lk(ld->mtx);
      if (ld->buckets.empty()) continue;
      fprintf(out, "[intercept_timer] === %s -> %s ===\n", ld->local.c_str(),
              ld->remote.c_str());

      std::vector<std::pair<bucket_key, bucket*>> rows;
      for (auto& bk : ld->buckets) rows.push_back({bk.first, &bk.second});
      std::sort(rows.begin(), rows.end(), [](auto& a, auto& b) {
        if (a.first.kind != b.first.kind) return a.first.kind < b.first.kind;
        return a.first.bytes < b.first.bytes;
      });

      for (auto& r : rows) {
        const bucket& b = *r.second;
        double avg_us = b.count ? b.total_ns / double(b.count) / 1e3 : 0.0;
        // per-op bandwidth: bytes / avg_ns == GB/s (1e9-byte GB).
        double gbps =
            (b.count && b.total_ns > 0.0)
                ? double(r.first.bytes) * double(b.count) / b.total_ns
                : 0.0;
        fprintf(out,
                "[intercept_timer]   %-12s %10zu B  %8llu ops  "
                "avg %9.3f us  min %9.3f  max %9.3f us  ~%7.3f GB/s",
                op_name(r.first.kind), r.first.bytes,
                static_cast<unsigned long long>(b.count), avg_us,
                b.count ? b.min_ns / 1e3 : 0.0, b.max_ns / 1e3, gbps);
        // For recvs, the late-poster signal: completed < 50us after posting
        // means the message was already waiting in the unexpected queue.
        if (r.first.kind == OP_TRECV && b.count)
          fprintf(out, "  [imm %llu = %.1f%%]",
                  static_cast<unsigned long long>(b.imm),
                  100.0 * double(b.imm) / double(b.count));
        if (b.errors)
          fprintf(out, "  [%llu err]",
                  static_cast<unsigned long long>(b.errors));
        fputc('\n', out);
      }
    }

    uint64_t d = dropped.load(std::memory_order_relaxed);
    if (d) {
      fprintf(out,
              "[intercept_timer] %llu ops not timed: in-flight ring (%zu slots) "
              "full -- raise kMaxInflight\n",
              static_cast<unsigned long long>(d), kMaxInflight);
    }

    if (close_out) fclose(out);

    write_ops_csv();
  }

  // Per-op CSV (opt-in). Writes one row per recorded op to its own per-rank
  // file, resolving each sample's link_id to NIC strings via the already-
  // resolved link registry. Millions of rows, but at exit and off the hot path.
  void write_ops_csv() {
    if (!ops_on || !ops_buf) return;
    uint64_t used = ops_idx.load(std::memory_order_relaxed);
    if (used > ops_cap) used = ops_cap;
    if (used == 0) {
      // Loud, because a wrong INTERCEPT_TIMER_OPS_NIC prefix (e.g. a raw NID
      // instead of the nid<<1 string form) otherwise fails silently on every
      // rank at once.
      if (ops_nic_on)
        fprintf(stderr,
                "[intercept_timer] rank %s: per-op mode on but 0 ops matched "
                "INTERCEPT_TIMER_OPS_NIC=%s (wrong prefix? string form is "
                "0x%%06x of nid<<1)\n",
                ops_rank.c_str(), ops_nic.c_str());
      return;
    }

    // link_id -> (local, remote) from the already-resolved link registry; every
    // link that recorded an op was resolved on its first completion.
    std::vector<std::pair<std::string, std::string>> dict;
    {
      std::lock_guard<std::mutex> lk(links_mtx);
      dict.resize(next_link_id);
      for (auto& kv : links) {
        link_data* ld = kv.second.get();
        if (ld->id < dict.size()) dict[ld->id] = {ld->local, ld->remote};
      }
    }

    char host[65] = {0};
    if (gethostname(host, sizeof(host) - 1) != 0) std::strcpy(host, "?");
    long pid = static_cast<long>(getpid());
    const char* dir = getenv("INTERCEPT_TIMER_DIR");
    bool to_stderr =
        dir && (std::strcmp(dir, "stderr") == 0 || std::strcmp(dir, "-") == 0);
    char path[600];
    snprintf(path, sizeof(path), "%s/intercept_timer_ops.%s.r%s.p%ld.csv",
             (dir && !to_stderr) ? dir : ".", host, ops_rank.c_str(), pid);
    FILE* f = fopen(path, "w");
    if (!f) {
      fprintf(stderr, "[intercept_timer] could not open %s for per-op CSV\n",
              path);
      return;
    }
    // One big stdio buffer: this is millions of formatted rows, so keep it to a
    // few large writes.
    static char iobuf[1 << 20];
    setvbuf(f, iobuf, _IOFBF, sizeof(iobuf));

    fprintf(f,
            "# anchor_mono_ns=%llu anchor_wall_ns=%llu  (start_ns/end_ns are "
            "std::chrono::steady_clock ns since anchor; absolute steady_ns = "
            "start_ns + anchor_mono_ns; same-process events stamped with "
            "steady_clock map in by subtracting anchor_mono_ns)\n",
            static_cast<unsigned long long>(anchor_mono_ns()),
            static_cast<unsigned long long>(anchor_wall_ns));
    fprintf(f, "start_ns,end_ns,dur_ns,op,bytes,local_nic,remote_nic,error\n");
    for (uint64_t i = 0; i < used; ++i) {
      const op_sample& o = ops_buf[i];
      const char* ln = "?";
      const char* rn = "?";
      if (o.link_id < dict.size()) {
        ln = dict[o.link_id].first.c_str();
        rn = dict[o.link_id].second.c_str();
      }
      fprintf(f, "%llu,%llu,%llu,%s,%u,%s,%s,%u\n",
              static_cast<unsigned long long>(o.start_ns),
              static_cast<unsigned long long>(o.start_ns + o.dur_ns),
              static_cast<unsigned long long>(o.dur_ns),
              op_name(static_cast<op_kind>(o.kind)), o.bytes, ln, rn,
              static_cast<unsigned>(o.error));
    }
    fclose(f);

    uint64_t dr = ops_dropped.load(std::memory_order_relaxed);
    fprintf(stderr, "[intercept_timer] %s rank %s per-op -> %s (%llu ops)\n",
            host, ops_rank.c_str(), path,
            static_cast<unsigned long long>(used));
    if (dr)
      fprintf(stderr,
              "[intercept_timer] WARNING: %llu ops dropped (buffer full); "
              "raise INTERCEPT_TIMER_OPS_MB\n",
              static_cast<unsigned long long>(dr));
  }
};

state* st(void* user) { return static_cast<state*>(user); }

// Claim a ring slot for a new in-flight op, record its bookkeeping and stamp the
// start time, and return the slot for the caller to redirect the op's context at.
// Returns nullptr if the next slot is still busy with an earlier op that hasn't
// completed (ring full) -- the caller then forwards the op untimed.
op_record* claim(state* s, op_kind kind, void* user_ctx, size_t bytes,
                 struct fid_ep* ep, fi_addr_t peer) {
  auto idx = s->seq.fetch_add(1, std::memory_order_relaxed) % kMaxInflight;
  auto& rec = s->ring[idx];
  bool expected = false;
  if (!rec.busy.compare_exchange_strong(expected, true,
                                        std::memory_order_acquire,
                                        std::memory_order_relaxed)) {
    s->dropped.fetch_add(1, std::memory_order_relaxed);
    return nullptr;
  }
  rec.user_ctx = user_ctx;
  rec.bytes = bytes;
  rec.kind = kind;
  rec.ep = ep;
  rec.peer = peer;
  rec.start = clock_type::now();
  // trecvs stay out of the per-NIC busy/achieved aggregate: their in-flight
  // time is dominated by waiting for the peer to send, not by the NIC working.
  if (kind != OP_TRECV) s->nic_issue(ep, rec.start);
  return &rec;
}

// Undo a claim for an op the provider rejected synchronously (next() < 0): no
// completion will arrive, so free the ring slot and reverse the NIC accounting.
void abort_op(state* s, op_record* rec) {
  if (rec->kind != OP_TRECV) s->nic_unissue(rec->ep);
  rec->busy.store(false, std::memory_order_release);
}

size_t rma_bytes(const struct fi_msg_rma* msg) {
  size_t b = 0;
  for (size_t i = 0; i < msg->rma_iov_count; ++i) b += msg->rma_iov[i].len;
  return b;
}

// Total payload of a tagged-message iovec (fi_tsendmsg).
size_t iov_bytes(const struct iovec* iov, size_t count) {
  size_t b = 0;
  for (size_t i = 0; i < count; ++i) b += iov[i].iov_len;
  return b;
}

// True iff `p` points exactly at one of our ring slots (not just inside the
// array, and not at one-past-the-end). This is how we tell our own redirected
// context apart from a context the application set itself.
bool owned_slot(state* s, void* p) {
  auto* rec = static_cast<op_record*>(p);
  if (rec < s->ring || rec >= s->ring + kMaxInflight) return false;
  auto off = reinterpret_cast<char*>(rec) - reinterpret_cast<char*>(s->ring);
  return off % sizeof(op_record) == 0;
}

/* ---- lifecycle ----------------------------------------------------------- */

void on_ep_bind(struct fid_ep* ep, struct fid* bfid, uint64_t flags,
                void* user) {
  (void)flags;
  auto* s = st(user);
  s->nic.start(s->anchor, s->anchor_wall_ns);  // first fabric activity -> kick off NIC sampling
  if (!bfid || bfid->fclass != FI_CLASS_AV) return;
  std::lock_guard<std::mutex> lk(s->ep_av_mtx);
  s->ep_av[ep] = reinterpret_cast<struct fid_av*>(bfid);
}

/* ---- issue-side shims ---------------------------------------------------- */

ssize_t sh_msg_send(struct fid_ep* ep, const void* buf, size_t len, void* desc,
                    fi_addr_t dest_addr, void* context, lfi_msg_send_fn next,
                    void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_SEND, context, len, ep, dest_addr);
  if (!rec) return next(ep, buf, len, desc, dest_addr, context);
  ssize_t ret = next(ep, buf, len, desc, dest_addr, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_tagged_send(struct fid_ep* ep, const void* buf, size_t len,
                       void* desc, fi_addr_t dest_addr, uint64_t tag,
                       void* context, lfi_tagged_send_fn next, void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_TSEND, context, len, ep, dest_addr);
  if (!rec) return next(ep, buf, len, desc, dest_addr, tag, context);
  ssize_t ret = next(ep, buf, len, desc, dest_addr, tag, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

// fi_tsenddata: like fi_tsend plus a 64-bit immediate (remote CQ data) -- timed
// the same way (it carries a `context` and produces a CQ completion). The extra
// `data` is just threaded through untouched.
ssize_t sh_tagged_senddata(struct fid_ep* ep, const void* buf, size_t len,
                           void* desc, uint64_t data, fi_addr_t dest_addr,
                           uint64_t tag, void* context,
                           lfi_tagged_senddata_fn next, void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_TSENDDATA, context, len, ep, dest_addr);
  if (!rec) return next(ep, buf, len, desc, data, dest_addr, tag, context);
  ssize_t ret = next(ep, buf, len, desc, data, dest_addr, tag, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

// fi_tsendmsg: the context lives in the const fi_msg_tagged, so forward a copy
// with only `context` repointed at our slot (same trick as the rma *msg shims).
// Caveat: a tsendmsg whose flags/endpoint suppress the completion (FI_INJECT or
// selective completion without FI_COMPLETION) would leak its slot -- MPICH's
// large rendezvous path sets FI_COMPLETION, so this is the eager/inline edge.
ssize_t sh_tagged_sendmsg(struct fid_ep* ep, const struct fi_msg_tagged* msg,
                          uint64_t flags, lfi_tagged_sendmsg_fn next,
                          void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_TSENDMSG, msg->context,
                         iov_bytes(msg->msg_iov, msg->iov_count), ep, msg->addr);
  if (!rec) return next(ep, msg, flags);
  struct fi_msg_tagged copy = *msg;
  copy.context = rec;
  ssize_t ret = next(ep, &copy, flags);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_tagged_recv(struct fid_ep* ep, void* buf, size_t len, void* desc,
                       fi_addr_t src_addr, uint64_t tag, uint64_t ignore,
                       void* context, lfi_tagged_recv_fn next, void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_TRECV, context, len, ep, src_addr);
  if (!rec) return next(ep, buf, len, desc, src_addr, tag, ignore, context);
  ssize_t ret = next(ep, buf, len, desc, src_addr, tag, ignore, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_rma_write(struct fid_ep* ep, const void* buf, size_t len, void* desc,
                     fi_addr_t dest_addr, uint64_t addr, uint64_t key,
                     void* context, lfi_rma_write_fn next, void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_WRITE, context, len, ep, dest_addr);
  if (!rec) return next(ep, buf, len, desc, dest_addr, addr, key, context);
  ssize_t ret = next(ep, buf, len, desc, dest_addr, addr, key, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_rma_read(struct fid_ep* ep, void* buf, size_t len, void* desc,
                    fi_addr_t src_addr, uint64_t addr, uint64_t key,
                    void* context, lfi_rma_read_fn next, void* user) {
  auto* s = st(user);
  op_record* rec = claim(s, OP_READ, context, len, ep, src_addr);
  if (!rec) return next(ep, buf, len, desc, src_addr, addr, key, context);
  ssize_t ret = next(ep, buf, len, desc, src_addr, addr, key, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_rma_readv(struct fid_ep* ep, const struct iovec* iov, void** desc,
                     size_t count, fi_addr_t src_addr, uint64_t addr,
                     uint64_t key, void* context, lfi_rma_readv_fn next,
                     void* user) {
  auto* s = st(user);
  op_record* rec =
      claim(s, OP_READV, context, iov_bytes(iov, count), ep, src_addr);
  if (!rec)
    return next(ep, iov, desc, count, src_addr, addr, key, context);
  ssize_t ret = next(ep, iov, desc, count, src_addr, addr, key, rec);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

// The `msg` ops carry context inside a const struct, so we forward a copy with
// only `context` repointed at our slot; the provider's writes go to the copy.
ssize_t sh_rma_writemsg(struct fid_ep* ep, const struct fi_msg_rma* msg,
                        uint64_t flags, lfi_rma_writemsg_fn next, void* user) {
  auto* s = st(user);
  op_record* rec =
      claim(s, OP_WRITEMSG, msg->context, rma_bytes(msg), ep, msg->addr);
  if (!rec) return next(ep, msg, flags);
  struct fi_msg_rma copy = *msg;
  copy.context = rec;
  ssize_t ret = next(ep, &copy, flags);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

ssize_t sh_rma_readmsg(struct fid_ep* ep, const struct fi_msg_rma* msg,
                       uint64_t flags, lfi_rma_readmsg_fn next, void* user) {
  auto* s = st(user);
  op_record* rec =
      claim(s, OP_READMSG, msg->context, rma_bytes(msg), ep, msg->addr);
  if (!rec) return next(ep, msg, flags);
  struct fi_msg_rma copy = *msg;
  copy.context = rec;
  ssize_t ret = next(ep, &copy, flags);
  if (ret < 0) abort_op(s, rec);
  return ret;
}

/* ---- completion side ----------------------------------------------------- */

int sh_cq_open(struct fid_domain* domain, struct fi_cq_attr* attr,
               struct fid_cq** cq, void* context, lfi_cq_open_fn next,
               void* user) {
  int r = next(domain, attr, cq, context);
  if (r == 0 && cq && *cq) {
    auto* s = st(user);
    s->nic.start(s->anchor, s->anchor_wall_ns);  // first fabric activity -> kick off NIC sampling
    {
      std::lock_guard<std::mutex> lk(s->fmt_mtx);
      s->cq_format[*cq] = attr->format;
    }
    if (s->cqgap_on) s->gap_for(*cq);  // pre-register its poll-gap slot
  }
  return r;
}

// Return a pointer to the op_context field of completion entry `c`, given the
// CQ's entry format, or nullptr if the format carries no op_context.
void** op_context_at(fi_cq_format format, void* buf, ssize_t c) {
  switch (format) {
    case FI_CQ_FORMAT_CONTEXT:
      return &static_cast<fi_cq_entry*>(buf)[c].op_context;
    case FI_CQ_FORMAT_MSG:
      return &static_cast<fi_cq_msg_entry*>(buf)[c].op_context;
    case FI_CQ_FORMAT_DATA:
      return &static_cast<fi_cq_data_entry*>(buf)[c].op_context;
    case FI_CQ_FORMAT_TAGGED:
      return &static_cast<fi_cq_tagged_entry*>(buf)[c].op_context;
    case FI_CQ_FORMAT_UNSPEC:
    default:
      return nullptr;
  }
}

// Bytes actually delivered, for CQ formats that carry a length (recv-side
// completions report the received size there; the posted buffer can be bigger).
size_t len_at(fi_cq_format format, void* buf, ssize_t c) {
  switch (format) {
    case FI_CQ_FORMAT_MSG:
      return static_cast<fi_cq_msg_entry*>(buf)[c].len;
    case FI_CQ_FORMAT_DATA:
      return static_cast<fi_cq_data_entry*>(buf)[c].len;
    case FI_CQ_FORMAT_TAGGED:
      return static_cast<fi_cq_tagged_entry*>(buf)[c].len;
    default:
      return 0;
  }
}

// Shared completion-side processing for cq_read and cq_readfrom: time every
// completion whose op_context points at one of our ring slots, then restore
// the app's context before it sees the entry.
void consume(state* s, struct fid_cq* cq, void* buf, ssize_t r) {
  fi_cq_format format;
  {
    std::lock_guard<std::mutex> lk(s->fmt_mtx);
    auto it = s->cq_format.find(cq);
    if (it == s->cq_format.end()) {
      // CQ opened before we registered (or its open was suppressed): we don't
      // know the entry layout, so we can't locate op_context. Hand the
      // completions back untouched.
      return;
    }
    format = it->second;
  }

  auto now = clock_type::now();
  for (ssize_t c = 0; c < r; ++c) {
    void** ctx = op_context_at(format, buf, c);
    if (!ctx || !*ctx || !owned_slot(s, *ctx)) continue;

    auto* rec = static_cast<op_record*>(*ctx);
    double ns =
        std::chrono::duration<double, std::nano>(now - rec->start).count();
    if (rec->kind == OP_TRECV) {
      // Attribute the row to the bytes actually received, and keep recvs out
      // of the NIC busy/achieved aggregate (their latency is mostly idle wait
      // for the sender; see the header comment).
      if (size_t l = len_at(format, buf, c)) rec->bytes = l;
      s->record(rec, ns, /*error=*/false);
    } else {
      s->record(rec, ns, /*error=*/false);
      s->nic_complete(rec->ep, rec->bytes, now, /*ok=*/true);
    }
    *ctx = rec->user_ctx;  // restore the app's original context
    rec->busy.store(false, std::memory_order_release);  // free the slot
  }
}

ssize_t sh_cq_read(struct fid_cq* cq, void* buf, size_t count,
                   lfi_cq_read_fn next, void* user) {
  auto* s = st(user);
  s->poll_mark(cq);
  ssize_t r = next(cq, buf, count);
  if (r > 0) consume(s, cq, buf, r);
  return r;
}

ssize_t sh_cq_readfrom(struct fid_cq* cq, void* buf, size_t count,
                       fi_addr_t* src_addr, lfi_cq_readfrom_fn next,
                       void* user) {
  auto* s = st(user);
  s->poll_mark(cq);
  ssize_t r = next(cq, buf, count, src_addr);
  if (r > 0) consume(s, cq, buf, r);
  return r;
}

// Error completions arrive here (after cq_read returns -FI_EAVAIL). The error
// entry's op_context is our redirected slot; free it, count the error, and
// restore the app's context before it sees the entry.
ssize_t sh_cq_readerr(struct fid_cq* cq, struct fi_cq_err_entry* buf,
                      uint64_t flags, lfi_cq_readerr_fn next, void* user) {
  ssize_t r = next(cq, buf, flags);
  if (r <= 0 || !buf) return r;

  auto* s = st(user);
  if (buf->op_context && owned_slot(s, buf->op_context)) {
    auto* rec = static_cast<op_record*>(buf->op_context);
    s->record(rec, 0.0, /*error=*/true);
    if (rec->kind != OP_TRECV)
      s->nic_complete(rec->ep, rec->bytes, clock_type::now(), /*ok=*/false);
    buf->op_context = rec->user_ctx;
    rec->busy.store(false, std::memory_order_release);
  }
  return r;
}

}  // namespace

__attribute__((constructor)) static void install(void) {
  // Function-local static: lives for the whole process and is destroyed at exit,
  // which is where ~state() prints the aggregate timing report.
  static std::unique_ptr<state> g_state = std::make_unique<state>();

  struct lfi_hooks h = {};
  h.user = g_state.get();

  h.on_ep_bind = on_ep_bind;

  h.msg_send = sh_msg_send;
  h.tagged_send = sh_tagged_send;
  h.tagged_senddata = sh_tagged_senddata;
  h.tagged_sendmsg = sh_tagged_sendmsg;
  h.tagged_recv = sh_tagged_recv;
  h.rma_write = sh_rma_write;
  h.rma_writemsg = sh_rma_writemsg;
  h.rma_read = sh_rma_read;
  h.rma_readv = sh_rma_readv;
  h.rma_readmsg = sh_rma_readmsg;

  h.cq_open = sh_cq_open;
  h.cq_read = sh_cq_read;
  h.cq_readfrom = sh_cq_readfrom;
  h.cq_readerr = sh_cq_readerr;
  lfi_register(&h);
}