net_ib.cpp

#include <assert.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <poll.h>
#include <sys/types.h>
#include <unistd.h>
#include <netdb.h>

#include "net_ib.h"
#include "rocm_wrap.h"
#include "base.h"

namespace sccl {
namespace hardware {
namespace net {
namespace net_ib {

///////////////////////////////////////// 环境变量读取及设置 /////////////////////////////////////////
// 定义InfiniBand GID索引，默认值为0
SCCL_PARAM(IbGidIndex, "IB_GID_INDEX", 0);
// 定义InfiniBand超时时间，默认值为18
SCCL_PARAM(IbTimeout, "IB_TIMEOUT", 18);
// 定义InfiniBand重试次数，默认值为7
SCCL_PARAM(IbRetryCnt, "IB_RETRY_CNT", 7);
// 定义InfiniBand分区密钥，默认值为0
SCCL_PARAM(IbPkey, "IB_PKEY", 0);
// 定义是否使用InfiniBand内联传输，默认值为0（不使用）
SCCL_PARAM(IbUseInline, "IB_USE_INLINE", 0);
// 定义InfiniBand服务级别，默认值为0
SCCL_PARAM(IbSl, "IB_SL", 0);
// 定义InfiniBand流量类别，默认值为0
SCCL_PARAM(IbTc, "IB_TC", 0);
// 定义InfiniBand自动路由阈值，默认值为8192
SCCL_PARAM(IbArThreshold, "IB_AR_THRESHOLD", 8192);
// 定义InfiniBand PCI宽松排序选项，默认值为2
SCCL_PARAM(IbPciRelaxedOrdering, "IB_PCI_RELAXED_ORDERING", 2);
// 定义是否启用InfiniBand自适应路由，默认值为-2（可能表示禁用或默认设置）
SCCL_PARAM(IbAdaptiveRouting, "IB_ADAPTIVE_ROUTING", -2);
// 定义InfiniBand套接字客户端端口重用选项，默认值为0（不重用）
SCCL_PARAM(IbSockClientPortReuse, "IB_SOCK_CLIENT_PORT_REUSE", 0);
// 定义InfiniBand套接字服务器端口重用选项，默认值为0（不重用）
SCCL_PARAM(IbSockServerPortReuse, "IB_SOCK_SERVER_PORT_REUSE", 0);
// 定义是否禁用InfiniBand，默认值为0（不禁用）
SCCL_PARAM(IbDisable, "IB_DISABLE", 0);
// 定义是否合并InfiniBand虚拟功能，默认值为1（合并）
SCCL_PARAM(IbMergeVfs, "IB_MERGE_VFS", 1);
// 定义每个连接的InfiniBand队列对(QP)数量，默认值为1
SCCL_PARAM(IbQpsPerConn, "IB_QPS_PER_CONNECTION", 1);
// 定义是否禁用GDR刷新，默认值为0（不禁用）
SCCL_PARAM(IbGdrFlushDisable, "GDR_FLUSH_DISABLE", 0);
// 定义是否在队列对上分割数据，默认值为1（分割）
SCCL_PARAM(IbSplitDataOnQps, "IB_SPLIT_DATA_ON_QPS", 1);

///////////////////////////////////////// 参数及结构体设置 /////////////////////////////////////////
constexpr int MAXNAMESIZE = 64;
static char scclIbIfName[MAX_IF_NAME_SIZE + 1];          // 用于存储网络接口名称的字符数组
static union net_socket::scclSocketAddress scclIbIfAddr; // 定义一个联合体类型的变量，用于存储网络接口地址

struct scclIbMr {
    uintptr_t addr; // 内存地址
    int pages;      // 页数
    int refs;       // 引用计数
    ibv_mr* mr;     // InfiniBand内存注册对象指针
};

// 结构体用于缓存 InfiniBand 内存注册对象
struct scclIbMrCache {
    struct scclIbMr* slots; // 缓存槽，用于存储内存注册对象
    int capacity;
    int population; // 缓存的容量和当前已填充的数量
};

// 定义一个对齐到 64 字节边界的结构体 scclIbDev，用于表示 InfiniBand 设备
struct alignas(64) scclIbDev {
    pthread_mutex_t lock;         // 互斥锁，用于线程同步
    int device;                   // 设备编号
    uint64_t guid;                // 全局唯一标识符
    uint8_t port;                 // 端口号
    uint8_t link;                 // 链路层信息
    int speed;                    // 传输速度
    ibv_context* context;         // InfiniBand 上下文
    int pdRefs;                   // 保护域引用计数
    ibv_pd* pd;                   // 保护域
    char devName[MAXNAMESIZE];    // 设备名称
    char* pciPath;                // PCI 路径
    int realPort;                 // 实际使用的端口
    int maxQp;                    // 最大队列对数量
    struct scclIbMrCache mrCache; // 内存注册对象缓存
    int ar;                       // ADAPTIVE_ROUTING，自适应路由标志
};

struct userIbDev {
    char devName[MAXNAMESIZE];
    uint16_t port_en;
};

// 定义最大InfiniBand设备数量为16
static constexpr int MAX_IB_DEVS = 16;
// 定义一个结构体数组，用于存储InfiniBand设备信息
struct scclIbDev scclIbDevs[MAX_IB_DEVS];
// 定义一个结构体数组，用于存储用户级别的InfiniBand设备信息
struct userIbDev userIbDevs[MAX_IB_DEVS];

// 定义一个互斥锁，用于保护对InfiniBand设备的并发访问
pthread_mutex_t scclIbLock = PTHREAD_MUTEX_INITIALIZER;

// 定义一个静态整数，用于指示是否启用了InfiniBand的Relaxed Ordering模式
static int scclIbRelaxedOrderingEnabled = 0;

// 定义一个线程局部变量，用于存储重用的地址信息
static thread_local union net_socket::scclSocketAddress reusedAddr;

// 定义一个线程局部变量，用于存储重用的套接字文件描述符
static thread_local int reusedSockfd = -1;

// 定义一个线程ID，用于异步线程操作
pthread_t scclIbAsyncThread;

// 定义一个常量，表示InfiniBand网络接口的最大接收数量
static constexpr int SCCL_NET_IB_MAX_RECVS = 8;
// 定义一个常量，表示最大字符串长度
static constexpr int MAX_STR_LEN = 255;

// 为每个并发接收支持SCCL_NET_MAX_REQUESTS
static constexpr int MAX_REQUESTS = (SCCL_NET_MAX_REQUESTS * SCCL_NET_IB_MAX_RECVS);
static_assert(MAX_REQUESTS <= 256, "request id are encoded in wr_id and we need up to 8 requests ids per completion");

// Retain local and remote RoCE addresses for error logging
struct scclIbGidInfo {
    uint8_t link_layer;      // 链路层类型，表示网络连接的物理层类型
    union ibv_gid localGid;  // 本地设备的全局标识符（GID）
    union ibv_gid remoteGid; // 远程设备的全局标识符（GID）
};

/*
scclIbRequest 结构体用于封装 InfiniBand 通信请求的详细信息，包括通信接口、请求类型、数据缓冲区等。
联合体 union 根据请求类型（发送或接收）存储不同的数据结构，以支持灵活的通信操作。
*/
struct scclIbRequest {
    struct scclIbVerbs* verbs;           // 指向 scclIbVerbs 结构体的指针，包含 Infiniband 相关的操作
    int type;                            // 请求的类型，例如发送或接收
    int events;                          // 事件标志， 用于记录请求相关的事件状态
    struct net_socket::scclSocket* sock; // 指向 scclSocket 结构体的指针，表示网络套接字
    struct scclIbGidInfo* gidInfo;       // 指向 scclIbGidInfo 结构体的指针，包含全局标识符信息
    int nreqs;                           // 请求的数量
    // 联合体，用于存储不同类型请求的特定信息
    union {
        // send: 发送请求的相关信息
        struct {
            int size;      // 发送数据的大小
            void* data;    // 指向发送数据的指针
            uint32_t lkey; // 本地密钥，用于数据访问
            int offset;    // 数据偏移量
        } send;
        // recv: 接收请求的相关信息
        struct {
            int sizes[SCCL_NET_IB_MAX_RECVS]; // 接收数据的大小数组，最多包含 SCCL_NET_IB_MAX_RECVS 个元素
        } recv;
    };
};

/*用于封装 InfiniBand 通信所需的资源，便于管理和复用。*/
struct scclIbVerbs {
    int dev;                                 // 设备索引，标识使用的 InfiniBand 设备
    struct ibv_pd* pd;                       // 指向 InfiniBand 保护域（Protection Domain）的指针，用于内存注册和队列管理
    struct ibv_cq* cq;                       // 指向 InfiniBand 完成队列（Completion Queue）的指针，用于跟踪异步操作的状态
    uint64_t pad[1];                         // 填充字段，可能用于内存对齐或未来扩展
    struct scclIbRequest reqs[MAX_REQUESTS]; // 存储最大请求数（MAX_REQUESTS）的请求结构体数组
};

/*用于 InfiniBand 通信的发送队列（FIFO），存储待发送数据的元信息，供底层网络驱动或通信库使用*/
struct alignas(64) scclIbSendFifo {
    uint64_t addr;  // 目标内存地址（远程地址）
    int size;       // 发送数据的大小（字节）
    uint32_t rkey;  // 远程密钥（Remote Key），用于 InfiniBand 的远程内存访问（RMA）
    uint32_t nreqs; // 发送请求的数量（可能用于批量操作）
    uint32_t tag;   // 标签或标识符，用于区分不同的发送操作
    uint64_t idx;   // 索引值，可能用于跟踪或管理发送队列中的位置
};

static constexpr int SCCL_IB_MAX_QPS = 128; // 最大队列对数量
struct scclIbSendComm {
    struct scclIbVerbs verbs;                                            // RDMA verbs结构体
    struct scclIbSendFifo fifo[MAX_REQUESTS][SCCL_NET_IB_MAX_RECVS];     // 发送FIFO队列
    uint64_t fifoHead;                                                   // FIFO队列头指针
    struct scclIbRequest* fifoReqs[MAX_REQUESTS][SCCL_NET_IB_MAX_RECVS]; // FIFO请求指针数组
    struct ibv_send_wr wrs[SCCL_NET_IB_MAX_RECVS + 1];                   // 发送工作请求结构体数组
    struct ibv_sge sges[SCCL_NET_IB_MAX_RECVS];                          // 散布-聚集元素结构体数组
    struct net_socket::scclSocket sock;                                  // 套接字结构体

    int ready;                           // 是否准备好
    struct ibv_qp* qps[SCCL_IB_MAX_QPS]; // 队列对指针数组
    int nqps;                            // 队列对数量
    int qpIndex;                         // 当前队列对索引
    struct ibv_mr* fifoMr;               // FIFO内存区域指针
    int ar;                              // 自动重发标志
    struct scclIbGidInfo gidInfo;        // GID信息结构体
};

struct scclIbQpInfo {
    uint32_t lid;
    uint8_t ib_port;
    uint8_t link_layer;
    uint32_t qpn[SCCL_IB_MAX_QPS];

    // For RoCE
    uint64_t spn;
    uint64_t iid;
    enum ibv_mtu mtu;

    // FIFO RDMA info
    uint32_t fifoRkey;
    uint64_t fifoAddr;
};

struct scclIbGpuFlush {
    int enabled;
    int hostMem;
    struct ibv_mr* hostMr;
    struct ibv_sge sge;
    struct ibv_qp* qp;
};

struct scclIbRemFifo {
    struct scclIbSendFifo elems[MAX_REQUESTS][SCCL_NET_IB_MAX_RECVS];
    uint64_t fifoTail;
    uint64_t addr;
    uint32_t rkey;
    uint32_t flags;
    struct ibv_mr* mr;
    struct ibv_sge sge;
};

struct scclIbRecvComm {
    struct scclIbVerbs verbs;
    struct scclIbRemFifo remFifo;
    struct net_socket::scclSocket sock;
    int ready;
    struct ibv_qp* qps[SCCL_IB_MAX_QPS];
    int nqps;
    int qpIndex;
    struct scclIbGpuFlush gpuFlush;
    struct scclIbGidInfo gidInfo;
};
static_assert((offsetof(struct scclIbRecvComm, remFifo) % 32) == 0, "scclIbSendComm fifo must be 32-byte aligned");

///////////////////////////////////////// net_ib的函数 /////////////////////////////////////////

/**
 * @brief IB异步事件处理线程主函数
 *
 * 该函数作为独立线程运行，持续监听并处理IB设备的异步事件。
 * 对于每个接收到的异步事件（除IBV_EVENT_COMM_EST外），会输出警告日志。
 * 处理完成后必须调用wrap_ibv_ack_async_event进行事件确认。
 *
 * @param args 传入参数，应转换为ibv_context结构体指针
 * @return void* 线程返回值，始终返回NULL
 */
void* scclNetIb::scclIbAsyncThreadMain(void* args) {
    // 将传入的参数转换为InfiniBand上下文结构体指针
    struct ibv_context* context = (struct ibv_context*)args;

    // 无限循环，持续监听异步事件
    while(1) {
        // 定义一个结构体来存储异步事件
        struct ibv_async_event event;
        // 调用封装的函数获取异步事件，如果获取失败则退出循环
        if(scclSuccess != wrap_ibv_get_async_event(context, &event)) {
            break;
        }

        // 定义一个字符指针用于存储事件类型的字符串描述
        char* str;
        // 调用封装的函数将事件类型转换为字符串，如果转换失败则退出循环
        if(scclSuccess != wrap_ibv_event_type_str(&str, event.event_type)) {
            break;
        }

        // 如果事件类型不是通信建立事件，则输出警告信息
        if(event.event_type != IBV_EVENT_COMM_EST)
            WARN("NET/IB : Got async event : %s", str);

        // 调用封装的函数确认（acknowledge）异步事件，如果确认失败则退出循环
        if(scclSuccess != wrap_ibv_ack_async_event(&event)) {
            break;
        }
    }

    // 线程结束，返回NULL
    return NULL;
}

/**
 * @brief 获取IB设备的PCI路径并处理多端口和虚拟功能合并
 *
 * 该函数通过设备名称获取IB设备的真实PCI路径，并对多端口NIC和虚拟功能(VF)进行合并处理，
 * 将它们视为同一PCI设备。同时记录实际端口号。
 *
 * @param devName 输入参数，IB设备名称
 * @param path 输出参数，存储获取到的PCI路径
 * @param realPort 输出参数，记录实际端口号
 * @return scclResult_t 返回操作结果，成功返回scclSuccess
 */
scclResult_t scclNetIb::scclIbGetPciPath(char* devName, char** path, int* realPort) {
    // 定义一个字符数组用于存储设备路径
    char devicePath[PATH_MAX];
    // 构造设备路径字符串，格式为 "/sys/class/infiniband/<devName>/device"
    snprintf(devicePath, PATH_MAX, "/sys/class/infiniband/%s/device", devName);
    // 获取设备路径的绝对路径
    char* p = realpath(devicePath, NULL);
    if(p == NULL) {
        // 如果无法获取绝对路径，记录警告信息
        WARN("Could not find real path of %s (%s)", devName, devicePath);
    } else {
        // 处理多端口 NIC（网络接口卡），将路径末尾的端口编号替换为 '0'
        p[strlen(p) - 1] = '0';
        // 如果启用了虚拟函数（VF）合并，则将路径中倒数第3和第4字符替换为 '0'
        if(scclParamIbMergeVfs())
            p[strlen(p) - 3] = p[strlen(p) - 4] = '0';
        // 初始化 realPort 为 0，用于统计实际端口数量
        *realPort = 0;
        // 遍历已知的 InfiniBand 设备列表
        for(int d = 0; d < scclNIbDevs; d++) {
            // 如果当前路径与已知的设备 PCI 路径匹配，则增加实际端口计数
            if(strcmp(p, scclIbDevs[d].pciPath) == 0)
                (*realPort)++;
        }
    }
    // 将计算得到的绝对路径赋值给输出参数 path
    *path = p;
    // 返回成功状态
    return scclSuccess;
}

static int ibvWidths[] = {1, 4, 8, 12, 2};
static int ibvSpeeds[] = {2500,  /* SDR */
                          5000,  /* DDR */
                          10000, /* QDR */
                          10000, /* QDR */
                          14000, /* FDR */
                          25000, /* EDR */
                          50000, /* HDR */
                          100000 /* NDR */};

/**
 * 查找第一个被设置的bit位
 * @param val 要检查的整数值
 * @param max 最大检查位数
 * @return 第一个被设置的bit位索引，若未找到则返回max
 */
static int firstBitSet(int val, int max) {
    int i = 0;
    while(i < max && ((val & (1 << i)) == 0))
        i++;
    return i;
}

/**
 * 根据输入的宽度值，返回对应的IB(InfiniBand)链路宽度索引
 * @param width 输入的宽度值
 * @return 返回ibvWidths数组中对应的宽度索引值
 */
int scclNetIb::scclIbWidth(int width) { return ibvWidths[firstBitSet(width, sizeof(ibvWidths) / sizeof(int) - 1)]; }

/**
 * 根据给定的速度值查找并返回对应的IB传输速率
 * @param speed 输入的速度值
 * @return 返回ibvSpeeds数组中第一个匹配的IB传输速率
 */
int scclNetIb::scclIbSpeed(int speed) { return ibvSpeeds[firstBitSet(speed, sizeof(ibvSpeeds) / sizeof(int) - 1)]; }

/**
 * 检查当前IB设备是否支持宽松排序(Relaxed Ordering)模式
 *
 * @return 1表示支持，0表示不支持
 * @note 通过查询IBVERBS_1.8 API的ibv_reg_mr_iova2函数来检测IBV_ACCESS_RELAXED_ORDERING支持
 * @see scclParamIbPciRelaxedOrdering() 获取当前配置的RO模式
 */
int scclNetIb::scclIbRelaxedOrderingCapable(void) {
    int roMode     = scclParamIbPciRelaxedOrdering();
    scclResult_t r = scclInternalError;
    if(roMode == 1 || roMode == 2) {
        // Query IBVERBS_1.8 API - needed for IBV_ACCESS_RELAXED_ORDERING support
        r = wrap_ibv_reg_mr_iova2(NULL, NULL, NULL, 0, 0, 0);
    }
    return r == scclInternalError ? 0 : 1;
}

/**
 * @brief 获取并处理用户指定的IB设备环境变量
 *
 * 该函数检查并处理环境变量SCCL_IB_HCA的值，支持以下特殊前缀：
 * - '^' 表示反向匹配
 * - '=' 表示精确匹配
 *
 * @param shownIbHcaEnv 计数器，用于控制日志输出次数
 * @return char* 处理后的IB设备环境变量值
 */
char* scclNetIb::scclIbGetIbHca(int& shownIbHcaEnv, bool* searchNot, bool* searchExact) {
    // 检查用户是否定义了要使用的IB设备:端口
    char* userIbEnv = getenv("SCCL_IB_HCA");
    if(userIbEnv != NULL && shownIbHcaEnv++ == 0)
        INFO(SCCL_LOG_NET, "SCCL_IB_HCA set to %s", userIbEnv);

    *searchNot = userIbEnv && userIbEnv[0] == '^';
    if(*searchNot)
        userIbEnv++;

    *searchExact = userIbEnv && userIbEnv[0] == '=';
    if(*searchExact)
        userIbEnv++;

    return userIbEnv;
}

/**
 * @brief 从系统文件中读取字符串内容
 *
 * 该函数通过拼接路径和文件名，打开指定文件并读取其内容到字符串缓冲区中。
 * 如果读取失败或文件为空，会将缓冲区置为空字符串并记录警告信息。
 *
 * @param path 文件所在目录路径
 * @param fileName 要读取的文件名
 * @param strValue 用于存储读取内容的字符串缓冲区
 * @return scclResult_t 始终返回scclSuccess
 *
 * @note 缓冲区最大长度为MAX_STR_LEN，超出部分会被截断
 *       文件内容末尾会自动添加字符串结束符'\0'
 */
scclResult_t scclNetIb::scclGetStrFromSys(const char* path, const char* fileName, char* strValue) {
    char filePath[PATH_MAX];
    sprintf(filePath, "%s/%s", path, fileName);
    int offset = 0;
    FILE* file;
    if((file = fopen(filePath, "r")) != NULL) {
        while(feof(file) == 0 && ferror(file) == 0 && offset < MAX_STR_LEN) {
            int len = fread(strValue + offset, 1, MAX_STR_LEN - offset, file);
            offset += len;
        }
        fclose(file);
    }
    if(offset == 0) {
        strValue[0] = '\0';
        INFO(SCCL_LOG_NET, "System detection : could not read %s, ignoring", filePath);
    } else {
        strValue[offset - 1] = '\0';
    }
    return scclSuccess;
}

/**
 * @brief 检查IB设备是否支持GPU Direct RDMA (GDR)
 *
 * 该函数用于检测当前系统环境是否支持GPU Direct RDMA功能。
 * 在HIP平台下会检查内核模块加载状态、BIOS版本和NUMA平衡设置，
 * 其他平台默认不支持。
 *
 * @param ibDev IB设备号
 * @return scclResult_t 返回scclSuccess表示支持，返回scclSystemError表示不支持
 */
scclResult_t scclNetIb::scclIbGdrSupport(int ibDev) {
    static int moduleLoaded = -1;
    if(moduleLoaded == -1) {
#if defined(__HIP_PLATFORM_HCC__) || defined(__HCC__) || defined(__HIPCC__)
        moduleLoaded = (access("/sys/kernel/mm/memory_peers/amdkfd/version", F_OK) == -1) ? 0 : 1;
        char strValue[MAX_STR_LEN];
        SCCLCHECK(scclGetStrFromSys("/sys/devices/virtual/dmi/id", "bios_version", strValue));
        if(strncmp("Hyper-V UEFI Release", strValue, 20) == 0) {
            int roMode = scclParamIbPciRelaxedOrdering();
            SCCLCHECK(scclGetStrFromSys("/proc/sys/kernel", "numa_balancing", strValue));
            if(strcmp(strValue, "1") == 0 && roMode == 0)
                moduleLoaded = 0;
        }
#else
        moduleLoaded = 0;
#endif
    }
    if(moduleLoaded == 0)
        return scclSystemError;

    return scclSuccess;
}

/**
 * @brief 检查设备是否支持DMA-BUF功能
 *
 * 该函数用于检测指定IB设备是否支持DMA-BUF内存注册功能。
 * 通过尝试注册一个无效的DMA-BUF文件描述符来测试支持性。
 * 结果会被缓存以避免重复检测。
 *
 * @param dev 设备索引
 * @return scclResult_t 返回scclSuccess表示支持，scclSystemError表示不支持
 */
scclResult_t scclNetIb::scclIbDmaBufSupport(int dev) {
    static int dmaBufSupported = -1;
    if(dmaBufSupported == -1) {
        scclResult_t res;
        SCCLCHECKGOTO(rocmLibraryInit(), res, failure);
        struct ibv_pd* pd;
        struct ibv_context* ctx;
        ctx = scclIbDevs[dev].context;
        SCCLCHECKGOTO(wrap_ibv_alloc_pd(&pd, ctx), res, failure);
        // Test kernel DMA-BUF support with a dummy call (fd=-1)
        (void)wrap_direct_ibv_reg_dmabuf_mr(pd, 0ULL /*offset*/, 0ULL /*len*/, 0ULL /*iova*/, -1 /*fd*/, 0 /*flags*/);
        // ibv_reg_dmabuf_mr() will fail with EOPNOTSUPP/EPROTONOSUPPORT if not supported (EBADF otherwise)
        dmaBufSupported = (errno != EOPNOTSUPP && errno != EPROTONOSUPPORT) ? 1 : 0;
        SCCLCHECKGOTO(wrap_ibv_dealloc_pd(pd), res, failure);
    }
    if(dmaBufSupported == 0)
        return scclSystemError;
    return scclSuccess;

failure:
    dmaBufSupported = 0;
    return scclSystemError;
}

struct scclIbHandle {
    union net_socket::scclSocketAddress connectAddr; // Filled by the target (目标填充)
    uint64_t magic;                                  // random number to help debugging (用于调试的随机数)
    struct scclIbCommStage stage;                    // Used by the other side when connecting (连接时由另一侧使用)
};

/**
 * @brief 初始化InfiniBand Verbs资源
 *
 * 该函数用于初始化指定设备的InfiniBand Verbs资源，包括：
 * - 分配保护域(PD)
 * - 创建完成队列(CQ)
 *
 * @param dev 设备索引
 * @param ctx IB设备上下文
 * @param verbs 要初始化的Verbs结构体指针
 * @return scclResult_t 返回操作结果，scclSuccess表示成功
 *
 * @note 该函数会递增设备的PD引用计数，并在首次调用时为设备分配PD
 * @note 创建的CQ大小为2*MAX_REQUESTS*IB_QPS_PER_CONNECTION，以支持接收请求的双重完成
 */
scclResult_t scclNetIb::scclIbInitVerbs(int dev, struct ibv_context* ctx, struct scclIbVerbs* verbs) {
    verbs->dev = dev;

    pthread_mutex_lock(&scclIbDevs[dev].lock);
    if(0 == scclIbDevs[dev].pdRefs++) {
        scclResult_t res;
        SCCLCHECKGOTO(wrap_ibv_alloc_pd(&scclIbDevs[dev].pd, ctx), res, failure);
        if(0) {
        failure:
            pthread_mutex_unlock(&scclIbDevs[dev].lock);
            return res;
        }
    }
    verbs->pd = scclIbDevs[dev].pd;
    pthread_mutex_unlock(&scclIbDevs[dev].lock);

    // Recv requests can generate 2 completions (one for the post FIFO, one for the Recv).
    SCCLCHECK(wrap_ibv_create_cq(&verbs->cq, ctx, 2 * MAX_REQUESTS * scclParamIbQpsPerConn(), NULL, NULL, 0));
    return scclSuccess;
}

/**
 * 创建并初始化一个InfiniBand队列对(QP)
 *
 * @param ib_port IB端口号
 * @param verbs IB verbs结构体指针
 * @param access_flags QP访问权限标志
 * @param qp 输出的QP指针
 *
 * @return 返回scclSuccess表示成功，否则返回错误码
 *
 * @note QP类型为可靠连接(RC)，发送队列大小为2*MAX_REQUESTS，
 *       接收队列大小为MAX_REQUESTS，支持内联数据发送(如果配置启用)
 */
scclResult_t scclNetIb::scclIbCreateQp(uint8_t ib_port, struct scclIbVerbs* verbs, int access_flags, struct ibv_qp** qp) {
    struct ibv_qp_init_attr qpInitAttr;
    memset(&qpInitAttr, 0, sizeof(struct ibv_qp_init_attr));
    qpInitAttr.send_cq = verbs->cq;
    qpInitAttr.recv_cq = verbs->cq;
    qpInitAttr.qp_type = IBV_QPT_RC;
    // We might send 2 messages per send (RDMA and RDMA_WITH_IMM)
    qpInitAttr.cap.max_send_wr     = 2 * MAX_REQUESTS;
    qpInitAttr.cap.max_recv_wr     = MAX_REQUESTS;
    qpInitAttr.cap.max_send_sge    = 1;
    qpInitAttr.cap.max_recv_sge    = 1;
    qpInitAttr.cap.max_inline_data = scclParamIbUseInline() ? sizeof(struct scclIbSendFifo) : 0;
    SCCLCHECK(wrap_ibv_create_qp(qp, verbs->pd, &qpInitAttr));
    struct ibv_qp_attr qpAttr;
    memset(&qpAttr, 0, sizeof(struct ibv_qp_attr));
    qpAttr.qp_state        = IBV_QPS_INIT;
    qpAttr.pkey_index      = scclParamIbPkey();
    qpAttr.port_num        = ib_port;
    qpAttr.qp_access_flags = access_flags;
    SCCLCHECK(wrap_ibv_modify_qp(*qp, &qpAttr, IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT | IBV_QP_ACCESS_FLAGS));
    return scclSuccess;
}

/**
 * 将IB QP状态修改为RTR（Ready to Receive）状态
 *
 * @param qp IB QP指针
 * @param qpn 目标QP号
 * @param info QP配置信息，包含MTU、链路层类型、端口号等参数
 *
 * @return 成功返回scclSuccess，失败返回错误码
 *
 * @note 根据链路层类型(以太网/IB)设置不同的AH属性
 *       以太网需要设置全局路由头(GRH)相关参数
 *       IB链路需要设置目标LID
 */
scclResult_t scclNetIb::scclIbRtrQp(struct ibv_qp* qp, uint32_t qpn, struct scclIbQpInfo* info) {
    struct ibv_qp_attr qpAttr;
    memset(&qpAttr, 0, sizeof(struct ibv_qp_attr));
    qpAttr.qp_state           = IBV_QPS_RTR;
    qpAttr.path_mtu           = info->mtu;
    qpAttr.dest_qp_num        = qpn;
    qpAttr.rq_psn             = 0;
    qpAttr.max_dest_rd_atomic = 1;
    qpAttr.min_rnr_timer      = 12;
    if(info->link_layer == IBV_LINK_LAYER_ETHERNET) {
        qpAttr.ah_attr.is_global                     = 1;
        qpAttr.ah_attr.grh.dgid.global.subnet_prefix = info->spn;
        qpAttr.ah_attr.grh.dgid.global.interface_id  = info->iid;
        qpAttr.ah_attr.grh.flow_label                = 0;
        qpAttr.ah_attr.grh.sgid_index                = scclParamIbGidIndex();
        qpAttr.ah_attr.grh.hop_limit                 = 255;
        qpAttr.ah_attr.grh.traffic_class             = scclParamIbTc();
    } else {
        qpAttr.ah_attr.is_global = 0;
        qpAttr.ah_attr.dlid      = info->lid;
    }
    qpAttr.ah_attr.sl            = scclParamIbSl();
    qpAttr.ah_attr.src_path_bits = 0;
    qpAttr.ah_attr.port_num      = info->ib_port;
    SCCLCHECK(wrap_ibv_modify_qp(
        qp, &qpAttr, IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN | IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER));
    return scclSuccess;
}

/**
 * 将IB(InfiniBand)队列对(QP)状态修改为RTS(Ready To Send)状态
 *
 * @param qp IB队列对指针
 * @return 成功返回scclSuccess，失败返回错误码
 *
 * 该函数配置QP属性并调用ibv_modify_qp将其状态改为RTS状态，
 * 设置了超时时间、重试次数、RNR重试次数、SQ PSN和最大RD原子操作数等参数。
 */
scclResult_t scclNetIb::scclIbRtsQp(struct ibv_qp* qp) {
    struct ibv_qp_attr qpAttr;
    memset(&qpAttr, 0, sizeof(struct ibv_qp_attr));
    qpAttr.qp_state      = IBV_QPS_RTS;
    qpAttr.timeout       = scclParamIbTimeout();
    qpAttr.retry_cnt     = scclParamIbRetryCnt();
    qpAttr.rnr_retry     = 7;
    qpAttr.sq_psn        = 0;
    qpAttr.max_rd_atomic = 1;
    SCCLCHECK(wrap_ibv_modify_qp(qp, &qpAttr, IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC));
    return scclSuccess;
}

typedef enum : int {
    SCCL_NET_IB_REQ_UNUSED = 0,
    SCCL_NET_IB_REQ_SEND   = 1,
    SCCL_NET_IB_REQ_RECV   = 2,
    SCCL_NET_IB_REQ_FLUSH  = 3
} NetIbReq_t;
const char* reqTypeStr[] = {"Unused", "Send", "Recv", "Flush"};

// The SendFifo needs to be 32-byte aligned and each element needs
// to be a 32-byte multiple, so that an entry does not get split and
// written out of order when IB Relaxed Ordering is enabled
static_assert((offsetof(struct scclIbSendComm, fifo) % 32) == 0, "scclIbSendComm fifo must be 32-byte aligned");
static_assert((sizeof(struct scclIbSendFifo) % 32) == 0, "scclIbSendFifo element size must be 32-byte multiples");

/**
 * @brief 销毁IB Verbs资源
 *
 * 释放指定的IB Verbs资源，包括完成队列(CQ)和保护域(PD)。
 * 当PD的引用计数减至0时，会自动释放PD资源。
 * 该函数是线程安全的，使用互斥锁保护共享资源。
 *
 * @param verbs 指向要销毁的IB Verbs结构体
 * @return scclResult_t 返回操作结果，scclSuccess表示成功
 */
scclResult_t scclNetIb::scclIbDestroyVerbs(struct scclIbVerbs* verbs) {
    scclResult_t res;
    SCCLCHECK(wrap_ibv_destroy_cq(verbs->cq));

    pthread_mutex_lock(&scclIbDevs[verbs->dev].lock);
    if(0 == --scclIbDevs[verbs->dev].pdRefs) {
        SCCLCHECKGOTO(wrap_ibv_dealloc_pd(scclIbDevs[verbs->dev].pd), res, returning);
    }
    res = scclSuccess;
returning:
    pthread_mutex_unlock(&scclIbDevs[verbs->dev].lock);
    return res;
}

/**
 * @brief 从verbs请求池中获取一个未使用的请求结构体
 *
 * @param verbs 指向scclIbVerbs结构体的指针，包含请求池
 * @param req 输出参数，用于返回获取到的请求结构体指针
 * @return scclResult_t 成功返回scclSuccess，失败返回scclInternalError
 *
 * 该函数遍历verbs请求池，查找第一个未使用的请求(SCCL_NET_IB_REQ_UNUSED)，
 * 初始化其字段后返回。如果所有请求都在使用中，则返回错误。
 */
scclResult_t scclNetIb::scclIbGetRequest(struct scclIbVerbs* verbs, struct scclIbRequest** req) {
    for(int i = 0; i < MAX_REQUESTS; i++) {
        struct scclIbRequest* r = verbs->reqs + i;
        if(r->type == SCCL_NET_IB_REQ_UNUSED) {
            r->verbs   = verbs;
            r->events  = 1;
            r->sock    = NULL;
            r->gidInfo = NULL;
            *req       = r;
            return scclSuccess;
        }
    }
    WARN("NET/IB : unable to allocate requests");
    *req = NULL;
    return scclInternalError;
}

/**
 * 释放IB网络请求资源。
 *
 * 将请求类型标记为未使用状态，但不实际释放内存。
 *
 * @param r 要释放的IB网络请求指针
 * @return 总是返回scclSuccess表示操作成功
 */
scclResult_t scclNetIb::scclIbFreeRequest(struct scclIbRequest* r) {
    r->type = SCCL_NET_IB_REQ_UNUSED;
    return scclSuccess;
}

/**
 * @brief 执行IB网络的多发送操作
 *
 * 该函数处理IB网络的多发送请求，包括设置发送工作请求(WR)和分散/聚集元素(SGE)，
 * 并处理自适应路由(AR)和QP分割等高级功能。
 *
 * @param comm 指向scclIbSendComm结构的指针，包含发送通信上下文
 * @param slot 要使用的发送槽位索引
 * @return scclResult_t 返回操作结果，成功返回scclSuccess，失败返回错误码
 *
 * @note 1. 支持多QP分割发送，确保128B对齐
 *       2. 使用RDMA_WRITE_WITH_IMM发送立即数据
 *       3. 当请求数>32时会返回错误
 *       4. 自适应路由模式下会发送两次WR
 */
scclResult_t scclNetIb::scclIbMultiSend(struct scclIbSendComm* comm, int slot) {
    struct scclIbRequest** reqs           = comm->fifoReqs[slot];
    volatile struct scclIbSendFifo* slots = comm->fifo[slot];
    int nreqs                             = slots[0].nreqs;
    if(nreqs > SCCL_NET_IB_MAX_RECVS)
        return scclInternalError;

    uint64_t wr_id = 0ULL;

    for(int r = 0; r < nreqs; r++) {
        struct ibv_send_wr* wr = comm->wrs + r;
        memset(wr, 0, sizeof(struct ibv_send_wr));

        struct ibv_sge* sge = comm->sges + r;
        sge->addr           = (uintptr_t)reqs[r]->send.data;
        sge->lkey           = reqs[r]->send.lkey;

        wr->opcode              = IBV_WR_RDMA_WRITE;
        wr->send_flags          = 0;
        wr->wr.rdma.remote_addr = slots[r].addr;
        wr->wr.rdma.rkey        = slots[r].rkey;
        wr->next                = wr + 1;
        wr_id += (reqs[r] - comm->verbs.reqs) << (r * 8);
    }

    // Write size as immediate data. In the case of multi-send, only write
    // 0 or 1 as size to indicate whether there was data sent or received.
    uint32_t immData = 0;
    if(nreqs == 1) {
        immData = reqs[0]->send.size;
    } else {
        if(nreqs > 32) {
            WARN("Cannot store sizes of %d requests in a 32-bits field", nreqs);
            return scclInternalError;
        }
        for(int r = 0; r < nreqs; r++) {
            immData |= (reqs[r]->send.size ? 1 : 0) << r;
        }
    }

    struct ibv_send_wr* lastWr = comm->wrs + nreqs - 1;
    if(nreqs > 1 || (comm->ar && reqs[0]->send.size > scclParamIbArThreshold())) {
        // When using ADAPTIVE_ROUTING, send the bulk of the data first as an
        // RDMA_WRITE, then a 0-byte RDMA_WRITE_WITH_IMM to trigger a remote
        // completion.
        lastWr++;
        memset(lastWr, 0, sizeof(struct ibv_send_wr));
    }
    lastWr->wr_id      = wr_id;
    lastWr->opcode     = IBV_WR_RDMA_WRITE_WITH_IMM;
    lastWr->imm_data   = immData;
    lastWr->next       = NULL;
    lastWr->send_flags = IBV_SEND_SIGNALED;

    // Multi-QP: make sure IB writes are multiples of 128B so that LL and LL128 protocols still work
    const int align = 128;
    const int nqps  = scclParamIbSplitDataOnQps() ? comm->nqps : 1;
    for(int q = 0; q < nqps; q++) {
        for(int r = 0; r < nreqs; r++) {
            int chunkSize = DIVUP(DIVUP(reqs[r]->send.size, nqps), align) * align;
            int length    = std::min(reqs[r]->send.size - reqs[r]->send.offset, chunkSize);
            if(length <= 0) {
                comm->wrs[r].sg_list = NULL;
                comm->wrs[r].num_sge = 0;
            } else {
                comm->sges[r].length = length;
                comm->wrs[r].sg_list = comm->sges + r;
                comm->wrs[r].num_sge = 1;
            }
        }
        struct ibv_send_wr* bad_wr;
        SCCLCHECK(wrap_ibv_post_send(comm->qps[comm->qpIndex], comm->wrs, &bad_wr));
        comm->qpIndex = (comm->qpIndex + 1) % comm->nqps;

        for(int r = 0; r < nreqs; r++) {
            int chunkSize = DIVUP(DIVUP(reqs[r]->send.size, nqps), align) * align;
            reqs[r]->send.offset += chunkSize;
            comm->sges[r].addr += chunkSize;
            comm->wrs[r].wr.rdma.remote_addr += chunkSize;
        }
    }

    return scclSuccess;
}

/**
 * @brief 通过IB Verbs RDMA写入操作向远程FIFO队列提交数据
 *
 * @param comm 指向接收通信上下文的指针
 * @param n 要发送的数据块数量
 * @param data 数据指针数组
 * @param sizes 数据大小数组
 * @param tags 数据标签数组
 * @param mhandles 内存句柄数组
 * @param req 请求结构体指针
 * @return scclResult_t 返回操作结果(scclSuccess表示成功)
 *
 * @note 该函数会将数据打包到本地FIFO元素中，并通过RDMA写入到远程FIFO队列。
 *       每MAX_REQUESTS次操作会触发一次带信号(SIGNALED)的发送，以避免发送队列堵塞。
 *       使用IBV_WR_RDMA_WRITE操作码进行数据传输。
 */
scclResult_t scclNetIb::scclIbPostFifo(struct scclIbRecvComm* comm, int n, void** data, int* sizes, int* tags, void** mhandles, struct scclIbRequest* req) {
    struct ibv_send_wr wr;
    memset(&wr, 0, sizeof(wr));

    int slot                         = comm->remFifo.fifoTail % MAX_REQUESTS;
    struct scclIbSendFifo* localElem = comm->remFifo.elems[slot];

    for(int i = 0; i < n; i++) {
        localElem[i].addr  = (uint64_t)data[i];
        struct ibv_mr* mr  = (struct ibv_mr*)mhandles[i];
        localElem[i].rkey  = mr->rkey;
        localElem[i].nreqs = n;
        localElem[i].size  = sizes[i]; // Sanity/Debugging
        localElem[i].tag   = tags[i];
        localElem[i].idx   = comm->remFifo.fifoTail + 1;
    }

    wr.wr.rdma.remote_addr   = comm->remFifo.addr + slot * SCCL_NET_IB_MAX_RECVS * sizeof(struct scclIbSendFifo);
    wr.wr.rdma.rkey          = comm->remFifo.rkey;
    comm->remFifo.sge.addr   = (uint64_t)localElem;
    comm->remFifo.sge.length = n * sizeof(struct scclIbSendFifo);
    wr.sg_list               = &comm->remFifo.sge;
    wr.num_sge               = 1;
    wr.opcode                = IBV_WR_RDMA_WRITE;
    wr.send_flags            = comm->remFifo.flags; // IBV_SEND_INLINE

    // We need to occasionally post a request with the IBV_SEND_SIGNALED flag, otherwise
    // the send queue will never empty.
    //
    // From https://www.rdmamojo.com/2014/06/30/working-unsignaled-completions/
    // "How to use Unsignaled Completion?" / "Gotchas and Pitfalls"
    // All posted Send Requested, Signaled and Unsignaled, are considered outstanding until
    // a Work Completion that they, or Send Requests that were posted after them, was polled
    // from the Completion Queue associated with the Send Queue. This means if one works with
    // a Queue Pair that was configured to work with Unsignaled Completions, he must make
    // sure that occasionally (before the Send Queue is full with outstanding Send Requests)
    // a Send Request that generate Work Completion will be posted.
    //
    // Not following this rule may lead to a case that the Send Queue is full with Send
    // Requests that won't generate Work Completion:
    //
    //  - The Send Queue is full, so no new Send Requests can be posted to it
    //  - The Send Queue can't be emptied, since no Work Completion can be generated anymore
    //    (the reason is that no Work Completion, that can generate Work Completion that
    //    polling it will empty the Send Queue, can be posted)
    //  - The status of all posted Send Request is considered unknown
    //
    if(slot == 0) {
        wr.send_flags |= IBV_SEND_SIGNALED;
        wr.wr_id = req - comm->verbs.reqs;
        req->events++;
    }

    struct ibv_send_wr* bad_wr;
    SCCLCHECK(wrap_ibv_post_send(comm->qps[0], &wr, &bad_wr));
    comm->remFifo.fifoTail++;

    return scclSuccess;
}

////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////// scclNetIb调用的函数 ////////////////////////////////////////
scclNetIb::scclNetIb() : scclNetBase("IB") {}

scclNetIb::~scclNetIb() {
    if(ibComm != nullptr) {
        free(ibComm);
    }
}

/**
 * @brief 初始化InfiniBand硬件设备
 *
 * 该函数负责检测和初始化可用的InfiniBand设备，包括：
 * - 加载IB Verbs符号
 * - 检测网络接口
 * - 查询设备属性
 * - 处理用户指定的HCA设备
 * - 创建异步线程处理IB事件
 *
 * @return scclResult_t 返回操作状态，scclSuccess表示成功，scclInternalError表示失败
 *
 * @note 函数内部会处理环境变量SCCL_IB_HCA来过滤特定设备
 * @note 使用互斥锁scclIbLock保证线程安全
 */
scclResult_t scclNetIb::init() {
    SCCLCHECK(scclCalloc(&ibComm, 1));

    // 如果IB被禁用，返回内部错误
    if(scclParamIbDisable())
        return scclInternalError;

    // 尝试初始化包装IB符号，如果失败返回内部错误
    if(wrap_ibv_symbols() != scclSuccess) {
        return scclInternalError;
    } else {
        INFO(SCCL_LOG_NET, "SCCL IB init done");
    }

    static int shownIbHcaEnv = 0;

    // 如果IB设备数量未初始化，开始初始化过程
    if(scclNIbDevs == -1) {
        pthread_mutex_lock(&scclIbLock);
        wrap_ibv_fork_init();

        if(scclNIbDevs == -1) {
            scclNIbDevs = 0;
            // 查找网络接口
            if(net_socket::scclFindSocketInterfaces(scclIbIfName, &scclIbIfAddr, MAX_IF_NAME_SIZE, 1) != 1) {
                WARN("NET/IB : No IP interface found.");
                return scclInternalError;
            }

            // 检测IB卡
            int nIbDevs;
            struct ibv_device** devices;
            struct netIf userIfs[MAX_IB_DEVS];
            bool searchNot, searchExact;

            // 获取用户指定的IB HCA（InfiniBand Host Channel Adapter）环境变量
            char* userIbEnv = scclIbGetIbHca(shownIbHcaEnv, &searchNot, &searchExact);
            // 解析用户指定的IB接口列表，将结果存储在userIfs数组中，最多解析MAX_IB_DEVS个接口
            int nUserIfs = parseStringList(userIbEnv, userIfs, MAX_IB_DEVS);

            // 获取设备列表
            if(scclSuccess != wrap_ibv_get_device_list(&devices, &nIbDevs))
                return scclInternalError;

            // 遍历所有设备
            for(int d = 0; d < nIbDevs && scclNIbDevs < MAX_IB_DEVS; d++) {
                struct ibv_context* context;
                // 尝试打开设备
                if(scclSuccess != wrap_ibv_open_device(&context, devices[d]) || context == NULL) {
                    WARN("NET/IB : Unable to open device %s", devices[d]->name);
                    continue;
                }
                int nPorts = 0;
                struct ibv_device_attr devAttr;
                memset(&devAttr, 0, sizeof(devAttr));
                // 查询设备属性
                if(scclSuccess != wrap_ibv_query_device(context, &devAttr)) {
                    WARN("NET/IB : Unable to query device %s", devices[d]->name);
                    if(scclSuccess != wrap_ibv_close_device(context)) {
                        return scclInternalError;
                    }
                    continue;
                }
                // 遍历设备的所有端口
                for(int port = 1; port <= devAttr.phys_port_cnt; port++) {
                    struct ibv_port_attr portAttr;
                    // 查询端口属性
                    if(scclSuccess != wrap_ibv_query_port(context, port, &portAttr)) {
                        WARN("NET/IB : Unable to query port %d", port);
                        continue;
                    }
                    // 检查端口状态和链接层
                    if(portAttr.state != IBV_PORT_ACTIVE)
                        continue;
                    if(portAttr.link_layer != IBV_LINK_LAYER_INFINIBAND && portAttr.link_layer != IBV_LINK_LAYER_ETHERNET)
                        continue;

                    // 检查用户指定的HCA/端口
                    if(!(matchIfList(devices[d]->name, port, userIfs, nUserIfs, searchExact) ^ searchNot)) {
                        continue;
                    }

                    INFO(SCCL_LOG_NET,
                         "NET/IB: [%d] %s: port=%d/IB=%s, speed:%d/%d",
                         d,
                         devices[d]->name,
                         port,
                         portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND ? "IB" : "RoCE",
                         scclIbSpeed(portAttr.active_speed),
                         scclIbSpeed(portAttr.active_speed) * scclIbWidth(portAttr.active_width));

                    pthread_mutex_init(&scclIbDevs[scclNIbDevs].lock, NULL);

                    INFO(SCCL_LOG_NET, "d=%d, node_guid=%llu, sys_image_guid=%llu\n", d, devAttr.node_guid, devAttr.sys_image_guid);
                    // 设置Infiniband设备的属性
                    {
                        scclIbDevs[scclNIbDevs].device  = d;                                                                       // 设备索引
                        scclIbDevs[scclNIbDevs].guid    = devAttr.sys_image_guid;                                                  // 系统图像GUID
                        scclIbDevs[scclNIbDevs].port    = port;                                                                    // 端口编号
                        scclIbDevs[scclNIbDevs].link    = portAttr.link_layer;                                                     // 链路层类型
                        scclIbDevs[scclNIbDevs].speed   = scclIbSpeed(portAttr.active_speed) * scclIbWidth(portAttr.active_width); // 计算设备速度
                        scclIbDevs[scclNIbDevs].context = context;                                                                 // 设备上下文
                        scclIbDevs[scclNIbDevs].pdRefs  = 0;                                                                       // 保护域引用计数
                        scclIbDevs[scclNIbDevs].pd      = NULL;                                                                    // 保护域指针
                        strncpy(scclIbDevs[scclNIbDevs].devName, devices[d]->name, MAXNAMESIZE);                                   // 复制设备名称
                        SCCLCHECK(scclIbGetPciPath(
                            scclIbDevs[scclNIbDevs].devName, &scclIbDevs[scclNIbDevs].pciPath, &scclIbDevs[scclNIbDevs].realPort)); // 获取PCI路径和实际端口
                        scclIbDevs[scclNIbDevs].maxQp              = devAttr.max_qp;                                                // 最大队列对数量
                        scclIbDevs[scclNIbDevs].mrCache.capacity   = 0;                                                             // MR缓存容量
                        scclIbDevs[scclNIbDevs].mrCache.population = 0;                                                             // MR缓存人口
                        scclIbDevs[scclNIbDevs].mrCache.slots      = NULL;                                                          // MR缓存槽

                        // 默认在IB网络上启用ADAPTIVE_ROUTING，但允许通过环境参数覆盖
                        scclIbDevs[scclNIbDevs].ar = (portAttr.link_layer == IBV_LINK_LAYER_INFINIBAND) ? 1 : 0; // 根据链路层类型设置自适应路由
                        if(scclParamIbAdaptiveRouting() != -2)
                            scclIbDevs[scclNIbDevs].ar = scclParamIbAdaptiveRouting(); // 如果环境参数设置，则覆盖默认值
                    }

                    // 创建一个新的线程，用于处理SCCL Infiniband的异步操作
                    pthread_create(&scclIbAsyncThread, NULL, scclIbAsyncThreadMain, context);

                    // 设置新创建线程的名称，以便于调试和识别
                    scclSetThreadName(scclIbAsyncThread, "SCCL IbAsync %2d", scclNIbDevs);
                    // 分离线程，使其在完成后自动回收资源，不需要调用pthread_join()
                    pthread_detach(scclIbAsyncThread);

                    scclNIbDevs++; // 增加Infiniband设备的计数
                    nPorts++;      // 增加端口计数

                    // 再次调用pthread_detach，这行代码可能是多余的，需检查是否为误写
                    pthread_detach(scclIbAsyncThread);
                }
                // 如果没有活动端口，关闭设备
                if(nPorts == 0 && scclSuccess != wrap_ibv_close_device(context)) {
                    return scclInternalError;
                }
            }

            // 释放设备列表
            if(nIbDevs && (scclSuccess != wrap_ibv_free_device_list(devices))) {
                return scclInternalError;
            };
        }

        // 如果没有找到设备，打印信息
        if(scclNIbDevs == 0) {
            WARN("NET/IB : No device found.");
        } else {
            char line[1024];
            line[0] = '\0';
            // 确定是否启用了RELAXED_ORDERING
            scclIbRelaxedOrderingEnabled = scclIbRelaxedOrderingCapable();
            for(int d = 0; d < scclNIbDevs; d++) {
                snprintf(line + strlen(line),
                         1023 - strlen(line),
                         " -- [%d]%s:%d/%s; ",
                         d,
                         scclIbDevs[d].devName,
                         scclIbDevs[d].port,
                         scclIbDevs[d].link == IBV_LINK_LAYER_INFINIBAND ? "IB" : "RoCE");
            }

            // 确保line字符串以null字符结尾，防止字符串操作时出现未定义行为
            line[1023] = '\0';

            // 定义一个字符数组addrline，用于存储转换后的地址字符串
            char addrline[SOCKET_NAME_MAXLEN + 1];

            // 记录日志信息，描述当前网络/IB设备的配置和状态
            // line 是设备的相关信息字符串
            // scclIbRelaxedOrderingEnabled 是一个布尔值，指示是否启用了Relaxed Ordering
            // scclIbIfName 是IB接口的名称
            // net_socket::scclSocketToString 是一个函数，用于将socket地址转换为字符串
            // addrline 是存储转换后地址字符串的数组
            INFO(SCCL_LOG_NET,
                 "NET/IB : Using%s %s; OOB %s:%s",
                 line,
                 scclIbRelaxedOrderingEnabled ? "[RO]" : "",
                 scclIbIfName,
                 net_socket::scclSocketToString(&scclIbIfAddr, addrline));
        }
        pthread_mutex_unlock(&scclIbLock);
    }
    return scclSuccess;
}

/**
 * 获取可用的InfiniBand设备数量
 *
 * @param ndev [out] 用于存储设备数量的指针
 * @return scclResult_t 返回操作结果，scclSuccess表示成功
 */
scclResult_t scclNetIb::devices(int* ndev) {
    *ndev = scclNIbDevs;
    return scclSuccess;
}

/**
 * @brief 获取指定IB设备的网络属性
 *
 * 该函数用于查询指定InfiniBand设备的各项属性，包括设备名称、PCI路径、GUID、
 * 指针支持类型、速度、延迟、端口号、最大通信数和最大接收数等。
 *
 * @param dev 设备索引
 * @param props 用于存储设备属性的结构体指针
 * @return scclResult_t 返回操作结果，成功返回scclSuccess
 */
scclResult_t scclNetIb::getProperties(int dev, scclNetProperties_t* props) {
    props->name    = scclIbDevs[dev].devName;
    props->pciPath = scclIbDevs[dev].pciPath;
    props->guid    = scclIbDevs[dev].guid;

    props->ptrSupport = SCCL_PTR_HOST;
    if(scclIbGdrSupport(dev) == scclSuccess) {
        props->ptrSupport |= SCCL_PTR_CUDA; // GDR support via nv_peermem
    }
    if(scclIbDmaBufSupport(dev) == scclSuccess) {
        props->ptrSupport |= SCCL_PTR_DMABUF; // GDR support via DMA-BUF
    }
    props->speed    = scclIbDevs[dev].speed;
    props->latency  = 0; // Not set
    props->port     = scclIbDevs[dev].port + scclIbDevs[dev].realPort;
    props->maxComms = scclIbDevs[dev].maxQp;
    props->maxRecvs = SCCL_NET_IB_MAX_RECVS;

    return scclSuccess;
}

/**
 * @brief 在指定设备上创建并初始化IB监听通信
 *
 * @param dev 设备号
 * @param opaqueHandle 不透明的句柄指针，用于存储连接信息
 * @param listenComm 返回的监听通信结构体指针
 * @return scclResult_t 返回操作结果状态码
 *
 * 该函数会：
 * 1. 分配并初始化监听通信结构体
 * 2. 设置设备号和魔法数
 * 3. 根据配置决定是否复用套接字
 * 4. 启动套接字监听并获取连接地址
 */
scclResult_t scclNetIb::listen(int dev, void* opaqueHandle, void** listenComm) {
    memset(ibComm, 0, sizeof(struct scclIbListenComm));

    struct scclIbHandle* handle = (struct scclIbHandle*)opaqueHandle;
    // 静态断言，确保 scclIbHandle 结构体的大小不超过 SCCL_NET_HANDLE_MAXSIZE
    static_assert(sizeof(struct scclIbHandle) < SCCL_NET_HANDLE_MAXSIZE, "scclIbHandle size too large");
    // 将 handle 指向的内存区域清零，大小为 scclIbHandle 结构体的大小
    memset(handle, 0, sizeof(struct scclIbHandle));

    // 设置设备和处理句柄
    ibComm->dev   = dev;
    handle->magic = SCCL_SOCKET_MAGIC;
    SCCLCHECK(net_socket::scclSocketInit(&ibComm->sock, &scclIbIfAddr, handle->magic, net_socket::scclSocketTypeNetIb, NULL, 1));

    // 如果启用了端口复用，则复用套接字地址和文件描述符
    if(scclParamIbSockServerPortReuse()) {
        if(reusedSockfd == -1) {
            SCCLCHECK(scclSocketListen(&ibComm->sock));
            memcpy(&reusedAddr, &ibComm->sock.addr, sizeof(union net_socket::scclSocketAddress));
            reusedSockfd = ibComm->sock.fd;
        } else {
            memcpy(&ibComm->sock.addr, &reusedAddr, sizeof(union net_socket::scclSocketAddress));
            ibComm->sock.fd = reusedSockfd;
        }
    } else {
        SCCLCHECK(net_socket::scclSocketListen(&ibComm->sock));
    }

    // 获取套接字地址并设置监听通信
    SCCLCHECK(net_socket::scclSocketGetAddr(&ibComm->sock, &handle->connectAddr));
    *listenComm = ibComm;

    return scclSuccess;
}

/**
 * @brief 建立IB网络连接并初始化通信资源
 *
 * 该函数负责完成以下操作：
 * 1. 初始化socket连接
 * 2. 创建IB QP队列对
 * 3. 交换QP信息
 * 4. 完成QP状态转换(RTR/RTS)
 * 5. 注册内存区域
 *
 * @param dev 设备索引
 * @param opaqueHandle 包含连接信息的句柄
 * @param sendComm 输出参数，返回建立的发送通信上下文
 * @return scclResult_t 返回操作结果状态码
 *
 * @note 该函数使用状态机模式处理异步连接过程
 * @warning 不能重复连接已建立的sendComm
 */
scclResult_t scclNetIb::connect(int dev, void* opaqueHandle, void** sendComm) {
    struct scclIbHandle* handle   = (struct scclIbHandle*)opaqueHandle;
    struct scclIbCommStage* stage = &handle->stage;
    struct scclIbSendComm* comm   = (struct scclIbSendComm*)stage->comm;
    int ready;
    *sendComm = NULL;

    if(stage->state == scclIbCommStateConnect)
        goto ib_connect_check;
    if(stage->state == scclIbCommStateSend)
        goto ib_send;
    if(stage->state == scclIbCommStateConnecting)
        goto ib_connect;
    if(stage->state == scclIbCommStateConnected)
        goto ib_send_ready;
    if(stage->state != scclIbCommStateStart) {
        WARN("Error: trying to connect already connected sendComm");
        return scclInternalError;
    }

    SCCLCHECK(scclIbMalloc((void**)&comm, sizeof(struct scclIbSendComm)));
    SCCLCHECK(net_socket::scclSocketInit(&comm->sock, &handle->connectAddr, handle->magic, net_socket::scclSocketTypeNetIb, NULL, 1));
    stage->comm  = comm;
    stage->state = scclIbCommStateConnect;
    SCCLCHECK(net_socket::scclSocketConnect(&comm->sock, scclParamIbSockClientPortReuse()));

ib_connect_check:
    /* since scclSocketConnect is async, we must check if connection is complete */
    SCCLCHECK(net_socket::scclSocketReady(&comm->sock, &ready));
    if(!ready)
        return scclSuccess;

    // IB Setup
    struct ibv_context* ctx;
    ctx = scclIbDevs[dev].context;
    SCCLCHECK(scclIbInitVerbs(dev, ctx, &comm->verbs));
    uint8_t ib_port;
    ib_port    = scclIbDevs[dev].port;
    comm->nqps = scclParamIbQpsPerConn();
    for(int q = 0; q < comm->nqps; q++) {
        SCCLCHECK(scclIbCreateQp(ib_port, &comm->verbs, IBV_ACCESS_REMOTE_WRITE, comm->qps + q));
    }
    comm->ar = scclIbDevs[dev].ar; // ADAPTIVE_ROUTING

    // Send my QP Info to receiver through the socket. Hope this won't block.
    struct ibv_port_attr portAttr;
    SCCLCHECK(wrap_ibv_query_port(ctx, ib_port, &portAttr));
    struct scclIbQpInfo qpInfo;
    qpInfo.ib_port = ib_port;
    for(int q = 0; q < comm->nqps; q++)
        qpInfo.qpn[q] = comm->qps[q]->qp_num;
    qpInfo.mtu = portAttr.active_mtu;

    // Prepare my fifo
    SCCLCHECK(wrap_ibv_reg_mr(&comm->fifoMr,
                              comm->verbs.pd,
                              comm->fifo,
                              sizeof(struct scclIbSendFifo) * MAX_REQUESTS * SCCL_NET_IB_MAX_RECVS,
                              IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ));
    qpInfo.fifoRkey = comm->fifoMr->rkey;
    qpInfo.fifoAddr = (uint64_t)comm->fifo;

    // RoCE support
    qpInfo.lid        = portAttr.lid;
    qpInfo.link_layer = comm->gidInfo.link_layer = portAttr.link_layer;
    if(qpInfo.link_layer == IBV_LINK_LAYER_INFINIBAND) { // IB
        for(int q = 0; q < comm->nqps; q++)
            INFO(SCCL_LOG_NET, "NET/IB: Dev %d Port %d qpn %d mtu %d LID %d", dev, ib_port, qpInfo.qpn[q], qpInfo.mtu, qpInfo.lid);
    } else { // RoCE
        SCCLCHECK(wrap_ibv_query_gid(ctx, ib_port, scclParamIbGidIndex(), &comm->gidInfo.localGid));
        qpInfo.spn = comm->gidInfo.localGid.global.subnet_prefix;
        qpInfo.iid = comm->gidInfo.localGid.global.interface_id;
        for(int q = 0; q < comm->nqps; q++)
            INFO(SCCL_LOG_NET,
                 "NET/IB: Dev %d Port %d qpn %d mtu %d GID %ld (%lX/%lX)",
                 dev,
                 ib_port,
                 qpInfo.qpn[q],
                 qpInfo.mtu,
                 scclParamIbGidIndex(),
                 qpInfo.spn,
                 qpInfo.iid);
    }

    stage->state  = scclIbCommStateSend;
    stage->offset = 0;
    SCCLCHECK(scclIbMalloc((void**)&stage->buffer, sizeof(qpInfo)));
    memcpy(stage->buffer, &qpInfo, sizeof(qpInfo));

ib_send:
    SCCLCHECK(scclSocketProgress(SCCL_SOCKET_SEND, &comm->sock, stage->buffer, sizeof(qpInfo), &stage->offset));
    if(stage->offset != sizeof(qpInfo))
        return scclSuccess;

    stage->state  = scclIbCommStateConnecting;
    stage->offset = 0;
    // Clear the staging buffer for re-use
    memset(stage->buffer, 0, sizeof(qpInfo));

ib_connect:
    struct scclIbQpInfo remQpInfo;
    SCCLCHECK(scclSocketProgress(SCCL_SOCKET_RECV, &comm->sock, stage->buffer, sizeof(scclIbQpInfo), &stage->offset));
    if(stage->offset != sizeof(remQpInfo))
        return scclSuccess;

    memcpy(&remQpInfo, stage->buffer, sizeof(scclIbQpInfo));

    comm->gidInfo.remoteGid.global.subnet_prefix = remQpInfo.spn;
    comm->gidInfo.remoteGid.global.interface_id  = remQpInfo.iid;
    for(int q = 0; q < comm->nqps; q++) {
        struct ibv_qp* qp = comm->qps[q];
        SCCLCHECK(scclIbRtrQp(qp, remQpInfo.qpn[q], &remQpInfo));
        SCCLCHECK(scclIbRtsQp(qp));
    }

    comm->ready   = 1;
    stage->state  = scclIbCommStateConnected;
    stage->offset = 0;

ib_send_ready:
    SCCLCHECK(scclSocketProgress(SCCL_SOCKET_SEND, &comm->sock, &comm->ready, sizeof(int), &stage->offset));
    if(stage->offset != sizeof(int))
        return scclSuccess;

    free(stage->buffer);
    stage->state = scclIbCommStateStart;

    *sendComm = comm;
    return scclSuccess;
}

/**
 * @brief 接受IB连接请求并建立通信通道
 *
 * 该函数处理IB连接的接受过程，包括以下步骤：
 * 1. 初始化接收通信结构体
 * 2. 接受socket连接
 * 3. 交换QP信息
 * 4. 创建并配置QP队列
 * 5. 设置远程FIFO信息
 * 6. 处理GPU直接RDMA刷新缓冲区
 * 7. 完成握手过程
 *
 * @param listenComm 监听通信句柄
 * @param recvComm 输出参数，接收通信句柄
 * @return scclResult_t 返回操作结果，成功返回scclSuccess
 */
scclResult_t scclNetIb::accept(void* listenComm, void** recvComm) {
    struct scclIbListenComm* lComm = (struct scclIbListenComm*)listenComm;
    struct scclIbCommStage* stage  = &lComm->stage;
    struct scclIbRecvComm* rComm   = (struct scclIbRecvComm*)stage->comm;
    int ready;
    *recvComm = NULL;

    if(stage->state == scclIbCommStateAccept)
        goto ib_accept_check;
    if(stage->state == scclIbCommStateRecv)
        goto ib_recv;
    if(stage->state == scclIbCommStateSend)
        goto ib_send;
    if(stage->state == scclIbCommStatePendingReady)
        goto ib_recv_ready;
    if(stage->state != scclIbCommStateStart) {
        WARN("Listencomm in unknown state %d", stage->state);
        return scclInternalError;
    }

    SCCLCHECK(scclIbMalloc((void**)&rComm, sizeof(struct scclIbRecvComm)));
    stage->comm  = rComm;
    stage->state = scclIbCommStateAccept;
    SCCLCHECK(net_socket::scclSocketInit(&rComm->sock));
    SCCLCHECK(net_socket::scclSocketAccept(&rComm->sock, &lComm->sock));

ib_accept_check:
    SCCLCHECK(net_socket::scclSocketReady(&rComm->sock, &ready));
    if(!ready)
        return scclSuccess;

    struct scclIbQpInfo remQpInfo;
    stage->state  = scclIbCommStateRecv;
    stage->offset = 0;
    SCCLCHECK(scclIbMalloc((void**)&stage->buffer, sizeof(remQpInfo)));

ib_recv:
    SCCLCHECK(net_socket::scclSocketProgress(SCCL_SOCKET_RECV, &rComm->sock, stage->buffer, sizeof(remQpInfo), &stage->offset));
    if(stage->offset != sizeof(remQpInfo))
        return scclSuccess;

    /* copy back the received info */
    memcpy(&remQpInfo, stage->buffer, sizeof(struct scclIbQpInfo));

    rComm->gidInfo.remoteGid.global.subnet_prefix = remQpInfo.spn;
    rComm->gidInfo.remoteGid.global.interface_id  = remQpInfo.iid;

    // IB setup
    struct ibv_context* ctx;
    uint8_t ib_port;
    ctx     = scclIbDevs[lComm->dev].context;
    ib_port = scclIbDevs[lComm->dev].port;
    struct ibv_port_attr portAttr;
    SCCLCHECK(wrap_ibv_query_port(ctx, ib_port, &portAttr));
    SCCLCHECK(wrap_ibv_query_gid(ctx, ib_port, scclParamIbGidIndex(), &rComm->gidInfo.localGid));

    // QP Creation
    SCCLCHECK(scclIbInitVerbs(lComm->dev, ctx, &rComm->verbs));
    rComm->nqps = scclParamIbQpsPerConn();
    for(int q = 0; q < rComm->nqps; q++) {
        SCCLCHECK(scclIbCreateQp(ib_port, &rComm->verbs, IBV_ACCESS_REMOTE_WRITE, rComm->qps + q));
    }

    // Adjust the MTU
    remQpInfo.mtu = (enum ibv_mtu)std::min(remQpInfo.mtu, portAttr.active_mtu);

    // Setup QP
    for(int q = 0; q < rComm->nqps; q++) {
        struct ibv_qp* qp = rComm->qps[q];
        SCCLCHECK(scclIbRtrQp(qp, remQpInfo.qpn[q], &remQpInfo));
        SCCLCHECK(scclIbRtsQp(qp));
    }

    // Retain remote fifo info and prepare my RDMA ops
    rComm->remFifo.rkey = remQpInfo.fifoRkey;
    rComm->remFifo.addr = remQpInfo.fifoAddr;
    SCCLCHECK(wrap_ibv_reg_mr(&rComm->remFifo.mr,
                              rComm->verbs.pd,
                              &rComm->remFifo.elems,
                              sizeof(struct scclIbSendFifo) * MAX_REQUESTS * SCCL_NET_IB_MAX_RECVS,
                              IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ));
    rComm->remFifo.sge.lkey = rComm->remFifo.mr->lkey;
    if(scclParamIbUseInline())
        rComm->remFifo.flags = IBV_SEND_INLINE;

    // Allocate Flush dummy buffer for GPU Direct RDMA
    rComm->gpuFlush.enabled =
        ((scclIbGdrSupport(lComm->dev) == scclSuccess || scclIbDmaBufSupport(lComm->dev) == scclSuccess) && (scclParamIbGdrFlushDisable() == 0)) ? 1 : 0;
    if(rComm->gpuFlush.enabled) {
        SCCLCHECK(wrap_ibv_reg_mr(&rComm->gpuFlush.hostMr, rComm->verbs.pd, &rComm->gpuFlush.hostMem, sizeof(int), IBV_ACCESS_LOCAL_WRITE));
        rComm->gpuFlush.sge.addr   = (uint64_t)&rComm->gpuFlush.hostMem;
        rComm->gpuFlush.sge.length = 1;
        rComm->gpuFlush.sge.lkey   = rComm->gpuFlush.hostMr->lkey;
        SCCLCHECK(scclIbCreateQp(ib_port, &rComm->verbs, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ, &rComm->gpuFlush.qp));
        struct scclIbQpInfo localQpInfo;
        localQpInfo.lid        = portAttr.lid;
        localQpInfo.link_layer = portAttr.link_layer;
        localQpInfo.ib_port    = ib_port;
        localQpInfo.spn        = rComm->gidInfo.localGid.global.subnet_prefix;
        localQpInfo.iid        = rComm->gidInfo.localGid.global.interface_id;
        localQpInfo.mtu        = portAttr.active_mtu;
        SCCLCHECK(scclIbRtrQp(rComm->gpuFlush.qp, rComm->gpuFlush.qp->qp_num, &localQpInfo));
        SCCLCHECK(scclIbRtsQp(rComm->gpuFlush.qp));
    }

    // Fill Handle
    struct scclIbQpInfo qpInfo;
    qpInfo.lid        = portAttr.lid;
    qpInfo.link_layer = rComm->gidInfo.link_layer = portAttr.link_layer;
    qpInfo.ib_port                                = ib_port;
    for(int q = 0; q < rComm->nqps; q++)
        qpInfo.qpn[q] = rComm->qps[q]->qp_num;
    qpInfo.spn = rComm->gidInfo.localGid.global.subnet_prefix;
    qpInfo.iid = rComm->gidInfo.localGid.global.interface_id;
    qpInfo.mtu = remQpInfo.mtu;

    stage->state  = scclIbCommStateSend;
    stage->offset = 0;
    if(stage->buffer)
        free(stage->buffer);
    SCCLCHECK(scclIbMalloc((void**)&stage->buffer, sizeof(struct scclIbQpInfo)));
    memcpy(stage->buffer, &qpInfo, sizeof(struct scclIbQpInfo));

ib_send:
    SCCLCHECK(net_socket::scclSocketProgress(SCCL_SOCKET_SEND, &rComm->sock, stage->buffer, sizeof(struct scclIbQpInfo), &stage->offset));
    if(stage->offset < sizeof(struct scclIbQpInfo))
        return scclSuccess;

    stage->offset = 0;
    stage->state  = scclIbCommStatePendingReady;

ib_recv_ready:
    SCCLCHECK(net_socket::scclSocketProgress(SCCL_SOCKET_RECV, &rComm->sock, &rComm->ready, sizeof(int), &stage->offset));
    if(stage->offset != sizeof(int))
        return scclSuccess;

    free(stage->buffer);
    *recvComm = rComm;

    /* reset lComm stage */
    stage->state  = scclIbCommStateStart;
    stage->offset = 0;
    stage->comm   = NULL;
    stage->buffer = NULL;
    return scclSuccess;
}

/* DMA-BUF support */
scclResult_t scclNetIb::regMrDmaBuf(void* comm, void* data, size_t size, int type, uint64_t offset, int fd, void** mhandle) {
    static_assert(offsetof(struct scclIbSendComm, verbs) == offsetof(struct scclIbRecvComm, verbs), "Send and recv comms must have verbs at the same offset");
    assert(size > 0);

    static __thread uintptr_t pageSize = 0;
    if(pageSize == 0)
        pageSize = sysconf(_SC_PAGESIZE);

    struct scclIbVerbs* verbs   = (struct scclIbVerbs*)comm;
    struct scclIbMrCache* cache = &scclIbDevs[verbs->dev].mrCache;
    uintptr_t addr              = (uintptr_t)data & -pageSize;
    size_t pages                = ((uintptr_t)data + size - addr + pageSize - 1) / pageSize;
    scclResult_t res;
    pthread_mutex_lock(&scclIbDevs[verbs->dev].lock);
    for(int slot = 0; /*true*/; slot++) {
        if(slot == cache->population) {                // didn't find in cache
            if(cache->population == cache->capacity) { // must grow cache
                cache->capacity = cache->capacity < 32 ? 32 : 2 * cache->capacity;
                SCCLCHECKGOTO(scclRealloc(&cache->slots, cache->population, cache->capacity), res, returning);
            }
            // Deregister / register
            struct ibv_mr* mr;
            unsigned int flags = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ;
            if(scclIbRelaxedOrderingEnabled)
                flags |= IBV_ACCESS_RELAXED_ORDERING;
            if(fd != -1) {
                /* DMA-BUF support */
                SCCLCHECKGOTO(wrap_ibv_reg_dmabuf_mr(&mr, verbs->pd, offset, pages * pageSize, addr, fd, flags), res, returning);
            } else {
                if(scclIbRelaxedOrderingEnabled) {
                    // Use IBVERBS_1.8 API - needed for IBV_ACCESS_RELAXED_ORDERING support
                    SCCLCHECKGOTO(wrap_ibv_reg_mr_iova2(&mr, verbs->pd, (void*)addr, pages * pageSize, addr, flags), res, returning);
                } else {
                    SCCLCHECKGOTO(wrap_ibv_reg_mr(&mr, verbs->pd, (void*)addr, pages * pageSize, flags), res, returning);
                }
            }

            INFO(SCCL_LOG_NET, "regAddr %llx size %lld rkey %x fd %d", (unsigned long long)addr, (long long)pages * pageSize, mr->rkey, fd);
            cache->population += 1;
            cache->slots[slot].addr  = addr;
            cache->slots[slot].pages = pages;
            cache->slots[slot].refs  = 1;
            cache->slots[slot].mr    = mr;
            *mhandle                 = (void*)mr;
            res                      = scclSuccess;
            goto returning;
        } else if(cache->slots[slot].addr == addr && cache->slots[slot].pages == pages) {
            cache->slots[slot].refs += 1;
            *mhandle = (void*)cache->slots[slot].mr;
            res      = scclSuccess;
            goto returning;
        }
    }
returning:
    pthread_mutex_unlock(&scclIbDevs[verbs->dev].lock);
    return res;
}

scclResult_t scclNetIb::regMr(void* comm, void* data, int size, int type, void** mhandle) {
    return regMrDmaBuf(comm, data, (size_t)size, type, 0ULL, -1, mhandle);
}

/**
 * @brief 注销IB内存区域(MR)
 *
 * 该函数用于注销指定的IB内存区域(MR)，并更新MR缓存。如果MR的引用计数减至0，
 * 则从缓存中移除并调用ibv_dereg_mr释放资源。
 *
 * @param comm 通信上下文指针
 * @param mhandle 要注销的内存区域句柄
 * @return scclResult_t 返回操作结果(scclSuccess表示成功)
 */
scclResult_t scclNetIb::deregMr(void* comm, void* mhandle) {
    struct scclIbVerbs* verbs   = (struct scclIbVerbs*)comm;
    struct scclIbMrCache* cache = &scclIbDevs[verbs->dev].mrCache;
    scclResult_t res;
    pthread_mutex_lock(&scclIbDevs[verbs->dev].lock);
    for(int i = 0; i < cache->population; i++) {
        if(mhandle == cache->slots[i].mr) {
            if(0 == --cache->slots[i].refs) {
                memmove(&cache->slots[i], &cache->slots[--cache->population], sizeof(struct scclIbMr));
                if(cache->population == 0) {
                    free(cache->slots);
                    cache->slots    = NULL;
                    cache->capacity = 0;
                }
                SCCLCHECKGOTO(wrap_ibv_dereg_mr((struct ibv_mr*)mhandle), res, returning);
            }
            res = scclSuccess;
            goto returning;
        }
    }
    WARN("NET/IB: could not find mr %p inside cache of %d entries", mhandle, cache->population);
    res = scclInternalError;
returning:
    pthread_mutex_unlock(&scclIbDevs[verbs->dev].lock);
    return res;
}

scclResult_t scclNetIb::isend(void* sendComm, void* data, int size, int tag, void* mhandle, void** request) {
    struct scclIbSendComm* comm = (struct scclIbSendComm*)sendComm;
    if(comm->ready == 0) {
        WARN("NET/IB: isend() called when comm->ready == 0");
        return scclInternalError;
    }
    if(comm->ready == 0) {
        *request = NULL;
        return scclSuccess;
    }

    struct ibv_mr* mr = (struct ibv_mr*)mhandle;

    // Wait for the receiver to have posted the corresponding receive
    int nreqs = 0;
    volatile struct scclIbSendFifo* slots;

    int slot                    = (comm->fifoHead) % MAX_REQUESTS;
    struct scclIbRequest** reqs = comm->fifoReqs[slot];
    slots                       = comm->fifo[slot];
    uint64_t idx                = comm->fifoHead + 1;
    if(slots[0].idx != idx) {
        *request = NULL;
        return scclSuccess;
    }
    nreqs = slots[0].nreqs;
    // Wait until all data has arrived
    for(int r = 1; r < nreqs; r++)
        while(slots[r].idx != idx)
            ;
    __sync_synchronize(); // order the nreqsPtr load against tag/rkey/addr loads below
    for(int r = 0; r < nreqs; r++) {
        if(reqs[r] != NULL || slots[r].tag != tag)
            continue;

        // Sanity checks to catch user collective call count/size mismatches
        if(size > slots[r].size) {
            char line[SOCKET_NAME_MAXLEN + 1];
            union net_socket::scclSocketAddress addr;
            net_socket::scclSocketGetAddr(&comm->sock, &addr);
            WARN("NET/IB : req %d/%d tag %x peer %s collective mismatch error, local size %d remote size %d",
                 r,
                 nreqs,
                 tag,
                 net_socket::scclSocketToString(&addr, line),
                 size,
                 slots[r].size);
            return scclInvalidUsage;
        } // plus any potential programming errors
        else if(slots[r].size < 0 || slots[r].addr == 0 || slots[r].rkey == 0) {
            char line[SOCKET_NAME_MAXLEN + 1];
            union net_socket::scclSocketAddress addr;
            net_socket::scclSocketGetAddr(&comm->sock, &addr);
            WARN("NET/IB : req %d/%d tag %x peer %s posted incorrect receive info: size %d addr %lx rkey %x",
                 r,
                 nreqs,
                 tag,
                 net_socket::scclSocketToString(&addr, line),
                 slots[r].size,
                 slots[r].addr,
                 slots[r].rkey);
            return scclInternalError;
        }
        struct scclIbRequest* req;
        SCCLCHECK(scclIbGetRequest(&comm->verbs, &req));
        req->type        = SCCL_NET_IB_REQ_SEND;
        req->sock        = &comm->sock;
        req->verbs       = &comm->verbs;
        req->nreqs       = nreqs;
        req->send.size   = size;
        req->send.data   = data;
        req->send.lkey   = mr->lkey;
        req->send.offset = 0;
        req->events      = scclParamIbSplitDataOnQps() ? comm->nqps : 1;
        if(comm->gidInfo.link_layer == IBV_LINK_LAYER_ETHERNET)
            req->gidInfo = &comm->gidInfo;
        *request = reqs[r] = req;

        // If this is a multi-recv, send only when all requests have matched.
        for(int r = 0; r < nreqs; r++) {
            if(reqs[r] == NULL)
                return scclSuccess;
        }

        SCCLCHECK(scclIbMultiSend(comm, slot));

        // Clear slots[0]->nreqs, as well as other fields to help debugging and sanity checks
        memset((void*)slots, 0, sizeof(struct scclIbSendFifo));
        memset(reqs, 0, SCCL_NET_IB_MAX_RECVS * sizeof(struct scclIbRequest*));
        comm->fifoHead++;
        return scclSuccess;
    }

    *request = NULL;
    return scclSuccess;
}

scclResult_t scclNetIb::irecv(void* recvComm, int n, void** data, int* sizes, int* tags, void** mhandles, void** request) {
    struct scclIbRecvComm* comm = (struct scclIbRecvComm*)recvComm;
    if(comm->ready == 0) {
        WARN("NET/IB: irecv() called when comm->ready == 0");
        return scclInternalError;
    }
    if(comm->ready == 0) {
        *request = NULL;
        return scclSuccess;
    }
    if(n > SCCL_NET_IB_MAX_RECVS)
        return scclInternalError;

    struct scclIbRequest* req;
    SCCLCHECK(scclIbGetRequest(&comm->verbs, &req));
    req->type  = SCCL_NET_IB_REQ_RECV;
    req->sock  = &comm->sock;
    req->nreqs = n;
    if(comm->gidInfo.link_layer == IBV_LINK_LAYER_ETHERNET)
        req->gidInfo = &comm->gidInfo;
    for(int i = 0; i < n; i++)
        req->recv.sizes[i] = 0;

    struct ibv_recv_wr wr;
    memset(&wr, 0, sizeof(wr));
    wr.wr_id = req - comm->verbs.reqs;

    wr.sg_list = NULL;
    wr.num_sge = 0;

    const int nqps = scclParamIbSplitDataOnQps() ? comm->nqps : 1;
    for(int q = 0; q < nqps; q++) {
        struct ibv_qp* qp = comm->qps[comm->qpIndex];
        struct ibv_recv_wr* bad_wr;
        SCCLCHECK(wrap_ibv_post_recv(qp, &wr, &bad_wr));
        comm->qpIndex = (comm->qpIndex + 1) % comm->nqps;
    }
    req->events = nqps;

    *request = req;

    // Post to FIFO to notify sender
    SCCLCHECK(scclIbPostFifo(comm, n, data, sizes, tags, mhandles, req));
    return scclSuccess;
}

scclResult_t scclNetIb::iflush(void* recvComm, int n, void** data, int* sizes, void** mhandles, void** request) {
    struct scclIbRecvComm* comm = (struct scclIbRecvComm*)recvComm;
    int last                    = -1;
    for(int i = 0; i < n; i++)
        if(sizes[i])
            last = i;
    if(comm->gpuFlush.enabled == 0 || last == -1)
        return scclSuccess;

    // Only flush once using the last non-zero receive
    struct scclIbRequest* req;
    SCCLCHECK(scclIbGetRequest(&comm->verbs, &req));
    req->type         = SCCL_NET_IB_REQ_FLUSH;
    req->sock         = &comm->sock;
    struct ibv_mr* mr = (struct ibv_mr*)mhandles[last];

    struct ibv_send_wr wr;
    memset(&wr, 0, sizeof(wr));
    wr.wr_id = req - comm->verbs.reqs;

    wr.wr.rdma.remote_addr = (uint64_t)data[last];
    wr.wr.rdma.rkey        = mr->rkey;
    wr.sg_list             = &comm->gpuFlush.sge;
    wr.num_sge             = 1;
    wr.opcode              = IBV_WR_RDMA_READ;
    wr.send_flags          = IBV_SEND_SIGNALED;

    struct ibv_send_wr* bad_wr;
    SCCLCHECK(wrap_ibv_post_send(comm->gpuFlush.qp, &wr, &bad_wr));

    *request = req;
    return scclSuccess;
}

scclResult_t scclNetIb::test(void* request, int* done, int* sizes) {
    struct scclIbRequest* r = (struct scclIbRequest*)request;
    *done                   = 0;

    while(1) {
        if(r->events == 0) {
            *done = 1;
            if(sizes && r->type == SCCL_NET_IB_REQ_RECV) {
                for(int i = 0; i < r->nreqs; i++)
                    sizes[i] = r->recv.sizes[i];
            }
            SCCLCHECK(scclIbFreeRequest(r));
            return scclSuccess;
        }

        int wrDone = 0;
        struct ibv_wc wcs[4];
        SCCLCHECK(wrap_ibv_poll_cq(r->verbs->cq, 4, wcs, &wrDone));

        if(wrDone == 0)
            return scclSuccess;

        for(int w = 0; w < wrDone; w++) {
            struct ibv_wc* wc = wcs + w;
            if(wc->status != IBV_WC_SUCCESS) {
                char line[SOCKET_NAME_MAXLEN + 1];
                union net_socket::scclSocketAddress addr;
                net_socket::scclSocketGetAddr(r->sock, &addr);
                char localGidString[INET6_ADDRSTRLEN]  = "";
                char remoteGidString[INET6_ADDRSTRLEN] = "";
                const char *localGidStr = NULL, *remoteGidStr = NULL;
                if(r->gidInfo) {
                    localGidStr  = inet_ntop(AF_INET6, &r->gidInfo->localGid, localGidString, sizeof(localGidString));
                    remoteGidStr = inet_ntop(AF_INET6, &r->gidInfo->remoteGid, remoteGidString, sizeof(remoteGidString));
                }
                WARN("NET/IB : Got completion from peer %s with error %d, opcode %d, len %d, vendor err %d (%s)%s%s%s%s",
                     net_socket::scclSocketToString(&addr, line),
                     wc->status,
                     wc->opcode,
                     wc->byte_len,
                     wc->vendor_err,
                     reqTypeStr[r->type],
                     localGidStr ? " localGid " : "",
                     localGidString,
                     remoteGidStr ? " remoteGid " : "",
                     remoteGidString);
                return scclRemoteError;
            }

            struct scclIbRequest* req = r->verbs->reqs + (wc->wr_id & 0xff);
            if(req->type == SCCL_NET_IB_REQ_SEND) {
                for(int i = 0; i < req->nreqs; i++) {
                    struct scclIbRequest* sendReq = r->verbs->reqs + ((wc->wr_id >> (i * 8)) & 0xff);
                    if((sendReq->events <= 0))
                        return scclInternalError;
                    sendReq->events--;
                }
            } else {
                if(req && wc->opcode == IBV_WC_RECV_RDMA_WITH_IMM) {
                    if(req->type != SCCL_NET_IB_REQ_RECV)
                        return scclInternalError;
                    if(req->nreqs > 1) {
                        // In the case of a multi recv, we only set sizes to 0 or 1.
                        for(int i = 0; i < req->nreqs; i++) {
                            req->recv.sizes[i] = (wc->imm_data >> i) & 0x1;
                        }
                    } else {
                        req->recv.sizes[0] += wc->imm_data;
                    }
                }
                req->events--;
            }
        }
    }
}

scclResult_t scclNetIb::closeSend(void* sendComm) {
    struct scclIbSendComm* comm = (struct scclIbSendComm*)sendComm;
    if(comm) {
        SCCLCHECK(net_socket::scclSocketClose(&comm->sock));
        for(int q = 0; q < comm->nqps; q++)
            if(comm->qps[q] != NULL)
                SCCLCHECK(wrap_ibv_destroy_qp(comm->qps[q]));
        if(comm->fifoMr != NULL)
            SCCLCHECK(wrap_ibv_dereg_mr(comm->fifoMr));
        SCCLCHECK(scclIbDestroyVerbs(&comm->verbs));
        free(comm);
    }
    return scclSuccess;
}

scclResult_t scclNetIb::closeRecv(void* recvComm) {
    struct scclIbRecvComm* comm = (struct scclIbRecvComm*)recvComm;
    if(comm) {
        if(!scclParamIbSockServerPortReuse() || reusedSockfd != comm->sock.fd)
            SCCLCHECK(net_socket::scclSocketClose(&comm->sock));
        for(int q = 0; q < comm->nqps; q++)
            if(comm->qps[q] != NULL)
                SCCLCHECK(wrap_ibv_destroy_qp(comm->qps[q]));
        if(comm->gpuFlush.enabled) {
            if(comm->gpuFlush.qp != NULL)
                SCCLCHECK(wrap_ibv_destroy_qp(comm->gpuFlush.qp));
            if(comm->gpuFlush.hostMr != NULL)
                SCCLCHECK(wrap_ibv_dereg_mr(comm->gpuFlush.hostMr));
        }
        if(comm->remFifo.mr != NULL)
            SCCLCHECK(wrap_ibv_dereg_mr(comm->remFifo.mr));
        SCCLCHECK(scclIbDestroyVerbs(&comm->verbs));
        free(comm);
    }
    return scclSuccess;
}

scclResult_t scclNetIb::closeListen(void* listenComm) {
    struct scclIbListenComm* comm = (struct scclIbListenComm*)listenComm;
    if(comm) {
        SCCLCHECK(net_socket::scclSocketClose(&comm->sock));
        free(comm);
    }
    return scclSuccess;
}

} // namespace net_ib
} // namespace net
} // namespace hardware
} // namespace sccl